openclaw-skills/skills/kb-bilibili-content/bilibili_content.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Bilibili Content Downloader - B站内容获取工具

功能:
- download_video(bvid, quality=480) - 下载低分辨率视频
- extract_subtitle(bvid) - 提取字幕
- extract_article(cvid) - 获取专栏全文
- 命令行参数支持
- JSON 输出
"""

import argparse
import asyncio
import json
import os
import re
import sys
from datetime import datetime
from pathlib import Path
from typing import Optional, Dict, Any, List

import aiohttp


class BilibiliContent:
    """B站内容获取器"""
    
    def __init__(self, output_dir: str = "./output"):
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
    
    def _get_headers(self) -> Dict[str, str]:
        """获取请求头"""
        return {
            'Referer': 'https://www.bilibili.com/',
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
            'Accept': 'application/json, text/plain, */*',
            'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
        }
    
    async def _make_request(self, session, url: str, params: Dict = None) -> Optional[Dict]:
        """发送API请求"""
        try:
            import aiohttp
            async with session.get(url, params=params, headers=self._get_headers(), timeout=aiohttp.ClientTimeout(total=30)) as response:
                if response.status == 200:
                    return await response.json()
                return None
        except Exception as e:
            print(f"请求错误: {e}")
            return None
    
    async def download_video(self, bvid: str, quality: int = 480) -> Dict[str, Any]:
        """
        下载B站视频
        
        Args:
            bvid: B站视频BV号 (如: BV1xx411c7mD)
            quality: 视频质量 (默认480)
                - 127: 8K
                - 126:杜比
                - 125:HDR
                - 120:4K
                - 116:1080P60
                - 112:1080P+
                - 80:1080P
                - 74:720P60
                - 64:720P
                - 32:480P
                - 16:360P
        
        Returns:
            Dict: 包含状态、消息和文件路径的结果
        """
        result = {
            "success": False,
            "bvid": bvid,
            "message": "",
            "filepath": None,
            "metadata": {}
        }
        
        async with aiohttp.ClientSession() as session:
            # 1. 获取视频信息
            print(f"正在获取视频信息: {bvid}")
            api_url = "https://api.bilibili.com/x/web-interface/view"
            data = await self._make_request(session, api_url, {'bvid': bvid})
            
            if not data or data.get('code') != 0:
                result["message"] = f"获取视频信息失败: {data.get('message', '未知错误')}"
                return result
            
            video_data = data.get('data', {})
            title = video_data.get('title', '未知标题')
            aid = video_data.get('aid')
            cid = video_data.get('cid')
            duration = video_data.get('duration', 0)
            author = video_data.get('owner', {}).get('name', '未知作者')
            
            result["metadata"] = {
                "title": title,
                "author": author,
                "aid": aid,
                "cid": cid,
                "duration": duration,
                "bvid": bvid
            }
            
            print(f"视频标题: {title}")
            print(f"作者: {author}")
            print(f"时长: {duration}秒")
            
            # 2. 获取视频流地址
            print(f"\n正在获取视频流...")
            
            # 选择合适的质量
            quality_map = {127: 127, 126: 126, 125: 125, 120: 120, 116: 116, 112: 112, 80: 80, 74: 74, 64: 64, 32: 32, 16: 16}
            target_quality = quality_map.get(quality, 32)  # 默认480P
            
            playurl_api = "https://api.bilibili.com/x/player/playurl"
            params = {
                'bvid': bvid,
                'cid': cid,
                'qn': target_quality,
                'otype': 'json',
                'fnver': 0,
                'fnval': 16,  # 支持DASH格式
            }
            
            playurl_data = await self._make_request(session, playurl_api, params)
            
            if not playurl_data or playurl_data.get('code') != 0:
                result["message"] = f"获取视频流失败: {playurl_data.get('message', '未知错误')}"
                return result
            
            play_data = playurl_data.get('data', {})
            
            # 获取视频和音频流
            dash_data = play_data.get('dash')
            if dash_data:
                video_list = dash_data.get('video', [])
                audio_list = dash_data.get('audio', [])
                
                if video_list:
                    # 选择质量最高的视频流
                    video_stream = video_list[0]
                    video_url = video_stream.get('base_url') or video_stream.get('baseUrl')
                    video_size = video_list[0].get('size', 0)
                    
                    # 获取音频流
                    audio_stream = audio_list[0] if audio_list else None
                    audio_url = audio_stream.get('base_url') or audio_stream.get('baseUrl') if audio_stream else None
                    
                    print(f"找到视频流: {video_url[:100]}...")
                    
                    # 3. 下载视频流
                    print(f"\n正在下载视频...")
                    
                    # 下载视频
                    async with session.get(video_url, headers=self._get_headers()) as resp:
                        if resp.status == 200:
                            video_content = await resp.read()
                            print(f"视频下载完成: {len(video_content) / 1024 / 1024:.2f}MB")
                        else:
                            result["message"] = f"视频下载失败: HTTP {resp.status}"
                            return result
                    
                    # 下载音频
                    audio_content = None
                    if audio_url:
                        print(f"正在下载音频...")
                        async with session.get(audio_url, headers=self._get_headers()) as resp:
                            if resp.status == 200:
                                audio_content = await resp.read()
                                print(f"音频下载完成: {len(audio_content) / 1024 / 1024:.2f}MB")
                    
                    # 4. 合并音视频
                    print(f"\n正在合并音视频...")
                    
                    safe_title = "".join(c for c in title if c.isalnum() or c in (' ', '-', '_')).strip()[:50]
                    safe_title = re.sub(r'[<>:"/\\|?*]', '', safe_title)
                    
                    output_path = self.output_dir / f"{safe_title}_{bvid}"
                    
                    # 保存文件
                    if video_content:
                        video_path = output_path.with_suffix('.mp4')
                        with open(video_path, 'wb') as f:
                            f.write(video_content)
                        
                        result["filepath"] = str(video_path)
                        result["success"] = True
                        result["message"] = f"视频已保存: {video_path}"
                        print(f"✅ 下载完成: {video_path}")
                else:
                    result["message"] = "未找到视频流"
            else:
                # 回退到FLV格式
                durl = play_data.get('durl', [])
                if durl:
                    video_url = durl[0].get('url')
                    print(f"使用FLV格式: {video_url[:100]}...")
                    
                    async with session.get(video_url, headers=self._get_headers()) as resp:
                        if resp.status == 200:
                            video_content = await resp.read()
                            
                            safe_title = "".join(c for c in title if c.isalnum() or c in (' ', '-', '_')).strip()[:50]
                            safe_title = re.sub(r'[<>:"/\\|?*]', '', safe_title)
                            
                            output_path = self.output_dir / f"{safe_title}_{bvid}.flv"
                            with open(output_path, 'wb') as f:
                                f.write(video_content)
                            
                            result["filepath"] = str(output_path)
                            result["success"] = True
                            result["message"] = f"视频已保存: {output_path}"
                            print(f"✅ 下载完成: {output_path}")
                        else:
                            result["message"] = f"视频下载失败: HTTP {resp.status}"
                else:
                    result["message"] = "未找到可用的视频流"
        
        return result
    
    async def extract_subtitle(self, bvid: str) -> Dict[str, Any]:
        """
        提取B站视频字幕
        
        Args:
            bvid: B站视频BV号
        
        Returns:
            Dict: 包含字幕内容的JSON结果
        """
        result = {
            "success": False,
            "bvid": bvid,
            "subtitles": [],
            "message": ""
        }
        
        async with aiohttp.ClientSession() as session:
            # 获取视频信息
            api_url = "https://api.bilibili.com/x/web-interface/view"
            data = await self._make_request(session, api_url, {'bvid': bvid})
            
            if not data or data.get('code') != 0:
                result["message"] = f"获取视频信息失败: {data.get('message', '未知错误')}"
                return result
            
            cid = data.get('data', {}).get('cid')
            
            # 获取字幕列表
            subtitle_api = "https://api.bilibili.com/x/player/pagelist"
            page_data = await self._make_request(session, subtitle_api, {'bvid': bvid})
            
            # 获取字幕详情
            subtitle_list = []
            if page_data and page_data.get('code') == 0:
                cid = page_data.get('data', [{}])[0].get('cid', cid)
                
                subtitle_url = f"https://api.bilibili.com/x/player/v2?bvid={bvid}&cid={cid}"
                subtitle_data = await self._make_request(session, subtitle_url)
                
                if subtitle_data and subtitle_data.get('code') == 0:
                    subtitles = subtitle_data.get('data', {}).get('subtitle', {}).get('list', [])
                    
                    for sub in subtitles:
                        sub_title = sub.get('lan_doc', '未知语言')
                        sub_url = sub.get('subtitle_url')
                        
                        if sub_url:
                            # 下载字幕文件
                            async with session.get(sub_url) as resp:
                                if resp.status == 200:
                                    sub_content = await resp.json()
                                    subtitle_list.append({
                                        "language": sub.get('lan', 'unknown'),
                                        "title": sub_title,
                                        "content": sub_content
                                    })
            
            result["subtitles"] = subtitle_list
            result["success"] = len(subtitle_list) > 0
            result["message"] = f"找到 {len(subtitle_list)} 个字幕" if subtitle_list else "未找到字幕"
            
            # 保存字幕到文件
            if subtitle_list:
                output_path = self.output_dir / f"{bvid}_subtitles.json"
                with open(output_path, 'w', encoding='utf-8') as f:
                    json.dump(result, f, ensure_ascii=False, indent=2)
                result["filepath"] = str(output_path)
                print(f"✅ 字幕已保存: {output_path}")
        
        return result
    
    async def extract_article(self, cvid: str) -> Dict[str, Any]:
        """
        获取B站专栏全文
        
        Args:
            cvid: B站专栏CV号
        
        Returns:
            Dict: 包含专栏内容的JSON结果
        """
        result = {
            "success": False,
            "cvid": cvid,
            "title": "",
            "content": "",
            "message": ""
        }
        
        async with aiohttp.ClientSession() as session:
            # 获取专栏信息
            api_url = "https://api.bilibili.com/x/article/viewinfo"
            data = await self._make_request(session, api_url, {'id': cvid})
            
            if not data or data.get('code') != 0:
                result["message"] = f"获取专栏信息失败: {data.get('message', '未知错误')}"
                return result
            
            article_data = data.get('data', {})
            title = article_data.get('title', '')
            content = article_data.get('content', '')
            
            # 获取作者信息
            author = article_data.get('author', {}).get('name', '未知作者')
            publish_time = article_data.get('publish_time', 0)
            view = article_data.get('stats', {}).get('view', 0)
            like = article_data.get('stats', {}).get('like', 0)
            
            result["title"] = title
            result["content"] = content
            result["author"] = author
            result["publish_time"] = datetime.fromtimestamp(publish_time).strftime('%Y-%m-%d %H:%M:%S') if publish_time else ''
            result["stats"] = {"view": view, "like": like}
            result["success"] = bool(title and content)
            result["message"] = f"成功获取专栏: {title}" if result["success"] else "获取专栏内容失败"
            
            # 保存到文件
            if result["success"]:
                safe_title = "".join(c for c in title if c.isalnum() or c in (' ', '-', '_')).strip()[:30]
                safe_title = re.sub(r'[<>:"/\\|?*]', '', safe_title)
                
                # 保存JSON
                json_path = self.output_dir / f"{safe_title}_{cvid}.json"
                with open(json_path, 'w', encoding='utf-8') as f:
                    json.dump(result, f, ensure_ascii=False, indent=2)
                
                # 保存Markdown
                md_content = f"""# {title}

