diff --git a/skills/kb-bilibili-metadata/bilibili_extractor.py b/skills/kb-bilibili-metadata/bilibili_extractor.py index 21c0e73..32f517e 100644 --- a/skills/kb-bilibili-metadata/bilibili_extractor.py +++ b/skills/kb-bilibili-metadata/bilibili_extractor.py @@ -431,6 +431,174 @@ class BilibiliExtractor: else: return f"{minutes:02d}:{secs:02d}" + async def _extract_series_async(self, mid: str, series_id: str) -> Dict[str, Any]: + """ + 异步提取系列合集信息（不包含列表内容） + + Args: + mid: UP主ID + series_id: 系列ID + + Returns: + Dict containing series basic info + """ + result = { + 'type': 'series', + 'series_id': series_id, + 'mid': mid, + 'subscribed': False, + 'basic_info': {} + } + + if not HTTPX_AVAILABLE: + return {'error': 'httpx 未安装'} + + try: + async with httpx.AsyncClient(timeout=30.0, headers=DEFAULT_HEADERS) as client: + # 获取系列信息 + response = await client.get('https://api.bilibili.com/x/space/series', + params={'mid': mid, 'series_id': series_id, 'pn': 1, 'ps': 1}) + if response.status_code == 200: + data = response.json() + if data.get('code') == 0: + series_info = data.get('data', {}).get('list', {}) + result['basic_info'] = { + 'title': series_info.get('title', ''), + 'description': series_info.get('desc', ''), + 'cover': series_info.get('cover', ''), + 'mid': mid, + 'creator_name': '', + 'total_count': series_info.get('cnt', 0), + 'last_updated': '', + 'update_count': 0 + } + + # 获取创作者信息 + up_info = await client.get('https://api.bilibili.com/x/space/acc/info', + params={'mid': mid}) + if up_info.status_code == 200: + up_data = up_info.json() + if up_data.get('code') == 0: + result['basic_info']['creator_name'] = up_data.get('data', {}).get('name', '') + + except Exception as e: + logger.error(f"系列合集提取错误: {e}") + result = {'error': str(e)} + + return result + + async def _extract_readlist_async(self, rl_id: str) -> Dict[str, Any]: + """ + 异步提取专栏合集信息（不包含列表内容） + + Args: + rl_id: 专栏合集ID (rl开头) + + Returns: + Dict containing readlist basic info + """ + result = { + 'type': 'readlist', + 'readlist_id': rl_id, + 'subscribed': False, + 'basic_info': {} + } + + if not HTTPX_AVAILABLE: + return {'error': 'httpx 未安装'} + + try: + async with httpx.AsyncClient(timeout=30.0, headers=DEFAULT_HEADERS) as client: + # 获取专栏合集信息 + response = await client.get('https://api.bilibili.com/x/article/readlist', + params={'id': rl_id.replace('rl', '')}) + if response.status_code == 200: + data = response.json() + if data.get('code') == 0: + list_info = data.get('data', {}) + result['basic_info'] = { + 'title': list_info.get('name', ''), + 'description': list_info.get('description', ''), + 'cover': list_info.get('cover', ''), + 'mid': list_info.get('mid', ''), + 'creator_name': '', + 'total_count': list_info.get('articles_count', 0), + 'last_updated': '', + 'is_public': list_info.get('public', True) + } + + # 获取创作者信息 + mid = list_info.get('mid', '') + if mid: + up_info = await client.get('https://api.bilibili.com/x/space/acc/info', + params={'mid': mid}) + if up_info.status_code == 200: + up_data = up_info.json() + if up_data.get('code') == 0: + result['basic_info']['creator_name'] = up_data.get('data', {}).get('name', '') + + except Exception as e: + logger.error(f"专栏合集提取错误: {e}") + result = {'error': str(e)} + + return result + + async def _extract_favlist_async(self, fid: str) -> Dict[str, Any]: + """ + 异步提取收藏夹信息（不包含列表内容） + + Args: + fid: 收藏夹ID + + Returns: + Dict containing favlist basic info + """ + result = { + 'type': 'favlist', + 'favlist_id': fid, + 'subscribed': False, + 'basic_info': {} + } + + if not HTTPX_AVAILABLE: + return {'error': 'httpx 未安装'} + + try: + async with httpx.AsyncClient(timeout=30.0, headers=DEFAULT_HEADERS) as client: + # 获取收藏夹信息 + response = await client.get('https://api.bilibili.com/x/space/favlist', + params={'fid': fid, 'pn': 1, 'ps': 1}) + if response.status_code == 200: + data = response.json() + if data.get('code') == 0: + fav_info = data.get('data', {}) + result['basic_info'] = { + 'title': fav_info.get('title', ''), + 'description': fav_info.get('intro', ''), + 'cover': fav_info.get('cover', ''), + 'mid': fav_info.get('mid', ''), + 'creator_name': '', + 'total_count': fav_info.get('media_count', 0), + 'is_public': fav_info.get('public', True), + 'like_count': fav_info.get('like', 0) + } + + # 获取创作者信息 + mid = fav_info.get('mid', '') + if mid: + up_info = await client.get('https://api.bilibili.com/x/space/acc/info', + params={'mid': mid}) + if up_info.status_code == 200: + up_data = up_info.json() + if up_data.get('code') == 0: + result['basic_info']['creator_name'] = up_data.get('data', {}).get('name', '') + + except Exception as e: + logger.error(f"收藏夹提取错误: {e}") + result = {'error': str(e)} + + return result + # 测试代码 if __name__ == '__main__': diff --git a/skills/kb-bilibili-metadata/bilibili_metadata.py b/skills/kb-bilibili-metadata/bilibili_metadata.py index c5acc1e..2fee418 100755 --- a/skills/kb-bilibili-metadata/bilibili_metadata.py +++ b/skills/kb-bilibili-metadata/bilibili_metadata.py @@ -7,6 +7,7 @@ B站元数据提取 CLI 工具 """ import argparse +import asyncio import json import sys from typing import Dict, Any @@ -58,6 +59,39 @@ def extract_up(extractor: BilibiliExtractor, mid: str, output_file: str = None) return result +def extract_series(extractor: BilibiliExtractor, mid: str, series_id: str, output_file: str = None) -> Dict[str, Any]: + """提取系列合集元数据""" + result = asyncio.run(extractor._extract_series_async(mid, series_id)) + + if output_file: + with open(output_file, 'w', encoding='utf-8') as f: + json.dump(result, f, ensure_ascii=False, indent=2) + + return result + + +def extract_readlist(extractor: BilibiliExtractor, rl_id: str, output_file: str = None) -> Dict[str, Any]: + """提取专栏合集元数据""" + result = asyncio.run(extractor._extract_readlist_async(rl_id)) + + if output_file: + with open(output_file, 'w', encoding='utf-8') as f: + json.dump(result, f, ensure_ascii=False, indent=2) + + return result + + +def extract_favlist(extractor: BilibiliExtractor, fid: str, output_file: str = None) -> Dict[str, Any]: + """提取收藏夹元数据""" + result = asyncio.run(extractor._extract_favlist_async(fid)) + + if output_file: + with open(output_file, 'w', encoding='utf-8') as f: + json.dump(result, f, ensure_ascii=False, indent=2) + + return result + + def main(): """主函数""" parser = argparse.ArgumentParser( @@ -74,6 +108,15 @@ def main(): # 提取UP主信息 python bilibili_metadata.py --id "7770724" --type up + # 提取系列合集信息 + python bilibili_metadata.py --id "123456" --type series --mid "7770724" + + # 提取专栏合集信息 + python bilibili_metadata.py --id "rl123456" --type readlist + + # 提取收藏夹信息 + python bilibili_metadata.py --id "123456" --type favlist + # 输出到文件 python bilibili_metadata.py --id "BV1xx411c7mD" --type video --output result.json """ @@ -87,9 +130,13 @@ def main(): parser.add_argument( '--type', required=True, - choices=['video', 'article', 'up'], + choices=['video', 'article', 'up', 'series', 'readlist', 'favlist'], help='资源类型' ) + parser.add_argument( + '--mid', + help='UP主ID (用于series类型)' + ) parser.add_argument( '--output', '-o', @@ -127,6 +174,15 @@ def main(): result = extract_article(extractor, args.id, args.output) elif args.type == 'up': result = extract_up(extractor, args.id, args.output) + elif args.type == 'series': + if not args.mid: + print("错误: series类型需要提供 --mid 参数") + sys.exit(1) + result = extract_series(extractor, args.mid, args.id, args.output) + elif args.type == 'readlist': + result = extract_readlist(extractor, args.id, args.output) + elif args.type == 'favlist': + result = extract_favlist(extractor, args.id, args.output) else: print(f"错误: 不支持的类型 {args.type}") sys.exit(1)