OpenClaw Agent
672a811c54
- kb_workflow.py: 修复目录查询命令(--pending, --metadata等) - bilibili_metadata.py: headless默认True，逻辑保持纯净 - deep_analyzer.py: 添加缺失的pathlib.Path导入 - SKILL.md: 更新步骤说明，明确由workflow提供路径给子技能 fix: 修复Playwright启动失败(NameError)和路径自动获取问题
229 lines
6.8 KiB
Python
Executable File
229 lines
6.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
B站元数据提取 CLI 工具
|
||
|
||
支持命令行调用，输出 JSON 格式数据
|
||
"""
|
||
|
||
import argparse
|
||
import asyncio
|
||
import json
|
||
import os
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Dict, Any
|
||
|
||
from bilibili_extractor import BilibiliExtractor
|
||
|
||
|
||
def format_number(num: int) -> str:
|
||
"""格式化数字"""
|
||
if num >= 100000000:
|
||
return f"{num / 100000000:.1f}亿"
|
||
elif num >= 10000:
|
||
return f"{num / 10000:.1f}万"
|
||
else:
|
||
return f"{num:,}"
|
||
|
||
|
||
def extract_video(extractor: BilibiliExtractor, bvid: str, output_file: str = None) -> Dict[str, Any]:
|
||
"""提取视频元数据"""
|
||
result = extractor.extract_video_metadata(bvid)
|
||
|
||
# 如果需要输出到文件
|
||
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_article(extractor: BilibiliExtractor, cvid: str, output_file: str = None) -> Dict[str, Any]:
|
||
"""提取专栏元数据"""
|
||
result = extractor.extract_article_metadata(cvid)
|
||
|
||
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_up(extractor: BilibiliExtractor, mid: str, output_file: str = None) -> Dict[str, Any]:
|
||
"""提取UP主信息（包含订阅跟踪所需信息）"""
|
||
result = asyncio.run(extractor._extract_up_async(mid))
|
||
|
||
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_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(
|
||
description='B站元数据提取工具',
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog="""
|
||
示例:
|
||
# 提取视频元数据
|
||
python bilibili_metadata.py --id "BV1xx411c7mD" --type video
|
||
|
||
# 提取专栏元数据
|
||
python bilibili_metadata.py --id "cv43780186" --type article
|
||
|
||
# 提取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
|
||
"""
|
||
)
|
||
|
||
parser.add_argument(
|
||
'--id',
|
||
required=True,
|
||
help='B站资源ID（BV号、CV号或UP主ID）'
|
||
)
|
||
parser.add_argument(
|
||
'--type',
|
||
required=True,
|
||
choices=['video', 'article', 'up', 'series', 'readlist', 'favlist'],
|
||
help='资源类型'
|
||
)
|
||
parser.add_argument(
|
||
'--mid',
|
||
help='UP主ID (用于series类型)'
|
||
)
|
||
parser.add_argument(
|
||
'--output',
|
||
'-o',
|
||
help='输出文件路径（JSON格式）'
|
||
)
|
||
parser.add_argument(
|
||
'--headless',
|
||
action='store_true',
|
||
default=True,
|
||
help='使用无头模式（浏览器，默认开启，使用API）'
|
||
)
|
||
parser.add_argument(
|
||
'--timeout',
|
||
type=int,
|
||
default=30,
|
||
help='超时时间（秒，默认30）'
|
||
)
|
||
parser.add_argument(
|
||
'--pretty',
|
||
action='store_true',
|
||
default=True,
|
||
help='格式化输出（默认开启）'
|
||
)
|
||
|
||
args = parser.parse_args()
|
||
|
||
# 创建提取器
|
||
extractor = BilibiliExtractor(headless=args.headless, timeout=args.timeout)
|
||
|
||
try:
|
||
# 根据类型提取数据
|
||
if args.type == 'video':
|
||
result = extract_video(extractor, args.id, args.output)
|
||
elif args.type == 'article':
|
||
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(json.dumps({
|
||
"success": False,
|
||
"error": "series类型需要提供 --mid 参数"
|
||
}, ensure_ascii=False, indent=2))
|
||
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(json.dumps({
|
||
"success": False,
|
||
"error": f"不支持的类型 {args.type}"
|
||
}, ensure_ascii=False, indent=2))
|
||
sys.exit(1)
|
||
|
||
# 生成文件名
|
||
platform = "bilibili"
|
||
filename = f"{platform}_{args.id}.meta.json"
|
||
filepath = args.output or filename
|
||
|
||
# 统一输出格式
|
||
output = {
|
||
"success": True,
|
||
"filename": filename,
|
||
"filepath": str(Path(filepath).resolve()) if filepath else None,
|
||
"message": f"{args.type}元数据提取成功"
|
||
}
|
||
|
||
print(json.dumps(output, ensure_ascii=False, indent=2))
|
||
|
||
# 如果指定了输出路径，则写入文件
|
||
if args.output:
|
||
with open(args.output, 'w', encoding='utf-8') as f:
|
||
json.dump(result, f, ensure_ascii=False, indent=2)
|
||
|
||
except Exception as e:
|
||
print(json.dumps({
|
||
"success": False,
|
||
"error": str(e),
|
||
"detail": type(e).__name__
|
||
}, ensure_ascii=False, indent=2))
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|