> **来源**: [B站专栏](https://www.bilibili.com/read/cv{cvid})
> **作者**: {author}
> **发布时间**: {result["publish_time"]}
> **阅读**: {view} | **点赞**: {like}

---

{content}
"""
                md_path = self.output_dir / f"{safe_title}_{cvid}.md"
                with open(md_path, 'w', encoding='utf-8') as f:
                    f.write(md_content)
                
                result["json_path"] = str(json_path)
                result["md_path"] = str(md_path)
                print(f"✅ 专栏已保存: {md_path}")
        
        return result


async def main():
    """命令行入口"""
    parser = argparse.ArgumentParser(
        description='Bilibili Content Downloader - B站内容获取工具',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例:
  %(prog)s --id "BV1xx411c7mD" --type video          # 下载视频
  %(prog)s --id "BV1xx411c7mD" --type video --quality 720  # 下载720P视频
  %(prog)s --id "BV1xx411c7mD" --type subtitle        # 提取字幕
  %(prog)s --id "cv43780186" --type article           # 获取专栏
        """
    )
    
    parser.add_argument('--id', '-i', required=True, help='B站内容ID (BV号或CV号)')
    parser.add_argument('--type', '-t', required=True, choices=['video', 'subtitle', 'article'],
                        help='内容类型: video=视频, subtitle=字幕, article=专栏')
    parser.add_argument('--output', '-o', default='./output', help='输出目录 (默认: ./output)')
    parser.add_argument('--quality', '-q', type=int, default=480, 
                        help='视频质量 (默认: 480, 可选: 360, 480, 720, 1080)')
    parser.add_argument('--json', action='store_true', help='JSON格式输出')
    
    args = parser.parse_args()
    
    # 确定内容类型
    is_article = args.id.startswith('cv')
    
    if args.type == 'article' or is_article:
        content_type = 'article'
        content_id = args.id.replace('cv', '')
    elif args.type == 'subtitle':
        content_type = 'subtitle'
        content_id = args.id
    else:
        content_type = 'video'
        content_id = args.id
    
    # 创建获取器
    downloader = BilibiliContent(output_dir=args.output)
    
    # 执行操作
    if content_type == 'video':
        result = await downloader.download_video(content_id, quality=args.quality)
    elif content_type == 'subtitle':
        result = await downloader.extract_subtitle(content_id)
    else:
        result = await downloader.extract_article(content_id)
    
    # 输出结果
    if args.json:
        print(json.dumps(result, ensure_ascii=False, indent=2))
    else:
        if result["success"]:
            print(f"\n✅ {result['message']}")
        else:
            print(f"\n❌ {result['message']}")
        
        if result.get("filepath"):
            print(f"文件路径: {result['filepath']}")
    
    return result


if __name__ == '__main__':
    try:
        result = asyncio.run(main())
        sys.exit(0 if result.get("success") else 1)
    except KeyboardInterrupt:
        print("\n操作已取消")
        sys.exit(1)
    except Exception as e:
        print(f"\n错误: {e}")
        import traceback
        traceback.print_exc()
        sys.exit(1)