删除就技能
PR Test / lint (pull_request) Successful in 2s

This commit is contained in:
2026-03-16 16:55:21 +08:00
parent f98d649b5c
commit ae8f6080c9
25 changed files with 113 additions and 5415 deletions
@@ -0,0 +1,113 @@
---
status: resolved
trigger: "kb-workflow技能添加bilibili视频时,归档的文件名不对,analysis.md文件中的title, duration, author等元数据也不对。"
created: 2026-03-16T00:00:00Z
updated: 2026-03-16T00:10:00Z
resolved: 2026-03-16
---
## Current Focus
hypothesis: 修复已实施并自验证通过
test: 用户需要在实际环境中验证
expecting: 用户确认修复有效
next_action: 等待用户确认
## Symptoms
expected: |
1. analysis.md 的 frontmatter 应该包含正确的 title, duration, author(从 _metadata.json 获取)
2. .m4a 音频文件名应该遵循 `平台_BV号` 的规范(如 bilibili_BV16cNMzpExh.m4a)
actual: |
1. analysis.md 的 frontmatter 显示:
- title: "未命名" (应该是视频标题)
- duration: 0 (应该是 07:16 或秒数)
- author: "未知" (应该是 "阿周的Eureka")
2. .m4a 文件名是 "2-3月宏观经济运行状态复盘与推演 2026年政府工作报告解读 阿周的Eureka.m4a"
(应该是 bilibili_BV16cNMzpExh.m4a)
errors: 无显式错误,kb-workflow 命令执行成功,但输出不正确
reproduction: 对 Bilibili 视频 BV16cNMzpExh 运行 kb-workflow process
started: 首次测试时发现
additional_context: |
_metadata.json 文件中的元数据是正确的:
- title: "2-3月宏观经济运行状态复盘与推演 + 2026年政府工作报告解读 | 【阿周的Eureka】"
- author: "阿周的Eureka"
- duration: "07:16"
- bvid: "BV16cNMzpExh"
但 analysis.md 的 frontmatter 没有使用这些数据。
归档目录位于:kb-workflow/bilibili_BV16cNMzpExh/
## Eliminated
## Evidence
- timestamp: 2026-03-16
checked: kb-workflow/bilibili_BV16cNMzpExh/bilibili_BV16cNMzpExh_metadata.json
found: metadata 文件结构正确,title/author/duration 都在 metadata.metadata 下
implication: 元数据源文件是正确的
- timestamp: 2026-03-16
checked: kb-workflow/bilibili_BV16cNMzpExh/analysis.md
found: frontmatter 显示 title="未命名", author="未知", duration=0
implication: generate_frontmatter() 使用了默认值
- timestamp: 2026-03-16
checked: kb_pipeline/processors/analyze.py:111-116
found: _write_analysis_report(metadata=metadata, ...) 传递原始 metadata
implication: 原始 metadata 是嵌套结构 {metadata: {...}, stats: {...}}
- timestamp: 2026-03-16
checked: kb-deep-analyzer/deep_analyzer.py:436-470
found: generate_frontmatter() 使用 metadata.get('title', '未命名')
implication: 期望扁平结构,但收到嵌套结构,所以取不到值
- timestamp: 2026-03-16
checked: kb-deep-analyzer/deep_analyzer.py:382-414
found: _normalize_metadata() 可以正确扁平化嵌套结构
implication: DeepAnalyzer.analyze() 内部调用了 _normalize_metadata,但 _write_analysis_report() 没有调用
- timestamp: 2026-03-16
checked: kb_pipeline/platforms/bilibili.py:463-483
found: _sanitize_filename() 使用视频标题作为文件名
implication: 文件名不是规范的 bilibili_BV号 格式
- timestamp: 2026-03-16
checked: 单元测试验证
found: _normalize_metadata() 和 _sanitize_filename() 修复后测试通过
implication: 修复逻辑正确
## Resolution
root_cause: |
问题1 - analysis.md 元数据错误:
AnalyzeProcessor._write_analysis_report() 直接将原始嵌套 metadata 传递给 generate_frontmatter(),
而 generate_frontmatter() 期望扁平化的 metadata 结构。
原始结构: {metadata: {title, author, ...}, stats: {...}}
期望结构: {title, author, ...}
问题2 - 音频文件名不规范:
BilibiliAdapter._sanitize_filename() 使用视频标题作为文件名,
而不是规范的 "bilibili_BV号" 格式。
fix: |
问题1: 在 AnalyzeProcessor 中添加 _normalize_metadata() 方法,
在 _write_analysis_report() 中调用它将嵌套 metadata 扁平化
问题2: 修改 _sanitize_filename() 返回标准化的 "bilibili_{bvid}" 格式
verification: |
自验证测试通过:
- _normalize_metadata() 正确处理嵌套结构,返回扁平化的 metadata
- generate_frontmatter() 使用扁平化后的 metadata,生成正确的 frontmatter
- _sanitize_filename() 返回标准化的 "bilibili_BV号" 格式
files_changed:
- kb_pipeline/processors/analyze.py
- kb_pipeline/platforms/bilibili.py
@@ -1,24 +0,0 @@
{
"metadata": {
"title": "2-3月宏观经济运行状态复盘与推演 + 2026年政府工作报告解读 | 【阿周的Eureka】",
"author": "阿周的Eureka",
"bvid": "BV16cNMzpExh",
"aid": 116192639519967,
"mid": 46405906,
"pubtime": "2026-03-08 16:44:51",
"duration": "07:16",
"url": "https://www.bilibili.com/video/BV16cNMzpExh",
"tags": ""
},
"stats": {
"view": 78484,
"danmaku": 112,
"reply": 97,
"favorite": 4352,
"coin": 1762,
"share": 309,
"like": 5248
},
"subtitle": "",
"description": "各位玩家好,本期视频是对2-3月份的宏观经济运行状态,结合前几天发布的政府工作报告做的数据复盘与推演\n希望能够用一个更直观的方式和各位一起了解每个月宏观经济的运行情况\n这个【复盘经济系统运行状态】的系列,之后会作为每月的固定更新"
}
-34
View File
@@ -1,34 +0,0 @@
{
"schema_version": "1.0",
"content_id": "BV16cNMzpExh",
"platform": "bilibili",
"source_url": "https://www.bilibili.com/video/BV16cNMzpExh",
"created_at": "2026-03-15T23:52:15.244309",
"updated_at": "2026-03-15T23:52:18.638677",
"steps": {
"parse": {
"step_name": "parse",
"status": "completed",
"started_at": "2026-03-15T23:52:15.244320",
"completed_at": "2026-03-15T23:52:15.244530",
"error_message": null,
"output_summary": null
},
"download": {
"step_name": "download",
"status": "completed",
"started_at": "2026-03-15T23:52:15.244627",
"completed_at": "2026-03-15T23:52:18.525523",
"error_message": null,
"output_summary": null
},
"asr": {
"step_name": "asr",
"status": "failed",
"started_at": "2026-03-15T23:52:18.525793",
"completed_at": "2026-03-15T23:52:18.638670",
"error_message": "video_path 为空,无法执行 Whisper 语音识别。请确保视频文件已下载(对于没有官方字幕的视频,需要先下载视频文件)。",
"output_summary": null
}
}
}
@@ -1,34 +0,0 @@
{
"schema_version": "1.0",
"content_id": "BV16cNMzpExh",
"platform": "bilibili",
"source_url": "https://www.bilibili.com/video/BV16cNMzpExh",
"created_at": "2026-03-15T23:52:15.244309",
"updated_at": "2026-03-15T23:52:18.525804",
"steps": {
"parse": {
"step_name": "parse",
"status": "completed",
"started_at": "2026-03-15T23:52:15.244320",
"completed_at": "2026-03-15T23:52:15.244530",
"error_message": null,
"output_summary": null
},
"download": {
"step_name": "download",
"status": "completed",
"started_at": "2026-03-15T23:52:15.244627",
"completed_at": "2026-03-15T23:52:18.525523",
"error_message": null,
"output_summary": null
},
"asr": {
"step_name": "asr",
"status": "running",
"started_at": "2026-03-15T23:52:18.525793",
"completed_at": null,
"error_message": null,
"output_summary": null
}
}
}
-46
View File
@@ -1,46 +0,0 @@
# KB Bilibili Metadata Skill
**kb-bilibili-metadata** - Extract Bilibili video/article/up主 metadata.
## Features
- Video metadata extraction
- Article (专栏) metadata extraction
- UP主 information extraction
- JSON output format
- Playwright + HTTP API fallback
## Commands
```bash
# Video metadata
python bilibili_metadata.py --id "BV1xx411c7mD" --type video
# Article metadata
python bilibili_metadata.py --id "cv43780186" --type article
# UP主 info
python bilibili_metadata.py --id "7770724" --type up
```
## Output
```json
{
"success": true,
"data": {
"platform": "bilibili",
"type": "video",
"id": "BV1xx411c7mD",
"title": "视频标题",
"author": "UP主名字",
"stats": {"view": 100000, "like": 5000}
}
}
```
## Installation
```bash
cp -r kb-bilibili-metadata/ ~/skills/
```
-8
View File
@@ -1,8 +0,0 @@
"""
bilibili-metadata 技能包
B站元数据提取工具
"""
__version__ = "1.0.0"
__author__ = "OpenClaw"
-666
View File
@@ -1,666 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
B站元数据提取器
独立版本,不依赖 obsidian-kb 结构
支持视频、专栏、UP主信息提取
"""
import asyncio
import json
import logging
import re
from datetime import datetime
from typing import Dict, Any, Optional
from urllib.parse import urlparse
logger = logging.getLogger(__name__)
# 默认请求头
DEFAULT_HEADERS = {
'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',
'Referer': 'https://www.bilibili.com/',
'Accept': 'application/json, text/plain, */*',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
}
try:
from playwright.async_api import async_playwright
PLAYWRIGHT_AVAILABLE = True
except ImportError:
PLAYWRIGHT_AVAILABLE = False
try:
from undetected_playwright import stealth_async
STEALTH_AVAILABLE = True
except ImportError:
STEALTH_AVAILABLE = False
try:
import httpx
HTTPX_AVAILABLE = True
except ImportError:
HTTPX_AVAILABLE = False
class BilibiliExtractor:
"""
B站元数据提取器
支持:
- 视频信息提取(标题、作者、统计数据、字幕等)
- 专栏文章提取
- UP主信息提取
"""
# 正则模式
BV_PATTERN = re.compile(r'BV[a-zA-Z0-9]{10}')
CV_PATTERN = re.compile(r'cv(\d+)')
URL_PATTERN = re.compile(r'https?://www\.bilibili\.com/video/(BV[a-zA-Z0-9]{10})')
ARTICLE_URL_PATTERN = re.compile(r'https?://www\.bilibili\.com/read/(cv\d+)')
UP_URL_PATTERN = re.compile(r'https?://space\.bilibili\.com/(\d+)')
def __init__(self, headless: bool = True, timeout: int = 30):
"""
初始化提取器
Args:
headless: 是否使用无头模式(浏览器)
timeout: 超时时间(秒)
"""
self.headless = headless
self.timeout = timeout * 1000 # 转换为毫秒
def extract_video_metadata(self, bvid: str) -> Dict[str, Any]:
"""
提取视频元数据
Args:
bvid: B站视频BV号(如 BV1xx411c7mD)
Returns:
Dict containing metadata, stats, description, subtitle
"""
# 验证并解析BV号
bvid = self._resolve_bvid(bvid)
if not bvid:
raise ValueError(f"无效的BV号: {bvid}")
# 使用异步方法提取
return asyncio.run(self._extract_video_async(bvid))
def extract_article_metadata(self, cvid: str) -> Dict[str, Any]:
"""
提取专栏元数据
Args:
cvid: B站专栏CV号(如 cv43780186)
Returns:
Dict containing article metadata
"""
# 解析CV号
cvid = self._resolve_cvid(cvid)
if not cvid:
raise ValueError(f"无效的CV号: {cvid}")
return asyncio.run(self._extract_article_async(cvid))
def extract_up_metadata(self, mid: str) -> Dict[str, Any]:
"""
提取UP主信息
Args:
mid: B站用户ID(如 7770724)
Returns:
Dict containing UP主信息
"""
# 解析MID
mid = self._resolve_mid(mid)
if not mid:
raise ValueError(f"无效的MID: {mid}")
return asyncio.run(self._extract_up_async(mid))
def _resolve_bvid(self, source_id: str) -> Optional[str]:
"""解析并返回BV号"""
source = source_id.strip()
# 直接BV号
if self.BV_PATTERN.fullmatch(source):
return source
# 完整URL
match = self.URL_PATTERN.match(source)
if match:
return match.group(1)
return None
def _resolve_cvid(self, source_id: str) -> Optional[str]:
"""解析并返回CV号"""
source = source_id.strip()
# 直接CV号(带cv前缀)
if source.startswith('cv'):
match = self.CV_PATTERN.match(source)
if match:
return source
# 纯数字
if source.isdigit():
return f"cv{source}"
# 完整URL
match = self.ARTICLE_URL_PATTERN.match(source)
if match:
return match.group(1)
return None
def _resolve_mid(self, source_id: str) -> Optional[str]:
"""解析并返回MID"""
source = source_id.strip()
# 纯数字
if source.isdigit():
return source
# 完整URL
match = self.UP_URL_PATTERN.match(source)
if match:
return match.group(1)
return None
async def _extract_video_async(self, bvid: str) -> Dict[str, Any]:
"""异步提取视频信息"""
url = f"https://www.bilibili.com/video/{bvid}"
result = {
'metadata': {},
'stats': {},
'subtitle': '',
'description': ''
}
if not PLAYWRIGHT_AVAILABLE:
# 使用 HTTP API
return await self._extract_video_by_api(bvid)
async with async_playwright() as playwright:
try:
browser = await playwright.chromium.launch(
headless=self.headless,
args=[
'--no-sandbox',
'--disable-setuid-sandbox',
'--disable-blink-features=AutomationControlled',
'--disable-dev-shm-usage',
]
)
context = await browser.new_context(
viewport={'width': 1920, 'height': 1080},
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',
locale='zh-CN',
)
if STEALTH_AVAILABLE:
await stealth_async(context)
page = await context.new_page()
await page.goto(url, wait_until='domcontentloaded', timeout=self.timeout)
await asyncio.sleep(2)
content = await page.content()
video_data = await self._try_extract_video_methods(page, content, bvid)
if video_data:
result['metadata'] = {
'title': video_data.get('title', ''),
'author': video_data.get('author', ''),
'bvid': video_data.get('bvid', ''),
'aid': video_data.get('aid', ''),
'mid': video_data.get('mid', ''),
'pubtime': video_data.get('pubtime', ''),
'duration': video_data.get('duration', ''),
'url': url,
'tags': video_data.get('tags', ''),
}
result['stats'] = {
'view': video_data.get('view', 0),
'danmaku': video_data.get('danmaku', 0),
'reply': video_data.get('reply', 0),
'favorite': video_data.get('favorite', 0),
'coin': video_data.get('coin', 0),
'share': video_data.get('share', 0),
'like': video_data.get('like', 0),
}
result['description'] = video_data.get('desc', '')
await browser.close()
except Exception as e:
logger.error(f"视频提取错误: {e}")
# 回退到 API
return await self._extract_video_by_api(bvid)
return result
async def _extract_video_by_api(self, bvid: str) -> Dict[str, Any]:
"""使用B站API提取视频信息"""
result = {
'metadata': {},
'stats': {},
'subtitle': '',
'description': ''
}
try:
async with httpx.AsyncClient(timeout=30.0, headers=DEFAULT_HEADERS) as client:
# 获取视频信息
response = await client.get('https://api.bilibili.com/x/web-interface/view', params={'bvid': bvid})
if response.status_code == 200:
data = response.json()
if data.get('code') == 0:
video_data = data.get('data', {})
result['metadata'] = {
'title': video_data.get('title', ''),
'author': video_data.get('owner', {}).get('name', ''),
'bvid': video_data.get('bvid', ''),
'aid': video_data.get('aid', ''),
'mid': video_data.get('owner', {}).get('mid', ''),
'pubtime': datetime.fromtimestamp(video_data.get('pubdate', 0)).strftime('%Y-%m-%d %H:%M:%S') if video_data.get('pubdate') else '',
'duration': self._format_duration(video_data.get('duration', 0)),
'url': f"https://www.bilibili.com/video/{bvid}",
'tags': ', '.join([t.get('tag_name', '') for t in video_data.get('tag', [])]),
}
stat = video_data.get('stat', {})
result['stats'] = {
'view': stat.get('view', 0),
'danmaku': stat.get('danmaku', 0),
'reply': stat.get('reply', 0),
'favorite': stat.get('favorite', 0),
'coin': stat.get('coin', 0),
'share': stat.get('share', 0),
'like': stat.get('like', 0),
}
result['description'] = video_data.get('desc', '')
except Exception as e:
logger.error(f"API提取错误: {e}")
return result
async def _try_extract_video_methods(self, page, content: str, bvid: str) -> Dict[str, Any]:
"""尝试多种方法提取视频信息"""
video_data = {}
# 方法1: 从 __INITIAL_STATE__
try:
initial_state = await page.evaluate('() => typeof window.__INITIAL_STATE__ !== "undefined" ? window.__INITIAL_STATE__ : null')
if initial_state:
video_data = self._parse_video_state(initial_state)
if video_data and video_data.get('title'):
return video_data
except:
pass
# 方法2: 使用 API(更可靠)
return await self._extract_video_by_api(bvid)
def _parse_video_state(self, data: Dict) -> Dict[str, Any]:
"""解析 __INITIAL_STATE__ 数据"""
video_data = {}
try:
video = data.get('videoData', {})
video_data['title'] = video.get('title', '')
video_data['bvid'] = video.get('bvid', '')
video_data['aid'] = video.get('aid', '')
owner = video.get('owner', {})
video_data['author'] = owner.get('name', '')
video_data['mid'] = owner.get('mid', '')
stat = video.get('stat', {})
video_data['view'] = stat.get('view', 0)
video_data['danmaku'] = stat.get('danmaku', 0)
video_data['reply'] = stat.get('reply', 0)
video_data['favorite'] = stat.get('favorite', 0)
video_data['coin'] = stat.get('coin', 0)
video_data['share'] = stat.get('share', 0)
video_data['like'] = stat.get('like', 0)
pubdate = video.get('pubdate', 0)
video_data['pubtime'] = datetime.fromtimestamp(pubdate).strftime('%Y-%m-%d %H:%M:%S') if pubdate else ''
video_data['duration'] = self._format_duration(video.get('duration', 0))
video_data['desc'] = video.get('desc', '')
video_data['tags'] = ', '.join([t.get('tag_name', '') for t in video.get('tag', [])])
except:
pass
return video_data
async def _extract_article_async(self, cvid: str) -> Dict[str, Any]:
"""异步提取专栏信息"""
result = {}
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/viewinfo', params={'id': cvid[2:]})
if response.status_code == 200:
data = response.json()
if data.get('code') == 0:
article = data.get('data', {})
result = {
'cvid': cvid,
'title': article.get('title', ''),
'author': article.get('author_name', ''),
'mid': article.get('author_mid', ''),
'publish_time': datetime.fromtimestamp(article.get('publish_time', 0)).strftime('%Y-%m-%d %H:%M:%S') if article.get('publish_time') else '',
'view_count': article.get('stats', {}).get('view', 0),
'like_count': article.get('stats', {}).get('like', 0),
'coin_count': article.get('stats', {}).get('coin', 0),
'share_count': article.get('stats', {}).get('share', 0),
'reply_count': article.get('stats', {}).get('reply', 0),
'url': f"https://www.bilibili.com/read/{cvid}",
'summary': article.get('summary', ''),
'word_count': article.get('words', 0),
}
except Exception as e:
logger.error(f"专栏提取错误: {e}")
result = {'error': str(e)}
return result
async def _extract_up_async(self, mid: str) -> Dict[str, Any]:
"""
异步提取UP主信息(支持订阅跟踪)
Args:
mid: UP主ID
Returns:
Dict containing UP主完整信息
"""
result = {
'type': 'up',
'mid': mid,
'subscribed': False,
'basic_info': {},
'subscription_info': {}
}
if not HTTPX_AVAILABLE:
return {'error': 'httpx 未安装'}
try:
async with httpx.AsyncClient(timeout=30.0, headers=DEFAULT_HEADERS) as client:
# 获取UP主基本信息
response = await client.get('https://api.bilibili.com/x/space/acc/info', params={'mid': mid})
if response.status_code == 200:
data = response.json()
if data.get('code') == 0:
up = data.get('data', {})
result['basic_info'] = {
'name': up.get('name', ''),
'sex': up.get('sex', ''),
'face': up.get('face', ''),
'sign': up.get('sign', ''),
'level': up.get('level', 0),
'vip_type': up.get('vip', {}).get('type', 0),
'vip_status': up.get('vip', {}).get('status', 0),
'following_count': up.get('following', 0),
'follower_count': up.get('follower', 0),
'url': f"https://space.bilibili.com/{mid}",
}
# 获取UP主视频统计
stats_response = await client.get('https://api.bilibili.com/x/space/navnum', params={'mid': mid})
if stats_response.status_code == 200:
stats_data = stats_response.json()
if stats_data.get('code') == 0:
result['basic_info']['video_count'] = stats_data.get('data', {}).get('video', 0)
# 获取UP主专栏数量
article_response = await client.get('https://api.bilibili.com/x/space/article', params={'mid': mid, 'pn': 1, 'ps': 1})
if article_response.status_code == 200:
article_data = article_response.json()
if article_data.get('code') == 0:
result['basic_info']['article_count'] = article_data.get('data', {}).get('count', 0)
# 获取UP主最近视频(用于跟踪更新)
video_response = await client.get('https://api.bilibili.com/x/space/arc/search',
params={'mid': mid, 'pn': 1, 'ps': 1, 'order': 'pubtime'})
if video_response.status_code == 200:
video_data = video_response.json()
if video_data.get('code') == 0:
videos = video_data.get('data', {}).get('list', {}).get('vlist', [])
if videos:
latest_video = videos[0]
result['subscription_info'] = {
'last_video_id': latest_video.get('bvid', ''),
'last_video_title': latest_video.get('title', ''),
'last_video_time': datetime.fromtimestamp(latest_video.get('created', 0)).strftime('%Y-%m-%d %H:%M:%S') if latest_video.get('created') else '',
}
# 获取UP主最近专栏(用于跟踪更新)
article_response = await client.get('https://api.bilibili.com/x/space/article',
params={'mid': mid, 'pn': 1, 'ps': 1, 'sort': 'publish_time'})
if article_response.status_code == 200:
article_data = article_response.json()
if article_data.get('code') == 0:
articles = article_data.get('data', {}).get('articles', [])
if articles:
latest_article = articles[0]
if 'last_article' not in result['subscription_info']:
result['subscription_info']['last_article'] = {}
result['subscription_info']['last_article'] = {
'id': latest_article.get('id', 0),
'title': latest_article.get('title', ''),
'time': datetime.fromtimestamp(latest_article.get('publish_time', 0)).strftime('%Y-%m-%d %H:%M:%S') if latest_article.get('publish_time') else '',
}
except Exception as e:
logger.error(f"UP主提取错误: {e}")
result = {'error': str(e)}
return result
def _format_duration(self, seconds: int) -> str:
"""格式化时长(秒 -> HH:MM:SS)"""
if not seconds:
return ''
hours = seconds // 3600
minutes = (seconds % 3600) // 60
secs = seconds % 60
if hours > 0:
return f"{hours:02d}:{minutes:02d}:{secs:02d}"
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__':
extractor = BilibiliExtractor()
print("=" * 60)
print("B站元数据提取器测试")
print("=" * 60)
# 测试视频
print("\n测试视频提取: BV1MnFNz5EDL")
result = extractor.extract_video_metadata('BV1MnFNz5EDL')
print(json.dumps(result, ensure_ascii=False, indent=2))
-228
View File
@@ -1,228 +0,0 @@
#!/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()
-102
View File
@@ -1,102 +0,0 @@
# KB Deep Analyzer Skill
**kb-deep-analyzer** - AI 深度内容分析工具。
## Features
- 多类型分析:视频(video)、文章(article)
- 模板化输出
- 质量评分
- 自动标签
- **多格式元数据支持**: 支持扁平结构 `{"title":"xxx"}` 和嵌套结构 `{"data":{"metadata":{...}}}`
## 内容类型
| 类型 | 参数 | 输出格式 |
|------|------|----------|
| 视频 | `--type video` | 摘要 + 主题 + 见解 + 时间轴 |
| 文章 | `--type article` | 300字摘要 + 主题 + 原文 |
## Commands
```bash
# 分析视频(默认)
AI_API_KEY="your-api-key" AI_PROVIDER="zhipu" AI_MODEL="glm-5" \
python3 -u deep_analyzer.py --metadata meta.json --content content.json --output analysis.md --type video
# 分析文章(300字摘要 + 原文)
AI_API_KEY="your-api-key" AI_PROVIDER="zhipu" AI_MODEL="glm-5" \
python3 -u deep_analyzer.py --metadata meta.json --content content.json --output analysis.md --type article
# 指定 AI 提供商
python3 -u deep_analyzer.py -m meta.json -c content.json -o out.md --type video --provider zhipu --model glm-5
```
## 环境变量配置
**必须设置以下环境变量**:
```bash
export AI_API_KEY="your-api-key"
export AI_PROVIDER="zhipu" # 或 openai, minimax
export AI_MODEL="glm-5" # 模型名称
export AI_BASE_URL="https://..." # 可选,自定义 API 地址
```
### 重要提示
**必须使用 `-u` 参数运行 Python**,确保输出实时显示在终端:
```bash
python3 -u deep_analyzer.py ...
```
如果不使用 `-u` 参数,Python 的 stdout 缓冲会导致输出延迟或看不到进度。
## 执行方式选择
### 方式1: 并发执行(推荐,默认)
- 使用 `batch_size=3`(默认)
- 6个分片约 100 秒完成
- 适合快速分析
```bash
python3 -u deep_analyzer.py --metadata meta.json --content content.json -o out.md
```
### 方式2: 顺序执行
- 使用 `batch_size=1`
- 6个分片约 735 秒完成
- 适合调试和问题排查
```bash
# 在代码中设置 batch_size=1
# analyzer.analyze(metadata, content, "video", batch_size=1)
```
## Metadata Format
支持两种元数据格式:
```json
// 格式1: 扁平结构
{
"title": "视频标题",
"author": "作者名",
"duration": 619,
"bvid": "BVxxx"
}
// 格式2: 嵌套结构(kb-bilibili-metadata 输出)
{
"data": {
"metadata": {
"title": "视频标题",
"author": "作者名",
"duration": "10:19"
}
}
}
```
会自动转换为扁平格式进行分析。
-514
View File
@@ -1,514 +0,0 @@
"""Obsidian KB - AI客户端
支持多个 AI 提供商和 fallback 机制
"""
import os
import json
import requests
from typing import Dict, Any, List, Optional, Tuple
from src.models import ContentType
from src.config import get_config
from src.content_chunker import ContentChunker, get_chunking_strategy, chunk_content
class AIClient:
"""
AI客户端 - 支持分段分析长视频内容和 fallback 机制
"""
def __init__(self):
self.config = get_config()
self.chunker = ContentChunker()
# 获取 AI 配置
self._load_ai_config()
def _load_ai_config(self) -> None:
"""加载 AI 配置(主 provider + fallbacks)"""
self.ai_configs = []
# 加载主配置
main_provider = self.config.get('ai.provider', 'openai')
main_config = {
'provider': main_provider,
'api_key': self.config.get('ai.api_key'),
'model': self.config.get('ai.model', 'gpt-4'),
'base_url': self.config.get('ai.base_url'),
'is_primary': True
}
self.ai_configs.append(main_config)
# 加载 fallback 配置
fallbacks = self.config.get('ai.fallbacks', [])
for fb in fallbacks:
self.ai_configs.append({
'provider': fb.get('provider'),
'api_key': fb.get('api_key'),
'model': fb.get('model'),
'base_url': fb.get('base_url'),
'is_primary': False
})
def _call_api_with_fallback(self, prompt: str, max_tokens: int = 1000, response_format: str = 'text') -> Tuple[bool, str]:
"""
使用 fallback 机制调用 API
Args:
prompt: 提示词
max_tokens: 最大 token 数
response_format: 响应格式 ('text' 或 'json')
Returns:
(success, response) 元组
"""
last_error = None
for i, ai_config in enumerate(self.ai_configs):
try:
if not ai_config.get('api_key'):
print(f"[AIClient] 跳过 {ai_config.get('provider')} - API Key 为空")
continue
provider_type = ai_config.get('provider')
print(f"[AIClient] 尝试使用 {provider_type} (优先级 {i+1}/{len(self.ai_configs)})...")
response = self._call_single_provider(ai_config, prompt, max_tokens, response_format)
return True, response
except Exception as e:
last_error = e
print(f"[AIClient] {provider_type} 调用失败: {e}")
continue
# 所有 provider 都失败了
error_msg = f"所有 AI 提供商均失败,最后错误: {last_error}" if last_error else "没有可用的 AI 配置"
print(f"[AIClient] {error_msg}")
return False, self._mock_generate(prompt, max_tokens) # 返回 mock 响应作为兜底
def _call_single_provider(self, ai_config: Dict, prompt: str, max_tokens: int, response_format: str) -> str:
"""
调用单个 AI 提供商
Args:
ai_config: AI 配置
prompt: 提示词
max_tokens: 最大 token 数
response_format: 响应格式
Returns:
API 响应文本
"""
provider = ai_config.get('provider')
api_key = ai_config.get('api_key')
model = ai_config.get('model')
base_url = ai_config.get('base_url')
if provider == 'minimax':
return self._call_minimax(api_key, model, base_url, prompt, max_tokens, response_format)
elif provider == 'zhipu':
return self._call_zhipu(api_key, model, base_url, prompt, max_tokens, response_format)
elif provider == 'alibaba':
return self._call_alibaba(api_key, model, base_url, prompt, max_tokens, response_format)
elif provider == 'openai':
return self._call_openai(api_key, model, base_url, prompt, max_tokens, response_format)
else:
raise ValueError(f"不支持的 provider: {provider}")
def _call_minimax(self, api_key: str, model: str, base_url: str, prompt: str, max_tokens: int, response_format: str) -> str:
"""调用 MiniMax API (Anthropic Messages 兼容接口)"""
url = base_url or "https://api.minimaxi.com/anthropic/v1/messages"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"max_tokens": max_tokens,
"messages": [
{"role": "user", "content": prompt}
]
}
if response_format == 'json':
payload['messages'][0]['content'] = prompt + "\n\n请直接返回 JSON 格式,不要有其他文字。"
response = requests.post(url, headers=headers, json=payload, timeout=30)
response.raise_for_status()
data = response.json()
# MiniMax 返回的是 Anthropic 格式,content 是一个数组
# 需要找到 type='text' 的元素
for item in data.get('content', []):
if item.get('type') == 'text':
return item.get('text', '')
# 如果没有找到 text 类型,返回第一个元素的值(兼容旧格式)
return data.get('content', [{}])[0].get('text', '')
def _call_zhipu(self, api_key: str, model: str, base_url: str, prompt: str, max_tokens: int, response_format: str) -> str:
"""调用智谱 AI API (OpenAI 兼容接口)"""
# 确保 base_url 包含 /chat/completions
if base_url and not base_url.endswith('/chat/completions'):
url = f"{base_url}/chat/completions"
else:
url = base_url or "https://open.bigmodel.cn/api/paas/v4/chat/completions"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"max_tokens": max_tokens,
"messages": [
{"role": "user", "content": prompt}
]
}
response = requests.post(url, headers=headers, json=payload, timeout=30)
response.raise_for_status()
data = response.json()
return data['choices'][0]['message']['content']
def _call_alibaba(self, api_key: str, model: str, base_url: str, prompt: str, max_tokens: int, response_format: str) -> str:
"""调用阿里云通义千问 API (OpenAI 兼容接口)"""
url = f"{base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"max_tokens": max_tokens,
"messages": [
{"role": "user", "content": prompt}
]
}
response = requests.post(url, headers=headers, json=payload, timeout=30)
response.raise_for_status()
data = response.json()
return data['choices'][0]['message']['content']
def _call_openai(self, api_key: str, model: str, base_url: str, prompt: str, max_tokens: int, response_format: str) -> str:
"""调用 OpenAI API"""
url = base_url or "https://api.openai.com/v1/chat/completions"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"max_tokens": max_tokens,
"messages": [
{"role": "user", "content": prompt}
]
}
response = requests.post(url, headers=headers, json=payload, timeout=30)
response.raise_for_status()
data = response.json()
return data['choices'][0]['message']['content']
def quick_generate(self, prompt: str, max_tokens: int = 100) -> str:
"""快速生成(用于轻量扫描)"""
success, response = self._call_api_with_fallback(prompt, max_tokens, 'text')
return response
def generate_json(self, prompt: str) -> Dict[str, Any]:
"""生成JSON格式响应"""
success, response = self._call_api_with_fallback(prompt, 1000, 'json')
try:
return json.loads(response)
except:
return self._parse_json_from_text(response)
def _deep_analyze(self, content_text: str, content_type: str, duration: Optional[int] = None) -> Dict[str, Any]:
"""
C5.3: 深度分析入口,根据内容时长决定是否分段分析
Args:
content_text: 内容文本
content_type: 内容类型
duration: 视频时长(秒)
Returns:
分析结果字典
"""
duration = duration or 0
print(f"[AIClient] 深度分析: type={content_type}, duration={duration}s")
# 长视频(>=30分钟)且是视频类型,使用分段分析
if duration >= 1800 and content_type == "video":
print(f"[AIClient] 使用分段分析策略")
return self._analyze_in_chunks(content_text, duration)
# 短视频直接分析
print(f"[AIClient] 使用直接分析策略")
return self._analyze_direct(content_text)
def _analyze_in_chunks(self, content_text: str, duration: int) -> Dict[str, Any]:
"""
C5.3: 分段分析长视频
Args:
content_text: SRT字幕文本
duration: 视频时长(秒)
Returns:
合并后的分析结果
"""
# 1. 获取分段策略
strategy = get_chunking_strategy(duration)
print(f"[AIClient] 分段策略: {strategy['description']}")
# 2. 执行分片
chunks = chunk_content(content_text, duration)
print(f"[AIClient] 内容已分为 {len(chunks)} 段")
# 3. 逐段分析
chunk_summaries = []
for i, chunk in enumerate(chunks):
print(f"[AIClient] 分析第 {i+1}/{len(chunks)} 段...")
summary = self._analyze_chunk(chunk.text, chunk.index)
# 处理时间格式(可能是float秒数或字符串)
start_time = chunk.start_time
end_time = chunk.end_time
if isinstance(start_time, (int, float)):
start_time = self._format_time(start_time)
if isinstance(end_time, (int, float)):
end_time = self._format_time(end_time)
chunk_summaries.append({
'index': chunk.index,
'start_time': start_time,
'end_time': end_time,
'summary': summary
})
# 4. 合并所有段摘要
print(f"[AIClient] 合并 {len(chunk_summaries)} 段分析结果")
final_result = self._merge_summaries(chunk_summaries, "video")
# 5. 构建完整时间轴
final_result['timeline'] = self._build_timeline_from_chunks(chunk_summaries)
return final_result
def _analyze_chunk(self, chunk_text: str, chunk_index: int) -> str:
"""
C5.3: 分析单个内容分片
Args:
chunk_text: 分片文本
chunk_index: 分片索引
Returns:
分片摘要(限制500字以内)
"""
# 截断文本避免过长
text_sample = chunk_text[:3000] if len(chunk_text) > 3000 else chunk_text
prompt = f"""请对以下视频片段进行简要分析(限制在300字以内):
片段内容:
{text_sample}
请提供:
1. 这段的核心主题(50字以内)
2. 关键要点(3-5条)
3. 重要细节或引用
请用中文回答,控制在300字以内。"""
try:
summary = self.quick_generate(prompt, max_tokens=500)
# 确保不超过500字
if len(summary) > 500:
summary = summary[:497] + "..."
return summary
except Exception as e:
print(f"[AIClient] 分片 {chunk_index} 分析失败: {e}")
return f"第{chunk_index+1}段分析失败"
def _merge_summaries(self, chunk_summaries: List[Dict], content_type: str) -> Dict[str, Any]:
"""
C5.3: 合并所有段摘要,生成完整分析报告
Args:
chunk_summaries: 各段摘要列表
content_type: 内容类型
Returns:
完整分析结果
"""
# 构建合并提示
summaries_text = "\n\n".join([
f"=== 第{s['index']+1}段 ({s['start_time']}-{s['end_time']}) ===\n{s['summary']}"
for s in chunk_summaries
])
prompt = f"""基于以下各段分析,生成完整的视频分析报告:
{summaries_text}
请提供以下格式的JSON:
{{
"summary": "整体摘要(200字以内)",
"topics": ["主题1", "主题2", "主题3", "主题4"],
"insights": ["深入见解1", "深入见解2", "深入见解3"],
"actions": ["建议行动1", "建议行动2"],
"full_analysis": "详细分析(500字以内)"
}}"""
try:
result = self.generate_json(prompt)
# 确保必要字段存在
if 'topics' not in result:
result['topics'] = ["主题A", "主题B", "主题C"]
if 'insights' not in result:
result['insights'] = ["重要见解1", "重要见解2"]
if 'actions' not in result:
result['actions'] = []
if 'summary' not in result:
result['summary'] = "长视频分析完成"
if 'full_analysis' not in result:
result['full_analysis'] = summaries_text[:500]
return result
except Exception as e:
print(f"[AIClient] 合并摘要失败: {e}")
# 返回基础合并结果
return {
'summary': f'视频共分为{len(chunk_summaries)}个部分进行分析',
'topics': [f"主题{i+1}" for i in range(min(4, len(chunk_summaries)))],
'insights': ['从分段分析中综合得出'],
'actions': [],
'full_analysis': summaries_text[:800]
}
def _analyze_direct(self, content_text: str) -> Dict[str, Any]:
"""直接分析(不分段)"""
# 截断文本
text_sample = content_text[:5000] if len(content_text) > 5000 else content_text
prompt = f"""请分析以下内容,提供JSON格式结果:
{text_sample}
返回格式:
{{
"summary": "整体摘要(150字以内)",
"topics": ["主题1", "主题2", "主题3"],
"insights": ["见解1", "见解2"],
"actions": ["行动建议1"],
"timeline": [
{{"start_time": "00:00", "end_time": "05:00", "title": "第一部分", "summary": "..."}}
]
}}"""
try:
return self.generate_json(prompt)
except Exception as e:
print(f"[AIClient] 直接分析失败: {e}")
return {
'summary': '分析失败',
'topics': ['无法解析'],
'insights': ['请检查内容格式'],
'actions': [],
'timeline': []
}
def _format_time(self, seconds: float) -> str:
"""格式化秒数为时间字符串"""
hours = int(seconds // 3600)
minutes = int((seconds % 3600) // 60)
secs = int(seconds % 60)
if hours > 0:
return f"{hours:02d}:{minutes:02d}:{secs:02d}"
return f"{minutes:02d}:{secs:02d}"
def _build_timeline_from_chunks(self, chunk_summaries: List[Dict]) -> List[Dict[str, Any]]:
"""从分片摘要构建时间轴"""
timeline = []
for summary in chunk_summaries:
timeline.append({
'index': summary['index'],
'start_time': summary['start_time'],
'end_time': summary['end_time'],
'title': f"第{summary['index']+1}部分",
'summary': summary['summary'][:200] + "..." if len(summary['summary']) > 200 else summary['summary']
})
return timeline
def _mock_generate(self, prompt: str, max_tokens: int) -> str:
"""模拟AI生成(实际实现应调用真实API)"""
# 根据prompt内容返回模拟响应
prompt_lower = prompt.lower()
if 'json' in prompt_lower or 'json格式' in prompt:
if '主题' in prompt or 'topic' in prompt_lower:
return json.dumps({
'summary': '这是一个关于人工智能和技术发展的内容,涵盖了多个重要主题。',
'topics': ['人工智能', '机器学习', '未来发展', '技术趋势'],
'insights': [
'AI正在改变我们的工作方式',
'机器学习应用越来越广泛',
'未来技术发展趋势值得关注'
],
'actions': ['深入研究相关技术', '关注行业动态'],
'full_analysis': '详细分析内容...'
}, ensure_ascii=False)
return json.dumps({'result': '模拟JSON响应'})
if '摘要' in prompt or 'summary' in prompt_lower:
return '这是一段关于人工智能发展的重要讨论,涵盖了技术趋势和未来应用。'
if '关键词' in prompt or 'keyword' in prompt_lower:
return '人工智能, 机器学习, 深度学习, 神经网络, 自然语言处理'
return '模拟AI响应内容'
def _parse_json_from_text(self, text: str) -> Dict[str, Any]:
"""从文本中解析JSON"""
# 尝试提取JSON块
import re
json_match = re.search(r'\{[\s\S]*\}', text)
if json_match:
try:
return json.loads(json_match.group())
except:
pass
return {'raw_response': text}
class MultiAIClient:
"""
多AI客户端 - 支持多种AI提供商的客户端
"""
def __init__(self):
self.config = get_config()
self.primary_client = AIClient()
self.clients = {
'openai': self.primary_client,
'local': AIClient(), # 可以配置为本地模型
}
def _deep_analyze(self, content_text: str, content_type: str, duration: Optional[int] = None) -> Dict[str, Any]:
"""使用主客户端进行深度分析"""
return self.primary_client._deep_analyze(content_text, content_type, duration)
def quick_generate(self, prompt: str, max_tokens: int = 100, provider: str = None) -> str:
"""快速生成"""
client = self.clients.get(provider, self.primary_client)
return client.quick_generate(prompt, max_tokens)
-590
View File
@@ -1,590 +0,0 @@
#!/usr/bin/env python3
"""Deep Analyzer - AI 深度内容分析工具
一次性调用 AI 生成完整深度分析报告。
Usage:
python deep_analyzer.py --metadata meta.json --content content.json --output analysis.md [--type video]
python deep_analyzer.py --help
"""
import argparse
import json
import os
import sys
import re
from datetime import datetime
from pathlib import Path
from typing import Dict, Any, List
from dataclasses import dataclass, field
# ============== 数据模型 ==============
@dataclass
class AnalysisResult:
"""分析结果"""
summary: str = ""
main_topics: List[str] = field(default_factory=list)
key_insights: List[str] = field(default_factory=list)
recommended_tags: List[str] = field(default_factory=list)
full_analysis: str = ""
original_text: str = "" # 文章类型的原文内容
# ============== AI 客户端 ==============
class SimpleAIClient:
"""AI 客户端"""
def __init__(self, api_key: str = None, provider: str = None, model: str = None, base_url: str = None):
self.api_key = api_key or os.environ.get("AI_API_KEY", "")
self.provider = provider or os.environ.get("AI_PROVIDER", "")
self.model = model or os.environ.get("AI_MODEL", "glm-5")
self.base_url = base_url or os.environ.get("AI_BASE_URL", "")
self.max_retries = 3
def generate(self, prompt: str, max_tokens: int = 16000) -> str:
"""生成文本"""
api_key = os.environ.get("AI_API_KEY", self.api_key)
provider = os.environ.get("AI_PROVIDER", self.provider)
model = os.environ.get("AI_MODEL", self.model)
base_url = os.environ.get("AI_BASE_URL", self.base_url)
if not api_key:
raise ValueError("AI_API_KEY 未设置")
for attempt in range(self.max_retries):
try:
import time
start_time = time.time()
print(f"[AI Client] 开始调用 AI...")
if provider == "openai":
result = self._call_openai(prompt, max_tokens, api_key, model, base_url)
elif provider == "zhipu":
result = self._call_zhipu(prompt, max_tokens, api_key, model, base_url)
elif provider == "minimax":
result = self._call_minimax(prompt, max_tokens, api_key, model, base_url)
else:
raise ValueError(f"不支持的 AI 提供商: {provider}")
elapsed = time.time() - start_time
print(f"[AI Client] AI 调用完成,耗时: {elapsed:.2f}秒")
return result
except Exception as e:
if attempt < self.max_retries - 1:
print(f"[AI Client] 第 {attempt + 1} 次调用失败,{self.max_retries - attempt - 1} 次重试剩余: {e}")
continue
else:
raise
def _call_openai(self, prompt: str, max_tokens: int, api_key: str, model: str, base_url: str) -> str:
import requests
url = f"{base_url}/chat/completions" if base_url else "https://api.openai.com/v1/chat/completions"
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
payload = {"model": model, "max_tokens": max_tokens, "messages": [{"role": "user", "content": prompt}]}
response = requests.post(url, headers=headers, json=payload, timeout=600)
response.raise_for_status()
return response.json()['choices'][0]['message']['content']
def _call_zhipu(self, prompt: str, max_tokens: int, api_key: str, model: str, base_url: str) -> str:
import requests
base_url = base_url or "https://open.bigmodel.cn/api/coding/paas/v4"
url = f"{base_url}/chat/completions"
headers = {"Authorization": f"Bearer {api_key}"}
payload = {"model": model, "messages": [{"role": "user", "content": prompt}], "max_tokens": max_tokens}
response = requests.post(url, headers=headers, json=payload, timeout=600)
response.raise_for_status()
data = response.json()
message = data.get('choices', [{}])[0].get('message', {})
content = message.get('content', '')
reasoning = message.get('reasoning_content', '')
if not content and reasoning:
sentences = reasoning.strip().split('。')
useful = [s.strip() for s in sentences if s.strip() and len(s.strip()) > 10]
if useful:
return '。'.join(useful[-3:]) + '。'
return reasoning.strip()[:500]
return content
def _call_minimax(self, prompt: str, max_tokens: int, api_key: str, model: str, base_url: str) -> str:
import requests
url = base_url or "https://api.minimaxi.com/anthropic/v1/messages"
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
payload = {"model": model, "max_tokens": max_tokens, "messages": [{"role": "user", "content": prompt}]}
response = requests.post(url, headers=headers, json=payload, timeout=600)
response.raise_for_status()
data = response.json()
for item in data.get('content', []):
if item.get('type') == 'text':
return item.get('text', '')
return str(data)
# ============== 分析器 ==============
class DeepAnalyzer:
"""深度内容分析器"""
def __init__(self, ai_client: SimpleAIClient = None):
self.ai_client = ai_client or SimpleAIClient()
def analyze(self, metadata: Dict[str, Any], content: Dict[str, Any], content_type: str = "video") -> AnalysisResult:
"""一次性调用 AI 生成完整分析报告
Args:
metadata: 元数据字典
content: 内容字典
content_type: 内容类型,"video" 或 "article"
"""
# 标准化元数据
metadata = self._normalize_metadata(metadata)
title = metadata.get('title', '未知')
duration = metadata.get('duration', 0)
author = metadata.get('author', '')
pubtime = metadata.get('pubtime', '')
url = metadata.get('url', '')
bvid = metadata.get('bvid', '')
platform = metadata.get('platform', 'bilibili')
# 获取内容 - 支持多种字段名
text = content.get('text', '') or content.get('subtitle', '') or content.get('subtitles', '') or content.get('description', '')
print(f"[DeepAnalyzer] 开始分析: {title}")
print(f"[DeepAnalyzer] 类型: {content_type}, 内容长度: {len(text)} 字符")
# 构建 prompt(根据类型选择不同模板)
prompt = self._build_prompt(
title=title,
duration=duration,
duration_str=self._format_duration(duration),
author=author,
pubtime=pubtime,
url=url,
bvid=bvid,
platform=platform,
text=text,
content_type=content_type
)
# 调用 AI
try:
full_analysis = self.ai_client.generate(prompt, max_tokens=16000)
except Exception as e:
print(f"[DeepAnalyzer] AI 调用失败: {e}")
raise
# 解析结果
result = self._parse_analysis(full_analysis, text, duration, content_type)
# 文章/回答类型:追加原文
if content_type in ("article", "answer", "zhuanlan"):
result.original_text = text
return result
def _build_prompt(self, title: str, duration: int, duration_str: str, author: str,
pubtime: str, url: str, bvid: str, platform: str, text: str,
content_type: str = "video") -> str:
"""构建分析 prompt,根据内容类型选择不同模板"""
if content_type == "article":
return self._build_article_prompt(title, author, pubtime, url, platform, text)
else:
return self._build_video_prompt(title, duration, duration_str, author, pubtime, url, bvid, platform, text)
def _build_video_prompt(self, title: str, duration: int, duration_str: str, author: str,
pubtime: str, url: str, bvid: str, platform: str, text: str) -> str:
"""构建视频分析 prompt"""
# 计算分片数(按10分钟分片)
num_chunks = max(1, (duration + 599) // 600)
# 截取适当长度的文本(20-30K 字符)
max_chars = 30000
if len(text) > max_chars:
text_sample = text[:max_chars] + "\n\n[...]"
else:
text_sample = text
prompt = f"""你是一个专业的视频内容分析师。请根据以下信息生成深度分析报告:
【视频信息】
标题:{title}
作者:{author}
时长:{duration_str}({duration}秒)
发布日期:{pubtime}
平台:{platform}
URL:{url}
【内容】(完整字幕)
{text_sample}
【输出要求】必须使用以下格式:
## 标题
[自己总结的简洁标题,禁止使用原标题]
## 摘要
[200-400字]
## 主要主题
[从内容中提取不超过10个主题,每个不超过10字]
- 主题1
- 主题2
...
## 关键见解
[若干条深刻见解]
- 见解1
- 见解2
...
## 标签
[从内容中提取3-5个精炼标签,每个不超过10字]
- 标签1
- 标签2
...
## 时间轴
### 第 1 部分 (00:00 - 10:00)
[详细分析,至少500字]
### 第 2 部分 (10:00 - 20:00)
[详细分析,至少500字]
...(以此类推)
【重要】分析应足够详尽,读者阅读后可了解视频核心内容,无需再看原视频。"""
return prompt
def _build_article_prompt(self, title: str, author: str, pubtime: str,
url: str, platform: str, text: str) -> str:
"""构建文章分析 prompt"""
# 文章保留完整内容(不截断,因为需要追加原文)
# 但 prompt 中只取前 20000 字符供 AI 分析
max_chars = 20000
if len(text) > max_chars:
text_sample = text[:max_chars] + "\n\n[...内容过长已截断,但原文会完整保留...]"
else:
text_sample = text
prompt = f"""你是一个专业的文章内容分析师。请根据以下信息生成摘要:
【文章信息】
标题:{title}
作者:{author}
发布日期:{pubtime}
平台:{platform}
URL:{url}
【内容】
{text_sample}
【输出要求】必须使用以下格式:
## 摘要
[300字以内的摘要,描述文章的核心观点和主要内容]
## 主要主题
[从内容中提取不超过5个主题,每个不超过10字]
- 主题1
- 主题2
...
## 标签
[从内容中提取3-5个精炼标签,每个不超过10字]
- 标签1
- 标签2
...
【重要】
1. 摘要应精炼概括文章核心观点,让读者快速了解文章主旨
2. 不要输出时间轴
3. 不要输出关键见解(除非是特别重要的观点)"""
return prompt
def _parse_analysis(self, analysis_text: str, original_text: str, duration: int,
content_type: str = "video") -> AnalysisResult:
"""解析 AI 返回的分析结果"""
result = AnalysisResult()
result.full_analysis = analysis_text
# 提取各部分(使用正则)
lines = analysis_text.split('\n')
# 简单解析:按 ## 标题分割
current_section = ""
in_section = None
for line in lines:
if line.startswith('##'):
# 保存上一个 section
if in_section:
self._store_section(result, in_section, current_section)
# 开始新 section
section_name = line.lstrip('#').strip()
in_section = section_name
current_section = ""
elif in_section:
current_section += line + "\n"
# 存储最后一个 section
if in_section:
self._store_section(result, in_section, current_section)
# 如果解析失败,使用原始文本
if not result.summary:
result.summary = analysis_text[:500]
return result
def _store_section(self, result: AnalysisResult, section_name: str, content: str):
"""存储解析到的 section"""
content = content.strip()
if not content:
return
section_lower = section_name.lower()
if '摘要' in section_lower:
result.summary = content[:500]
elif '主题' in section_lower:
# 提取主题
topics = []
for line in content.split('\n'):
line = line.strip().lstrip('-').strip()
if line and len(line) <= 15:
topics.append(line)
result.main_topics = topics[:10]
elif '见解' in section_lower or '观点' in section_lower:
# 提取见解
insights = []
for line in content.split('\n'):
line = line.strip().lstrip('-').strip()
if line and len(line) > 5:
insights.append(line)
result.key_insights = insights
elif '标签' in section_lower:
# 提取标签
tags = []
for line in content.split('\n'):
line = line.strip().lstrip('-').strip()
if line and len(line) <= 10:
tags.append(line)
result.recommended_tags = tags[:5]
def _normalize_metadata(self, metadata: Dict) -> Dict:
"""标准化元数据"""
if not metadata:
return {}
# 如果已经扁平
if 'title' in metadata:
duration = metadata.get('duration', 0)
if isinstance(duration, str) and ':' in duration:
parts = duration.split(':')
try:
metadata['duration'] = int(parts[0]) * 60 + int(parts[1])
except:
pass
return metadata
# 嵌套结构
if 'metadata' in metadata:
meta = metadata['metadata']
normalized = {}
for k, v in meta.items():
normalized[k] = v
# 处理 duration
duration = normalized.get('duration', 0)
if isinstance(duration, str) and ':' in duration:
parts = duration.split(':')
try:
normalized['duration'] = int(parts[0]) * 60 + int(parts[1])
except:
pass
return normalized
return metadata
def _format_duration(self, seconds: int) -> str:
"""格式化时长"""
h = seconds // 3600
m = (seconds % 3600) // 60
s = seconds % 60
if h > 0:
return f"{h}:{m:02d}:{s:02d}"
return f"{m}:{s:02d}"
# ============== 工具函数 ==============
def load_json(file_path: str) -> Dict:
"""加载 JSON 文件"""
if not file_path or not os.path.exists(file_path):
return {}
with open(file_path, 'r', encoding='utf-8') as f:
return json.load(f)
def generate_frontmatter(metadata: Dict, result: AnalysisResult, content_type: str = "video") -> str:
"""生成 Markdown frontmatter"""
# 提取标签
tags = result.recommended_tags or []
# 格式化日期
pubtime = metadata.get('pubtime', '')
if pubtime:
date_str = str(pubtime)[:10]
else:
date_str = datetime.now().strftime('%Y-%m-%d')
fm = ["---",
f"title: \"{metadata.get('title', '未命名')}\"",
f"date: {date_str}",
f"author: {metadata.get('author', '未知')}",
f"platform: {metadata.get('platform', 'bilibili')}"]
# 视频类型添加 duration 和 bvid
if content_type == "video":
fm.insert(3, f"duration: {metadata.get('duration_str', metadata.get('duration', 0))}")
bvid = metadata.get('bvid', '')
if bvid:
fm.insert(5, f"bvid: {bvid}")
# 添加 URL
url = metadata.get('url', '')
if url:
fm.append(f"url: {url}")
if tags:
fm.append("tags: [" + ", ".join(f'"{t}"' for t in tags) + "]")
fm.append("---")
return '\n'.join(fm)
# ============== 主程序 ==============
def main():
parser = argparse.ArgumentParser(description="Deep Analyzer - AI 深度内容分析工具")
parser.add_argument("--metadata", "-m", required=True, help="元数据 JSON 文件路径")
parser.add_argument("--content", "-c", required=True, help="内容 JSON 文件路径")
parser.add_argument("--output", "-o", required=True, help="输出 Markdown 文件路径")
parser.add_argument("--type", "-t", default="video", choices=["video", "article"], help="内容类型")
parser.add_argument("--api-key", "-k", default=None, help="AI API Key")
parser.add_argument("--provider", "-p", default=None, choices=["openai", "zhipu", "minimax"], help="AI 提供商")
parser.add_argument("--model", default=None, help="AI 模型")
parser.add_argument("--base-url", "-u", default=None, help="API Base URL")
parser.add_argument("--no-frontmatter", action="store_true", help="不生成 frontmatter")
args = parser.parse_args()
# 加载数据
print(f"[Main] 加载元数据: {args.metadata}")
raw_metadata = load_json(args.metadata)
print(f"[Main] 加载内容: {args.content}")
content = load_json(args.content)
# 标准化元数据
metadata = {}
if 'title' in raw_metadata:
metadata = raw_metadata.copy()
# 处理 duration
duration = metadata.get('duration', 0)
if isinstance(duration, str) and ':' in duration:
parts = duration.split(':')
try:
metadata['duration'] = int(parts[0]) * 60 + int(parts[1])
except:
pass
metadata['duration_str'] = format_duration(metadata.get('duration', 0))
elif 'metadata' in raw_metadata:
meta = raw_metadata['metadata']
metadata = meta.copy()
duration = metadata.get('duration', 0)
if isinstance(duration, str) and ':' in duration:
parts = duration.split(':')
try:
metadata['duration'] = int(parts[0]) * 60 + int(parts[1])
except:
pass
metadata['duration_str'] = format_duration(metadata.get('duration', 0))
# 初始化
ai_client = SimpleAIClient(
api_key=args.api_key or os.environ.get("AI_API_KEY", ""),
provider=args.provider or os.environ.get("AI_PROVIDER", ""),
model=args.model or os.environ.get("AI_MODEL", "glm-5"),
base_url=args.base_url or os.environ.get("AI_BASE_URL", "")
)
analyzer = DeepAnalyzer(ai_client)
# 执行分析
print(f"[Main] 开始分析...")
result = analyzer.analyze(metadata, content, args.type)
# 使用 AI 生成的原始内容(保留 Markdown 格式)
full_analysis = result.full_analysis
# 删除内容中的"标签" section(只保留 frontmatter 中的 tags)
# 匹配 ## 标签 或 标签(无##)到下一个 ## 或文件结尾之间的内容
full_analysis = re.sub(r'(## ?)?标签\n[\s\S]*?(?=\n## ?|\n时间轴|---|\Z)', '', full_analysis)
# 清理多余的空行
full_analysis = re.sub(r'\n{3,}', '\n\n', full_analysis)
# 文章类型:追加原文
if args.type == "article" and result.original_text:
full_analysis = full_analysis.rstrip() + "\n\n## 原文\n\n" + result.original_text
# 写入输出
os.makedirs(os.path.dirname(os.path.abspath(args.output)), exist_ok=True)
with open(args.output, 'w', encoding='utf-8') as f:
# 添加 frontmatter
if not args.no_frontmatter:
fm = generate_frontmatter(metadata, result, args.type)
f.write(fm + "\n\n")
# 写入分析内容
f.write(full_analysis)
# 输出结果
output = {
"success": True,
"filename": Path(args.output).name,
"filepath": str(Path(args.output).resolve()),
"message": "深度分析完成",
"stats": {
"summary_length": len(result.summary) if result.summary else 0,
"topics_count": len(result.main_topics),
"insights_count": len(result.key_insights),
"tags_count": len(result.recommended_tags)
}
}
print(json.dumps(output, ensure_ascii=False, indent=2))
def format_duration(seconds: int) -> str:
"""格式化时长"""
h = seconds // 3600
m = (seconds % 3600) // 60
s = seconds % 60
if h > 0:
return f"{h}:{m:02d}:{s:02d}"
return f"{m}:{s:02d}"
if __name__ == "__main__":
try:
main()
except Exception as e:
print(json.dumps({"success": False, "error": str(e)}, ensure_ascii=False, indent=2))
sys.exit(1)
File diff suppressed because it is too large Load Diff
-247
View File
@@ -1,247 +0,0 @@
# KB Workflow 环境安装脚本
一键检查并安装 kb-workflow 知识库工作流所需的全部环境依赖。
## 功能特性
- ✅ 自动检测系统类型 (Ubuntu/Debian/Fedora/Arch)
- ✅ 检查系统依赖 (Python3, pip3, ffmpeg, tesseract, git)
- ✅ 检查 Python 依赖包 (httpx, aiohttp, playwright, easyocr 等)
- ✅ 自动创建知识库目录结构
- ✅ 生成环境变量配置示例
- ✅ 汇总显示所有需要执行的安装命令
## 使用方法
### 运行检查脚本
```bash
cd /home/starmen1265/.openclaw/workspace/skills
./install-kb-workflow.sh
```
脚本会检查所有依赖,并显示需要执行的安装命令。
### 示例输出
```
========================================
第一步: 检查系统依赖
========================================
[SUCCESS] Python3 已安装: Python 3.12.3
[WARN] pip3 未安装
[WARN] ffmpeg 未安装
...
========================================
需要安装的系统依赖
========================================
检测到以下依赖未安装:
- python3-pip
- ffmpeg
- tesseract-ocr
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
请运行以下命令安装:
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
sudo apt-get update
sudo apt-get install -y python3-pip ffmpeg tesseract-ocr tesseract-ocr-chi-sim tesseract-ocr-eng
```
## 依赖清单
### 系统依赖
| 依赖 | 用途 | 安装命令 |
|------|------|----------|
| **python3** | Python 运行时 | `sudo apt-get install python3` |
| **python3-pip** | Python 包管理器 | `sudo apt-get install python3-pip` |
| **ffmpeg** | 视频处理(OCR 截帧) | `sudo apt-get install ffmpeg` |
| **tesseract-ocr** | OCR 引擎(备用方案) | `sudo apt-get install tesseract-ocr` |
| **git** | 版本控制(归档提交) | `sudo apt-get install git` |
### Tesseract 语言包
```bash
# Ubuntu/Debian
sudo apt-get install tesseract-ocr-chi-sim tesseract-ocr-chi-tra tesseract-ocr-eng
# Fedora/RHEL
sudo dnf install tesseract-langpack-chi_sim tesseract-langpack-chi_tra tesseract-langpack-eng
```
### Python 依赖
```bash
# 基础 HTTP 库
pip3 install --user httpx aiohttp requests
# 图像和 OCR
pip3 install --user Pillow easyocr pytesseract
# 浏览器自动化
pip3 install --user playwright
python3 -m playwright install chromium
```
## 目录结构
安装完成后会创建以下目录结构:
```
~/work/knowledge-base/ # 知识库根目录(可配置)
├── archive/ # 归档目录
│ ├── pending/ # 待处理
│ └── subscriptions/ # 订阅信息
├── metadata/ # 元数据文件
├── content/ # 内容文件
│ ├── downloads/ # 视频下载
│ └── subtitles/ # 字幕文件
└── analyzed/ # 深度分析报告
```
## 环境变量
### 必需
- `KB_BASE_PATH` - 知识库根目录(默认: `~/work/knowledge-base`)
### 可选
- `BILIBILI_COOKIE` - B站 Cookie,用于下载会员视频
- `AI_API_KEY` - AI API 密钥,用于深度内容分析
- `AI_PROVIDER` - AI 提供商(openai/zhipu/minimax)
- `AI_MODEL` - AI 模型名称(如 gpt-4)
- `AI_BASE_URL` - 自定义 API 端点
- `KB_GIT_USER` - Git 用户名(用于归档提交)
- `KB_GIT_EMAIL` - Git 邮箱(用于归档提交)
- `KB_GIT_REMOTE` - Git 远程仓库地址
### 配置方法
将以下内容添加到 `~/.bashrc` 或 `~/.zshrc`:
```bash
# KB Workflow 配置
export KB_BASE_PATH="~/work/knowledge-base"
# 可选:B站 Cookie
export BILIBILI_COOKIE="SESSDATA=xxx; bili_jct=xxx"
# 可选:AI 配置
export AI_API_KEY="your-api-key"
export AI_PROVIDER="openai"
export AI_MODEL="gpt-4"
```
然后重新加载配置:
```bash
source ~/.bashrc # 或 source ~/.zshrc
```
## 故障排除
### 1. pip3 命令不存在
```bash
# 方法 1: 安装 pip3
sudo apt-get install python3-pip
# 方法 2: 使用 python3 -m pip
python3 -m pip install --user <package>
```
### 2. Playwright 浏览器安装失败
```bash
# 手动安装 Chromium
python3 -m playwright install chromium
# 如果仍然失败,尝试安装依赖
sudo apt-get install libnss3 libatk-bridge2.0-0 libxss1 libgtk-3-0
```
### 3. EasyOCR 安装失败
EasyOCR 依赖 PyTorch,可能需要较长时间安装:
```bash
# 先安装 PyTorch
pip3 install --user torch torchvision --index-url https://download.pytorch.org/whl/cpu
# 再安装 easyocr
pip3 install --user easyocr
```
### 4. 权限问题
如果使用 `--user` 安装的包无法找到,检查 PATH:
```bash
# 查看用户安装路径
python3 -m site --user-base
# 添加到 PATH(添加到 ~/.bashrc)
export PATH="$PATH:$(python3 -m site --user-base)/bin"
```
### 5. OCR 中文识别问题
确保已安装中文语言包:
```bash
# 检查已安装的语言包
tesseract --list-langs
# 应该能看到 chi_sim 和 chi_tra
```
## 更新
如果技能更新了依赖,可以重新运行检查脚本:
```bash
./install-kb-workflow.sh
```
## 子技能清单
| 技能 | 路径 | 功能 |
|------|------|------|
| kb-workflow | `kb-workflow/kb_workflow.py` | 主流程协调 |
| kb-link-parser | `kb-link-parser/link-parser.py` | 链接解析 |
| kb-bilibili-metadata | `kb-bilibili-metadata/bilibili_metadata.py` | B站元数据提取 |
| kb-bilibili-content | `kb-bilibili-content/bilibili_content.py` | B站内容下载 |
| kb-ocr-processor | `kb-ocr-processor/ocr_processor.py` | 视频 OCR 识别 |
| kb-deep-analyzer | `kb-deep-analyzer/deep_analyzer.py` | AI 深度分析 |
## 使用 kb-workflow
安装完成后,可以通过以下方式使用:
### 方式 1:通过主代理
直接向主代理发送消息:
```
添加到知识库 https://www.bilibili.com/video/BVxxxxx
```
### 方式 2:手动运行子技能
```bash
# 1. 链接解析
python3 kb-link-parser/link-parser.py --url "https://www.bilibili.com/video/BVxxxxx"
# 2. 元数据提取
python3 kb-bilibili-metadata/bilibili_metadata.py --id "BVxxxxx" --type video
# 3. 内容下载
python3 kb-bilibili-content/bilibili_content.py --id "BVxxxxx" --type subtitle-download
# 4. 深度分析
python3 kb-deep-analyzer/deep_analyzer.py --metadata meta.json --content content.json --output analysis.md
```
## 许可证
MIT License
-101
View File
@@ -1,101 +0,0 @@
# SkillBug.md
## 问题记录
### 2026-02-10: 子代理权限问题
**问题:** `sessions_spawn` 调用子代理时返回 `forbidden` 错误。
**原因:** OpenClaw 配置中 allowlist 为空,Python 脚本没有权限调用子代理。
**解决方案:** 改用 Prompt-based 模式(方案B)。
---
## 架构重构
### 之前(Script-based)
```
kb_workflow.py (Python脚本)
↓
sessions_spawn() # ❌ 失败(脚本没有权限)
```
### 之后(Prompt-based)
```
主代理读取 SKILL.md
↓
按照指导调用 sessions_spawn # ✅ 成功(主代理有权限)
```
### 文件变更
| 文件 | 变更 | 说明 |
|------|------|------|
| `SKILL.md` | 重写 | 详细定义 workflow 步骤,主代理直接读取执行 |
| `kb_workflow.py` | 简化 | 保留为辅助工具函数库(查重、归档、Git) |
| `skill_main.py` | 简化 | 只做触发检测,返回指导信息 |
| `SkillBug.md` | 更新 | 记录架构变更 |
### SKILL.md 结构
```
1. 触发条件
2. 步骤 1: 查重
3. 步骤 2: kb-link-parser
4. 步骤 3: 元数据技能
5. 步骤 4: 内容下载
6. 步骤 5: OCR(30分钟超时)
7. 步骤 6: 深度分析
8. 步骤 7: 归档和 Git
9. 步骤 8: 汇报完成
10. 目录结构
11. 支持平台
12. 注意事项
```
### kb_workflow.py 保留的函数
```python
extract_platform(filename) # 从文件名提取平台
extract_base_name(filename) # 从文件名提取基础名称
get_knowledge_base_path() # 获取知识库根目录
ensure_directories(base_path) # 确保目录存在
check_duplicate(url) # 查重
archive_and_git(analyzed_file) # 归档和 Git
generate_summary(...) # 生成汇报
```
---
### 2026-02-12: B站视频下载失败 - 403 Forbidden
**问题:** 使用 kb-bilibili-content 下载视频时返回 403 Forbidden 错误。
**原因:**
1. 视频可能需要B站会员权限
2. 或者需要登录cookie才能下载
3. 当前未配置 `BILIBILI_COOKIE` 环境变量
**复现场景:**
- 链接:https://www.bilibili.com/video/BV1Z9nXzUEhv
- 下载命令:`python3 bilibili_content.py --id "BV1Z9nXzUEhv" --type video`
- 结果:返回403错误页面,文件只有548字节
**影响:** 无法进行后续的OCR处理和深度分析
**建议解决方案:**
1. 配置 `BILIBILI_COOKIE` 环境变量
2. 或提供cookies文件路径
3. 或跳过OCR步骤,使用元数据+简介进行深度分析
**当前进度:**
- ✅ 步骤1:环境变量检查完成
- ✅ 步骤2:链接解析成功(bilibili_BV1Z9nXzUEhv)
- ✅ 步骤3:元数据提取成功(标题:新加坡分区漫游指南...)
- ❌ 步骤4:视频下载失败(403)
- ⏸️ 步骤5:OCR处理(阻塞)
- ⏸️ 步骤6:深度分析(阻塞)
- ⏸️ 步骤7:归档(阻塞)
-101
View File
@@ -1,101 +0,0 @@
# KB Zhihu Metadata Skill
**kb-zhihu-metadata** - Extract Zhihu question/answer/user metadata and crawl articles.
## Features
- Question metadata extraction
- Answer metadata extraction
- User information extraction
- **Article crawling using MediaCrawler** (NEW!)
- JSON output format
- Markdown conversion support
- **Cookie 支持**:可配置 ZHIHU_COOKIE 访问需要登录的内容
## 环境变量
| 变量名 | 用途 | 必需 |
|--------|------|------|
| `ZHIHU_COOKIE` | 知乎登录 Cookie | 推荐(访问专栏等需要) |
**获取 ZHIHU_COOKIE**:
1. 登录知乎网页版
2. 打开开发者工具 (F12) → Application → Cookies → zhihu.com
3. 复制 Cookie 字符串
```bash
# 添加到 ~/.kb_env
echo 'export ZHIHU_COOKIE="你的知乎Cookie"' >> ~/.kb_env
source ~/.kb_env
```
## Commands
```bash
# Question metadata
python zhihu_metadata.py --id "266309957" --type question
# Answer metadata
python zhihu_metadata.py --id "2003427891925509507" --type answer
# 专栏文章爬取 (使用 MediaCrawler) ⚡ NEW!
python zhihu_article_crawler.py --url "https://zhuanlan.zhihu.com/p/1894054073625461813" --output "~/work/knowledge-base/data/zhihu/"
```
## MediaCrawler 爬取文章
**新功能**:使用 MediaCrawler 爬取知乎专栏文章
### 环境要求
MediaCrawler 已安装在 `~/work/MediaCrawler`
### 使用方法
```bash
# 爬取单篇文章
python zhihu_article_crawler.py --url "https://zhuanlan.zhihu.com/p/XXXXX"
# 指定输出目录
python zhihu_article_crawler.py --url "https://zhuanlan.zhihu.com/p/XXXXX" --output "/path/to/output/"
# 输出示例
{
"success": true,
"type": "article",
"title": "如何高效的爬取微信公众号",
"filepath": "/home/xxx/work/knowledge-base/data/zhihu/zhihu_article_如何高效的爬取微信公众号.md",
"content_length": 4563,
"message": "爬取成功: 如何高效的爬取微信公众号"
}
```
### 工作原理
1. 调用 MediaCrawler 爬取文章
2. 从 JSON 输出中提取内容
3. 转换为 Markdown 格式(带 frontmatter)
4. 保存到指定目录
## Output
```json
{
"success": true,
"data": {
"platform": "zhihu",
"type": "question",
"id": "266309957",
"title": "问题标题",
"author": "作者名字",
"answer_count": 50,
"followers": 1000
}
}
```
## Installation
```bash
cp -r kb-zhihu-metadata/ ~/skills/
```
View File
-70
View File
@@ -1,70 +0,0 @@
import re
from bs4 import BeautifulSoup
from markdownify import MarkdownConverter as _MarkdownConverter
class MarkdownConverter:
"""HTML 转 Markdown 转换器,保留完整格式
处理知乎等平台的图片懒加载机制:
- 知乎图片的 src 属性可能是 SVG 占位符 (data:image/svg+xml;...)
- 真实图片 URL 存储在 data-actualsrc 或 data-original 属性中
"""
def __init__(self):
self.converter = _MarkdownConverter(
heading_style="ATX",
bullets="*,-",
code_blocks=True,
tables=True,
strip=["script", "style", "nav", "footer", "aside"]
)
def convert(self, html: str) -> str:
"""转换 HTML 为 Markdown
Args:
html: 输入的 HTML 字符串
Returns:
转换后的 Markdown 字符串
"""
# 预处理:处理懒加载图片
html = self._fix_lazy_load_images(html)
return self.converter.convert(html)
def _fix_lazy_load_images(self, html: str) -> str:
"""修复懒加载图片的 src 属性
将 data-actualsrc 或 data-original 属性的值替换到 src 属性中,
以便 markdownify 能够正确提取真实图片 URL。
同时移除 <noscript> 标签,因为知乎等平台在 noscript 中包含备用图片,
会导致修复后图片重复出现。
Args:
html: 输入的 HTML 字符串
Returns:
处理后的 HTML 字符串
"""
soup = BeautifulSoup(html, 'html.parser')
# 先移除 noscript 标签,避免图片重复
for noscript in soup.find_all('noscript'):
noscript.decompose()
for img in soup.find_all('img'):
src = img.get('src', '')
# 如果 src 不是占位符,保持原样
if not src or not src.startswith('data:'):
continue
# 尝试获取真实图片 URL
# 优先 data-actualsrc,其次 data-original
real_src = img.get('data-actualsrc') or img.get('data-original')
if real_src:
img['src'] = real_src
return str(soup)
-236
View File
@@ -1,236 +0,0 @@
#!/usr/bin/env python3
"""
知乎文章爬取脚本 - 基于 MediaCrawler
用于爬取知乎专栏文章并导出到知识库格式
"""
import argparse
import json
import os
import re
import sys
import subprocess
import shutil
from pathlib import Path
from datetime import datetime
# 配置
MEDIACRAWLER_DIR = Path.home() / "work" / "MediaCrawler"
DEFAULT_OUTPUT_DIR = Path.home() / "work" / "knowledge-base" / "data" / "zhihu"
def load_env():
"""加载环境变量"""
env_file = Path.home() / ".kb_env"
if env_file.exists():
with open(env_file) as f:
for line in f:
line = line.strip()
if line and not line.startswith("#") and "=" in line:
key, val = line.split("=", 1)
os.environ[key.strip()] = val.strip()
def get_zhihu_cookie():
"""获取知乎 Cookie"""
return os.environ.get("ZHIHU_COOKIE", "")
def crawl_zhihu_article(article_url: str, output_dir: Path) -> dict:
"""
使用 MediaCrawler 爬取知乎文章
Args:
article_url: 知乎文章链接
output_dir: 输出目录
Returns:
dict: 爬取结果
"""
# 清理旧数据
data_dir = MEDIACRAWLER_DIR / "data" / "zhihu" / "json"
data_dir.mkdir(parents=True, exist_ok=True)
# 备份现有数据
backup_dir = data_dir / "backup"
backup_dir.mkdir(exist_ok=True)
for f in data_dir.glob("detail_contents_*.json"):
shutil.move(str(f), str(backup_dir / f.name))
# 配置 MediaCrawler
config_file = MEDIACRAWLER_DIR / "config" / "zhihu_config.py"
# 读取配置文件并修改
with open(config_file) as f:
config_content = f.read()
# 强制更新配置中的 ID 列表(每次都更新,确保爬取正确的文章)
article_id = article_url.split("/")[-1]
# 使用正则表达式替换整个 ZHIHU_SPECIFIED_ID_LIST 部分
pattern = r'(ZHIHU_SPECIFIED_ID_LIST\s*=\s*\[)([\s\S]*?)(\])'
replacement = f'\\1\n "{article_url}", # article\n\\3'
new_config = re.sub(pattern, replacement, config_content)
# 如果正则替换失败(没有匹配到),使用简单替换
if new_config == config_content:
# 直接替换整个配置项
if 'ZHIHU_SPECIFIED_ID_LIST' in config_content:
# 找到配置项的范围并替换
lines = config_content.split('\n')
new_lines = []
in_list = False
for line in lines:
if 'ZHIHU_SPECIFIED_ID_LIST' in line and '=' in line:
new_lines.append(f'ZHIHU_SPECIFIED_ID_LIST = ["{article_url}", # article')
in_list = True
elif in_list and line.strip().startswith(']'):
new_lines.append(line)
in_list = False
elif not in_list:
new_lines.append(line)
new_config = '\n'.join(new_lines)
with open(config_file, "w") as f:
f.write(new_config)
# 运行 MediaCrawler
env = os.environ.copy()
cookie = get_zhihu_cookie()
if cookie:
env["ZHIHU_COOKIE"] = cookie
cmd = [
"python3", "main.py",
"--platform", "zhihu",
"--lt", "cookie",
"--type", "detail"
]
print(f"运行命令: {' '.join(cmd)}")
print(f"工作目录: {MEDIACRAWLER_DIR}")
result = subprocess.run(
cmd,
cwd=MEDIACRAWLER_DIR,
env=env,
capture_output=True,
text=True,
timeout=120
)
print(f"返回码: {result.returncode}")
if result.returncode != 0:
print(f"stderr: {result.stderr[:500]}")
# 查找最新的输出文件
output_files = sorted(data_dir.glob("detail_contents_*.json"), reverse=True)
if output_files:
latest_file = output_files[0]
with open(latest_file) as f:
data = json.load(f)
if data:
item = data[0]
return {
"success": True,
"title": item.get("title", ""),
"content": item.get("content_text", ""),
"url": item.get("content_url", ""),
"author": item.get("user_nickname", ""),
"created_time": item.get("created_time", 0),
"voteup": item.get("voteup_count", 0),
"comments": item.get("comment_count", 0),
"content_type": item.get("content_type", "article"),
}
return {"success": False, "error": "未获取到数据"}
def save_to_markdown(data: dict, output_path: Path) -> str:
"""保存为 Markdown 格式"""
created = datetime.fromtimestamp(data["created_time"]).strftime("%Y-%m-%d %H:%M:%S") if data.get("created_time") else ""
content = f"""---
title: "{data['title']}"
author: "{data['author']}"
url: "{data['url']}"
created: "{created}"
voteup: {data.get('voteup', 0)}
comments: {data.get('comments', 0)}
platform: zhihu
type: article
---
# {data['title']}
**作者**: {data['author']}
**来源**: [知乎专栏]({data['url']})
**发布时间**: {created}
**点赞**: {data.get('voteup', 0)}
**评论**: {data.get('comments', 0)}
---
{data['content']}
"""
with open(output_path, "w", encoding="utf-8") as f:
f.write(content)
return str(output_path)
def main():
parser = argparse.ArgumentParser(description="知乎文章爬取工具 (基于 MediaCrawler)")
parser.add_argument("--url", "-u", required=True, help="知乎文章链接")
parser.add_argument("--output", "-o", default=str(DEFAULT_OUTPUT_DIR), help="输出目录")
args = parser.parse_args()
# 加载环境变量
load_env()
# 确保输出目录存在
output_dir = Path(args.output)
output_dir.mkdir(parents=True, exist_ok=True)
print(f"目标文章: {args.url}")
print(f"输出目录: {output_dir}")
# 爬取文章
print("\n开始爬取...")
result = crawl_zhihu_article(args.url, output_dir)
if not result.get("success"):
print(f"爬取失败: {result.get('error', '未知错误')}")
sys.exit(1)
print(f"\n爬取成功!")
print(f"标题: {result['title']}")
print(f"内容长度: {len(result['content'])} 字符")
# 生成文件名
safe_title = "".join(c for c in result['title'] if c.isalnum() or c in " -_")[:50]
filename = f"zhihu_article_{safe_title}.md"
output_path = output_dir / filename
# 保存为 Markdown
save_to_markdown(result, output_path)
print(f"\n已保存到: {output_path}")
# 输出 JSON 格式
print("\n--- JSON 输出 ---")
print(json.dumps({
"success": True,
"type": "article",
"title": result["title"],
"filepath": str(output_path),
"content_length": len(result["content"]),
"message": f"爬取成功: {result['title']}"
}, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()
-479
View File
@@ -1,479 +0,0 @@
"""
知乎 Playwright 内容提取器
使用 Playwright 浏览器自动化工具抓取知乎内容
"""
import re
import sys
import os
from typing import Dict, Optional, Any
from urllib.parse import urlparse
# 添加当前目录到 Python 路径以支持绝对导入
current_dir = os.path.dirname(os.path.abspath(__file__))
if current_dir not in sys.path:
sys.path.insert(0, current_dir)
try:
from .markdown_converter import MarkdownConverter
except ImportError:
from markdown_converter import MarkdownConverter
class PlaywrightBrowser:
"""Playwright 浏览器管理器"""
def __init__(self, headless: bool = True, timeout: int = 30):
self.headless = headless
self.timeout = timeout * 1000
self.browser = None
self.context = None
self.page = None
def start(self, cookie: str = None):
"""启动浏览器"""
from playwright.sync_api import sync_playwright
self.playwright = sync_playwright().start()
self.browser = self.playwright.chromium.launch(headless=self.headless)
# 创建浏览器上下文
context_options = {
'user_agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36'
}
self.context = self.browser.new_context(**context_options)
# 如果提供了 Cookie,设置 Cookie
if cookie:
self._add_cookies(cookie)
self.page = self.context.new_page()
self.page.set_default_timeout(self.timeout)
def _add_cookies(self, cookie_str: str):
"""添加 Cookie 到浏览器上下文"""
cookies = []
for item in cookie_str.split(';'):
item = item.strip()
if '=' in item:
name, value = item.split('=', 1)
cookies.append({
'name': name.strip(),
'value': value.strip(),
'domain': '.zhihu.com',
'path': '/',
'secure': True,
'httpOnly': False,
'sameSite': 'Lax'
})
self.context.add_cookies(cookies)
def close(self):
"""关闭浏览器"""
if self.page:
self.page.close()
if self.context:
self.context.close()
if self.browser:
self.browser.close()
if hasattr(self, 'playwright'):
self.playwright.stop()
def goto(self, url: str, wait_until: str = "networkidle"):
"""导航到指定URL"""
return self.page.goto(url, wait_until=wait_until)
def wait_for_selector(self, selector: str, timeout: int = None):
"""等待选择器出现"""
timeout_ms = timeout * 1000 if timeout else self.timeout
return self.page.wait_for_selector(selector, timeout=timeout_ms)
def query_selector(self, selector: str):
"""查询选择器"""
return self.page.query_selector(selector)
def query_selector_all(self, selector: str):
"""查询所有匹配的选择器"""
return self.page.query_selector_all(selector)
def evaluate(self, script: str):
"""执行 JavaScript 代码"""
return self.page.evaluate(script)
class ZhihuPlaywrightExtractor:
"""知乎内容提取器 - 使用 Playwright"""
URL_PATTERNS = {
'zvideo': r'zhihu\.com/zvideo/(\d+)',
'zhuanlan': r'zhuanlan\.zhihu\.com/p/(\d+)',
'answer': r'zhihu\.com/question/\d+/answer/(\d+)',
'question': r'zhihu\.com/question/(\d+)',
}
def __init__(self, headless: bool = True, timeout: int = 30, cookie: str = None):
"""
初始化提取器
Args:
headless: 是否使用无头浏览器
timeout: 超时时间(秒)
cookie: 知乎 Cookie 字符串(可选)
"""
self.cookie = cookie
self.browser = PlaywrightBrowser(headless=headless, timeout=timeout)
self.markdown_converter = MarkdownConverter()
def identify_url_type(self, url: str) -> Optional[str]:
"""
识别 URL 类型
Returns:
'answer', 'article', 'question' 或 None
"""
for url_type, pattern in self.URL_PATTERNS.items():
if re.search(pattern, url):
return url_type
return None
def extract_id(self, url: str) -> Optional[str]:
"""
从 URL 中提取内容 ID
Returns:
内容 ID 或 None
"""
url_type = self.identify_url_type(url)
if url_type:
pattern = self.URL_PATTERNS[url_type]
match = re.search(pattern, url)
if match:
return match.group(1)
return None
def _extract_text_content(self, element) -> str:
"""提取元素的文本内容"""
if element is None:
return ''
return element.inner_text() if hasattr(element, 'inner_text') else ''
def _extract_html_content(self, element) -> str:
"""提取元素的 HTML 内容"""
if element is None:
return ''
return element.inner_html() if hasattr(element, 'inner_html') else ''
def _extract_zvideo_content(self) -> Dict[str, Any]:
"""提取知乎视频内容"""
result = {
'type': 'zvideo',
'title': '',
'author': '',
'content': '',
'content_markdown': '',
'publish_time': '',
'vote_up': 0,
}
# 打印页面标题用于调试
page_title = self.browser.page.title()
print(f"页面标题: {page_title}")
# 提取视频标题
title_elem = self.browser.query_selector('.VideoPlaylistInfo-title')
if title_elem:
result['title'] = self._extract_text_content(title_elem)
# 备用:尝试其他选择器
if not result['title']:
title_elem = self.browser.query_selector('h1')
if title_elem:
result['title'] = self._extract_text_content(title_elem)
# 提取作者信息
author_elem = self.browser.query_selector('.AuthorInfo-name')
if author_elem:
result['author'] = self._extract_text_content(author_elem)
# 提取视频描述
desc_elem = self.browser.query_selector('.Video-desc')
if desc_elem:
html_content = self._extract_html_content(desc_elem)
result['content'] = html_content
result['content_markdown'] = self.markdown_converter.convert(html_content)
# 提取发布时间
time_elem = self.browser.query_selector('.Video-meta time')
if time_elem:
result['publish_time'] = time_elem.get_attribute('datetime') or self._extract_text_content(time_elem)
# 提取点赞数
vote_elem = self.browser.query_selector('.VideoLikeButton-count')
if vote_elem:
vote_text = self._extract_text_content(vote_elem)
try:
result['vote_up'] = int(vote_text.replace(',', ''))
except (ValueError, AttributeError):
pass
return result
def _extract_zhuanlan_content(self) -> Dict[str, Any]:
"""提取知乎专栏文章内容"""
result = {
'type': 'zhuanlan',
'title': '',
'author': '',
'content': '',
'content_markdown': '',
'publish_time': '',
'vote_up': 0,
}
# 打印页面标题用于调试
page_title = self.browser.page.title()
print(f"页面标题: {page_title}")
# 提取文章标题
title_elem = self.browser.query_selector('.Post-Title')
if title_elem:
result['title'] = self._extract_text_content(title_elem)
# 备用
if not result['title']:
title_elem = self.browser.query_selector('h1')
if title_elem:
result['title'] = self._extract_text_content(title_elem)
# 提取作者信息
author_elem = self.browser.query_selector('.AuthorInfo-name')
if author_elem:
result['author'] = self._extract_text_content(author_elem)
# 提取文章内容
content_elem = self.browser.query_selector('.Post-RichText')
if content_elem:
html_content = self._extract_html_content(content_elem)
result['content'] = html_content
result['content_markdown'] = self.markdown_converter.convert(html_content)
# 提取发布时间
time_elem = self.browser.query_selector('.ContentItem-time')
if time_elem:
result['publish_time'] = time_elem.get_attribute('data-tooltip') or self._extract_text_content(time_elem)
# 提取点赞数
vote_elem = self.browser.query_selector('.VoteButton--up .VoteButton-label')
if vote_elem:
vote_text = self._extract_text_content(vote_elem)
try:
result['vote_up'] = int(vote_text.replace(',', ''))
except (ValueError, AttributeError):
pass
return result
def _extract_answer_content(self) -> Dict[str, Any]:
"""提取知乎回答内容"""
result = {
'type': 'answer',
'title': '', # 问题标题
'question_id': '',
'answer_id': '',
'author': '',
'content': '',
'content_markdown': '',
'publish_time': '',
'vote_up': 0,
}
# 打印页面标题用于调试
page_title = self.browser.page.title()
print(f"页面标题: {page_title}")
# 提取问题标题
question_title_elem = self.browser.query_selector('.QuestionHeader-title')
if not question_title_elem:
question_title_elem = self.browser.query_selector('.QuestionHeader-content h1')
if question_title_elem:
result['title'] = self._extract_text_content(question_title_elem)
# 提取回答内容
content_elem = self.browser.query_selector('.RichText')
if not content_elem:
content_elem = self.browser.query_selector('.AnswerItem-richText')
if not content_elem:
content_elem = self.browser.query_selector('[class*="RichText"]')
if content_elem:
html_content = self._extract_html_content(content_elem)
result['content'] = html_content
result['content_markdown'] = self.markdown_converter.convert(html_content)
# 提取作者信息
author_elem = self.browser.query_selector('.AuthorInfo-name')
if author_elem:
result['author'] = self._extract_text_content(author_elem)
# 提取发布时间
time_elem = self.browser.query_selector('.AnswerItem-time')
if not time_elem:
time_elem = self.browser.query_selector('.ContentItem-time')
if time_elem:
result['publish_time'] = time_elem.get_attribute('data-tooltip') or self._extract_text_content(time_elem)
# 提取点赞数
vote_elem = self.browser.query_selector('.VoteButton--up .VoteButton-label')
if not vote_elem:
vote_elem = self.browser.query_selector('.VoteButton .Count')
if vote_elem:
vote_text = self._extract_text_content(vote_elem)
try:
result['vote_up'] = int(vote_text.replace(',', ''))
except (ValueError, AttributeError):
pass
return result
def _extract_question_content(self) -> Dict[str, Any]:
"""提取知乎问题内容(包含所有回答)"""
result = {
'type': 'question',
'title': '',
'question_id': '',
'author': '',
'content': '',
'content_markdown': '',
'publish_time': '',
'answer_count': 0,
}
# 打印页面标题用于调试
page_title = self.browser.page.title()
print(f"页面标题: {page_title}")
# 提取问题标题
title_elem = self.browser.query_selector('.QuestionHeader-title')
if not title_elem:
title_elem = self.browser.query_selector('.QuestionHeader-content h1')
if title_elem:
result['title'] = self._extract_text_content(title_elem)
# 提取问题描述
desc_elem = self.browser.query_selector('.QuestionHeader-detail')
if not desc_elem:
desc_elem = self.browser.query_selector('.QuestionRichText')
if desc_elem:
html_content = self._extract_html_content(desc_elem)
result['content'] = html_content
result['content_markdown'] = self.markdown_converter.convert(html_content)
# 提取回答数量
answer_count_elem = self.browser.query_selector('.List-headerText span')
if answer_count_elem:
count_text = self._extract_text_content(answer_count_elem)
try:
# 提取数字
import re
match = re.search(r'(\d+)', count_text)
if match:
result['answer_count'] = int(match.group(1))
except (ValueError, AttributeError):
pass
return result
def extract(self, url: str) -> Optional[Dict[str, Any]]:
"""
从 URL 提取知乎内容
Args:
url: 知乎 URL
Returns:
包含提取内容的字典,结构如下:
{
'type': 'zvideo' | 'zhuanlan',
'title': str,
'author': str,
'content': str (HTML),
'content_markdown': str (Markdown),
'publish_time': str,
'vote_up': int,
}
如果提取失败返回 None
"""
try:
# 识别 URL 类型
url_type = self.identify_url_type(url)
if not url_type:
print(f"无法识别 URL 类型: {url}")
return None
print(f"URL 类型: {url_type}")
print(f"启动浏览器...")
# 启动浏览器(传入 Cookie)
self.browser.start(cookie=self.cookie)
print(f"访问页面: {url}")
# 访问页面
self.browser.goto(url, wait_until="domcontentloaded")
print("等待页面加载...")
import time
time.sleep(3) # 等待 JavaScript 执行
# 根据类型提取内容
if url_type == 'zvideo':
result = self._extract_zvideo_content()
elif url_type == 'zhuanlan':
result = self._extract_zhuanlan_content()
elif url_type == 'answer':
result = self._extract_answer_content()
elif url_type == 'question':
result = self._extract_question_content()
else:
print(f"不支持的类型: {url_type}")
return None
# 添加额外信息
result['url'] = url
result['content_id'] = self.extract_id(url)
print(f"提取成功! 标题: {result.get('title', '')[:50]}...")
return result
except Exception as e:
print(f"提取知乎内容时出错: {e}")
import traceback
traceback.print_exc()
return None
finally:
# 关闭浏览器
self.browser.close()
def __enter__(self):
"""上下文管理器入口"""
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""上下文管理器出口"""
self.browser.close()
return False
# 便捷函数
def extract_zhihu_content(url: str, headless: bool = True, timeout: int = 30) -> Optional[Dict[str, Any]]:
"""
便捷函数:提取知乎内容
Args:
url: 知乎 URL
headless: 是否使用无头浏览器
timeout: 超时时间(秒)
Returns:
提取的内容字典
"""
with ZhihuPlaywrightExtractor(headless=headless, timeout=timeout) as extractor:
return extractor.extract(url)
-206
View File
@@ -1,206 +0,0 @@
#!/usr/bin/env python3
"""
知乎元数据获取工具
支持以下内容类型:
- zvideo: 知乎视频 (https://www.zhihu.com/zvideo/xxx)
- zhuanlan: 知乎专栏文章 (https://zhuanlan.zhihu.com/p/xxx)
- answer: 知乎回答 (https://www.zhihu.com/question/xxx/answer/yyy)
- question: 知乎问题 (https://www.zhihu.com/question/xxx)
"""
import argparse
import json
import sys
import os
from typing import Dict, Optional, Any
# 添加当前目录到 Python 路径
current_dir = os.path.dirname(os.path.abspath(__file__))
if current_dir not in sys.path:
sys.path.insert(0, current_dir)
from zhihu_extractor import ZhihuPlaywrightExtractor
def extract_zvideo(video_id: str, headless: bool = True, cookie: str = None) -> Dict[str, Any]:
"""
获取知乎视频元数据
Args:
video_id: 视频 ID (数字字符串)
headless: 是否使用无头浏览器
cookie: 知乎 Cookie(可选)
Returns:
视频元数据字典
"""
url = f"https://www.zhihu.com/zvideo/{video_id}"
return _extract_metadata(url, "zvideo", headless, cookie)
def extract_zhuanlan(article_id: str, headless: bool = True, cookie: str = None) -> Dict[str, Any]:
"""
获取知乎专栏文章元数据
Args:
article_id: 文章 ID (数字字符串)
headless: 是否使用无头浏览器
cookie: 知乎 Cookie(可选)
Returns:
文章元数据字典
"""
url = f"https://zhuanlan.zhihu.com/p/{article_id}"
return _extract_metadata(url, "zhuanlan", headless, cookie)
def extract_answer(answer_id: str, headless: bool = True, cookie: str = None) -> Dict[str, Any]:
"""
获取知乎回答元数据
Args:
answer_id: 回答 ID (数字字符串)
headless: 是否使用无头浏览器
cookie: 知乎 Cookie(可选)
Returns:
回答元数据字典
"""
# 回答需要构建包含问题ID的URL,但answer_id本身足以识别
# 使用一个占位符问题ID,因为提取器会从URL中提取answer_id
url = f"https://www.zhihu.com/question/0/answer/{answer_id}"
return _extract_metadata(url, "answer", headless, cookie)
def extract_question(question_id: str, headless: bool = True, cookie: str = None) -> Dict[str, Any]:
"""
获取知乎问题元数据
Args:
question_id: 问题 ID (数字字符串)
headless: 是否使用无头浏览器
cookie: 知乎 Cookie(可选)
Returns:
问题元数据字典
"""
url = f"https://www.zhihu.com/question/{question_id}"
return _extract_metadata(url, "question", headless, cookie)
def _extract_metadata(url: str, content_type: str, headless: bool, cookie: str = None) -> Dict[str, Any]:
"""
内部函数:提取元数据
Args:
url: 知乎 URL
content_type: 内容类型 (zvideo/zhuanlan)
headless: 是否使用无头浏览器
cookie: 知乎 Cookie(可选)
Returns:
元数据字典
"""
extractor = ZhihuPlaywrightExtractor(headless=headless, cookie=cookie)
result = extractor.extract(url)
if result:
# 标准化输出格式
metadata = {
"type": content_type,
"id": result.get("content_id", ""),
"url": result.get("url", url),
"title": result.get("title", ""),
"author": result.get("author", ""),
"publish_time": result.get("publish_time", ""),
"vote_up": result.get("vote_up", 0),
"content_markdown": result.get("content_markdown", ""),
"raw_html": result.get("content", ""),
}
return metadata
else:
return {
"type": content_type,
"error": "Failed to extract content",
"url": url
}
def main():
"""命令行入口"""
parser = argparse.ArgumentParser(
description="知乎元数据获取工具",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
支持的类型:
zvideo - 知乎视频 (https://www.zhihu.com/zvideo/123456)
zhuanlan - 知乎专栏文章 (https://zhuanlan.zhihu.com/p/123456)
answer - 知乎回答 (https://www.zhihu.com/question/xxx/answer/yyy)
question - 知乎问题 (https://www.zhihu.com/question/xxx)
示例:
python zhihu_metadata.py --id "123456" --type zvideo
python zhihu_metadata.py --id "123456" --type zhuanlan
python zhihu_metadata.py --id "123456" --type answer
python zhihu_metadata.py --id "123456" --type question
"""
)
parser.add_argument(
"--id", "-i",
required=True,
help="知乎内容 ID (视频ID/文章ID/回答ID/问题ID)"
)
parser.add_argument(
"--type", "-t",
required=True,
choices=["zvideo", "zhuanlan", "answer", "question"],
help="内容类型: zvideo=视频, zhuanlan=专栏文章, answer=回答, question=问题"
)
parser.add_argument(
"--output", "-o",
default=None,
help="输出文件路径 (默认: stdout)"
)
parser.add_argument(
"--headless",
action="store_true",
default=True,
help="使用无头浏览器模式 (默认: True)"
)
args = parser.parse_args()
# 获取知乎 Cookie(优先使用环境变量,其次使用参数)
cookie = os.environ.get('ZHIHU_COOKIE')
# 提取元数据(传入 Cookie)
if args.type == "zvideo":
result = extract_zvideo(args.id, headless=args.headless, cookie=cookie)
elif args.type == "zhuanlan":
result = extract_zhuanlan(args.id, headless=args.headless, cookie=cookie)
elif args.type == "answer":
result = extract_answer(args.id, headless=args.headless, cookie=cookie)
elif args.type == "question":
result = extract_question(args.id, headless=args.headless, cookie=cookie)
else:
print(f"不支持的类型: {args.type}")
sys.exit(1)
# 输出结果
json_output = json.dumps(result, ensure_ascii=False, indent=2)
if args.output:
with open(args.output, 'w', encoding='utf-8') as f:
f.write(json_output)
print(f"结果已保存到: {args.output}")
else:
print(json_output)
if __name__ == "__main__":
main()
-1
View File
@@ -1 +0,0 @@
kb-deep-analyzer
-1
View File
@@ -1 +0,0 @@
kb-zhihu-metadata