diff --git a/kb-bilibili-metadata/SKILL.md b/kb-bilibili-metadata/SKILL.md new file mode 100644 index 0000000..d90d862 --- /dev/null +++ b/kb-bilibili-metadata/SKILL.md @@ -0,0 +1,46 @@ +# 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/ +``` diff --git a/kb-bilibili-metadata/__init__.py b/kb-bilibili-metadata/__init__.py new file mode 100644 index 0000000..0614e1f --- /dev/null +++ b/kb-bilibili-metadata/__init__.py @@ -0,0 +1,8 @@ +""" +bilibili-metadata 技能包 + +B站元数据提取工具 +""" + +__version__ = "1.0.0" +__author__ = "OpenClaw" diff --git a/kb-bilibili-metadata/__pycache__/bilibili_extractor.cpython-312.pyc b/kb-bilibili-metadata/__pycache__/bilibili_extractor.cpython-312.pyc new file mode 100644 index 0000000..49393a3 Binary files /dev/null and b/kb-bilibili-metadata/__pycache__/bilibili_extractor.cpython-312.pyc differ diff --git a/kb-bilibili-metadata/bilibili_extractor.py b/kb-bilibili-metadata/bilibili_extractor.py new file mode 100644 index 0000000..86fee7b --- /dev/null +++ b/kb-bilibili-metadata/bilibili_extractor.py @@ -0,0 +1,666 @@ +#!/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)) diff --git a/kb-bilibili-metadata/bilibili_metadata.py b/kb-bilibili-metadata/bilibili_metadata.py new file mode 100644 index 0000000..f4635fd --- /dev/null +++ b/kb-bilibili-metadata/bilibili_metadata.py @@ -0,0 +1,228 @@ +#!/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() diff --git a/kb-deep-analyzer/SKILL.md b/kb-deep-analyzer/SKILL.md new file mode 100644 index 0000000..0977cab --- /dev/null +++ b/kb-deep-analyzer/SKILL.md @@ -0,0 +1,102 @@ +# 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" + } + } +} +``` + +会自动转换为扁平格式进行分析。 diff --git a/kb-deep-analyzer/__pycache__/deep_analyzer.cpython-312.pyc b/kb-deep-analyzer/__pycache__/deep_analyzer.cpython-312.pyc new file mode 100644 index 0000000..08d1b7a Binary files /dev/null and b/kb-deep-analyzer/__pycache__/deep_analyzer.cpython-312.pyc differ diff --git a/kb-deep-analyzer/ai_client.py b/kb-deep-analyzer/ai_client.py new file mode 100644 index 0000000..cee3c73 --- /dev/null +++ b/kb-deep-analyzer/ai_client.py @@ -0,0 +1,514 @@ +"""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) diff --git a/kb-deep-analyzer/deep_analyzer.py b/kb-deep-analyzer/deep_analyzer.py new file mode 100644 index 0000000..93d4298 --- /dev/null +++ b/kb-deep-analyzer/deep_analyzer.py @@ -0,0 +1,590 @@ +#!/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) diff --git a/kb-deep-analyzer/deep_analyzer.py.bak b/kb-deep-analyzer/deep_analyzer.py.bak new file mode 100644 index 0000000..59c62c1 --- /dev/null +++ b/kb-deep-analyzer/deep_analyzer.py.bak @@ -0,0 +1,1727 @@ +#!/usr/bin/env python3 +"""Deep Analyzer - AI 深度内容分析工具 + +支持根据内容时长选择不同分析策略： +- 短视频 (<10分钟): 轻量分析 +- 长视频 (>=10分钟): 详细时间轴分析 + +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, Optional, Tuple +from dataclasses import dataclass, field, asdict + + +# ============== 数据模型 ============== + +@dataclass +class TimelineEntry: + """时间轴条目""" + index: int = 0 + start_time: str = "" + end_time: str = "" + title: str = "" + summary: str = "" + + +@dataclass +class AnalysisResult: + """分析结果""" + summary: str = "" + main_topics: List[str] = field(default_factory=list) + key_insights: List[str] = field(default_factory=list) + action_items: List[str] = field(default_factory=list) + recommended_tags: List[str] = field(default_factory=list) + full_analysis: str = "" + timeline: List[Dict[str, Any]] = field(default_factory=list) + + +# ============== AI 客户端 ============== + +# ============== 流式写入器 ============== + +import threading + + +class StreamingWriter: + """ + 流式写入器 - 支持多线程并发写入到文档不同位置 + + 使用方法: + 1. 初始化: writer = StreamingWriter(output_path, num_chunks) + 2. 预先创建带占位符的文档结构 + 3. 每批完成后调用 writer.write_batch(batch_results) + 4. 完成后调用 writer.mark_complete() 清理占位符 + """ + + def __init__(self, output_path: str, num_chunks: int = 0, batch_size: int = 3): + self.output_path = Path(output_path) + self.num_chunks = num_chunks + self.batch_size = batch_size + self.lock = threading.Lock() + self.written = {} # 记录已写入的位置 + self.batch_num = 0 + + def initialize_document(self, metadata: Dict, num_chunks: int) -> str: + """ + 初始化文档结构，返回带占位符的内容 + + 文档结构: + --- + title: "xxx" + ... + --- + + # 标题 + + ## 摘要 + + ## 主要主题 + + ## 关键见解 + + ## 建议行动 + + ## 时间轴 + ### 第 N 部分 (00:00 - 00:00) + [ANALYZING...] + """ + self.num_chunks = num_chunks + + # 生成时间轴占位符 + timeline_placeholders = "" + for i in range(num_chunks): + timeline_placeholders += f"""### 第 {i + 1} 部分 (待分析) +[ANALYZING_BATCH_{i // self.batch_size + 1}] + +""" + + content = f"""--- +title: "{metadata.get('title', '未命名')}" +date: {datetime.now().strftime('%Y-%m-%d')} +duration: {format_duration(metadata.get('duration', 0))} +author: {metadata.get('author', '未知')} +analyze_time: {datetime.now().strftime('%Y-%m-%d %H:%M')} +bvid: {metadata.get('bvid', '')} +platform: {metadata.get('platform', 'bilibili')} +--- + +# {metadata.get('title', '未命名')} + +## 摘要 +[WAITING_FOR_ANALYSIS...] + +## 主要主题 +[WAITING_FOR_ANALYSIS...] + +## 关键见解 +[WAITING_FOR_ANALYSIS...] + +## 建议行动 +[WAITING_FOR_ANALYSIS...] + +## 时间轴 +{timeline_placeholders} + +--- +*分析时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}* +""" + + # 写入初始化内容 + self.output_path.parent.mkdir(parents=True, exist_ok=True) + with open(self.output_path, 'w', encoding='utf-8') as f: + f.write(content) + + print(f"[StreamingWriter] 初始化文档: {self.output_path}") + return content + + def write_batch(self, batch_results: List[Dict], num_batches: int) -> None: + """ + 写入一批结果 + + Args: + batch_results: 当前批的结果列表 + num_batches: 总批次数 + """ + self.batch_num += 1 + + with self.lock: + # 读取当前内容 + if not self.output_path.exists(): + print(f"[StreamingWriter] 警告: 文件不存在 {self.output_path}") + return + + content = self.output_path.read_text(encoding='utf-8') + + for result in batch_results: + idx = result['index'] + start_time = result['start_time'] + end_time = result['end_time'] + summary = result['summary'] + + # 查找并替换占位符 + old_placeholder = f"### 第 {idx + 1} 部分 ({start_time} - {end_time})\n[ANALYZING_BATCH_{self.batch_num}]" + new_content = f"""### 第 {idx + 1} 部分 ({start_time} - {end_time}) +{summary}""" + + content = content.replace(old_placeholder, new_content, 1) + self.written[idx] = True + + print(f"[StreamingWriter] 写入段 {idx + 1}/{self.num_chunks}") + + # 如果这是最后一批，更新其他状态 + if self.batch_num >= num_batches: + # 更新摘要等 + pass + + # 写回文件 + self.output_path.write_text(content, encoding='utf-8') + print(f"[StreamingWriter] 批次 {self.batch_num}/{num_batches} 完成") + + def update_summary(self, summary_text: str, topics: List[str], insights: List[str]) -> None: + """更新摘要等全局内容""" + with self.lock: + if not self.output_path.exists(): + return + + content = self.output_path.read_text(encoding='utf-8') + + # 更新摘要 + if "[WAITING_FOR_ANALYSIS...]" in content: + content = content.replace("## 摘要\n[WAITING_FOR_ANALYSIS...]", f"## 摘要\n{summary_text}") + + # 更新主题 + topics_text = "\n".join(f"- {t}" for t in topics) + if "## 主要主题\n[WAITING_FOR_ANALYSIS...]" in content: + content = content.replace("## 主要主题\n[WAITING_FOR_ANALYSIS...]", f"## 主要主题\n{topics_text}") + + # 更新见解 + insights_text = "\n".join(f"- {i}" for i in insights) + if "## 关键见解\n[WAITING_FOR_ANALYSIS...]" in content: + content = content.replace("## 关键见解\n[WAITING_FOR_ANALYSIS...]", f"## 关键见解\n{insights_text}") + + self.output_path.write_text(content, encoding='utf-8') + print(f"[StreamingWriter] 更新摘要和主题") + + def cleanup_placeholder(self, placeholder: str) -> None: + """清理单个占位符""" + with self.lock: + if not self.output_path.exists(): + return + + content = self.output_path.read_text(encoding='utf-8') + content = content.replace(placeholder, "") + self.output_path.write_text(content, encoding='utf-8') + + +class SimpleAIClient: + """简化的 AI 客户端 - 支持多提供商 fallback""" + + # 时长阈值（秒） + SHORT_VIDEO_THRESHOLD = 600 # 10分钟 + LONG_VIDEO_THRESHOLD = 1800 # 30分钟 + + 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", "gpt-4") + self.base_url = base_url or os.environ.get("AI_BASE_URL", "") + self.max_retries = 3 # 最大重试次数 + self.max_concurrent = 3 # 最大并发数 + + def generate(self, prompt: str, max_tokens: int = 1000, task_info: str = None) -> str: + """生成文本，支持重试""" + # 每次调用时重新读取环境变量，解决 exec 传递问题 + 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 未设置，请配置环境变量") + + task_desc = f"[{task_info}]" if task_info else "" + + last_error = None + for attempt in range(self.max_retries): + try: + import time + start_time = time.time() + print(f"[AI Client] {task_desc} 开始调用 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] {task_desc} AI 调用完成，耗时: {elapsed:.2f}秒") + return result + except Exception as e: + elapsed = time.time() - start_time if 'start_time' in dir() else 0 + last_error = e + if attempt < self.max_retries - 1: + print(f"[AI Client] {task_desc} 第 {attempt + 1} 次调用失败 ({elapsed:.2f}秒)，{self.max_retries - attempt - 1} 次重试剩余: {e}") + continue + else: + print(f"[AI Client] {task_desc} 调用失败，已重试 {self.max_retries} 次: {e}") + raise + + def generate_json(self, prompt: str) -> Dict[str, Any]: + """生成 JSON 格式响应""" + response = self.generate(prompt, max_tokens=4000) + try: + return json.loads(response) + except json.JSONDecodeError: + # 尝试从文本中提取 JSON + return self._parse_json_from_text(response) + + def _call_openai(self, prompt: str, max_tokens: int, api_key: str = None, model: str = None, base_url: str = None) -> str: + """调用 OpenAI API""" + import requests + url = f"{self.base_url}/chat/completions" if self.base_url else "https://api.openai.com/v1/chat/completions" + + headers = { + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json" + } + + payload = { + "model": self.model, + "max_tokens": max_tokens, + "messages": [{"role": "user", "content": prompt}] + } + + response = requests.post(url, headers=headers, json=payload, timeout=600) + response.raise_for_status() + response.raise_for_status() + return response.json()['choices'][0]['message']['content'] + + def _call_zhipu(self, prompt: str, max_tokens: int, api_key: str = None, model: str = None, base_url: str = None) -> str: + """调用智谱 AI API - 返回 reasoning_content""" + import requests + base_url = self.base_url or "https://open.bigmodel.cn/api/coding/paas/v4" + url = f"{base_url}/chat/completions" + + headers = {"Authorization": f"Bearer {self.api_key}"} + payload = { + "model": self.model, + "messages": [{"role": "user", "content": prompt}], + "max_tokens": max_tokens + } + + response = requests.post(url, headers=headers, json=payload, timeout=600) + response.raise_for_status() + response.raise_for_status() + data = response.json() + + message = data.get('choices', [{}])[0].get('message', {}) + content = message.get('content', '') + reasoning = message.get('reasoning_content', '') + + # 如果 content 为空但有 reasoning_content，从 reasoning 中提取关键句 + if not content and reasoning: + # 从 reasoning_content 中提取最后几句话作为答案 + 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 _parse_json_from_text(self, text: str) -> Dict[str, Any]: + """从文本中解析 JSON，改进版本，支持更多格式""" + # 尝试直接解析 + try: + return json.loads(text) + except json.JSONDecodeError: + pass + + # 尝试提取 JSON 块 + json_match = re.search(r'```(?:json)?\s*([\s\S]*?)\s*```', text) + if json_match: + try: + return json.loads(json_match.group(1)) + except: + pass + + # 尝试匹配 { ... } + json_match = re.search(r'(\{[\s\S]*\})', text) + if json_match: + try: + return json.loads(json_match.group(1)) + except: + pass + + return {'raw_response': text} + + def _call_minimax(self, prompt: str, max_tokens: int, api_key: str = None, model: str = None, base_url: str = None) -> str: + """调用 MiniMax API""" + import requests + url = self.base_url or "https://api.minimaxi.com/anthropic/v1/messages" + + headers = { + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json" + } + + payload = { + "model": self.model, + "max_tokens": max_tokens, + "messages": [{"role": "user", "content": prompt}] + } + + response = requests.post(url, headers=headers, json=payload, timeout=600) + response.raise_for_status() + 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: + """深度内容分析器""" + + SHORT_VIDEO_THRESHOLD = 600 # 10分钟 + LONG_VIDEO_THRESHOLD = 1800 # 30分钟 + + 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", + on_batch_complete: callable = None, + batch_size: int = 3 + ) -> AnalysisResult: + """ + 分析内容 + + Args: + metadata: 元数据字典 + content: 内容数据字典（包含 subtitle, description 等） + content_type: 内容类型 + on_batch_complete: 每批完成时的回调函数 (batch_results) => None + batch_size: 每批处理数（默认3） + + Returns: + AnalysisResult 分析结果 + """ + # 先标准化元数据 + metadata = normalize_metadata(metadata) + + title = metadata.get('title', '未知') + duration = metadata.get('duration', 0) + + # 支持多种内容格式 + text = content.get('text', '') or content.get('subtitle', '') or content.get('description', '') + segments = content.get('segments', []) + + has_content = bool(text and text.strip()) + has_segments = bool(segments) + + print(f"[DeepAnalyzer] 开始分析: {title}") + print(f"[DeepAnalyzer] 类型: {content_type}, 时长: {duration}秒") + print(f"[DeepAnalyzer] 文本存在: {has_content}, 分段数: {len(segments)}") + print(f"[DeepAnalyzer] 文本长度: {len(text) if text else 0} 字符") + + CHUNK_THRESHOLD = 3000 + use_chunking = has_content and len(text) > CHUNK_THRESHOLD + + if has_content or has_segments: + if use_chunking: + print(f"[DeepAnalyzer] 文本超过 {CHUNK_THRESHOLD} 字，使用批处理分块分析") + result = self._analyze_chunks_concurrent( + title, text, duration, segments, metadata, + batch_size=batch_size, + on_segment_complete=on_batch_complete + ) + elif duration < self.SHORT_VIDEO_THRESHOLD: + print(f"[DeepAnalyzer] 使用轻量分析策略 (时长 < 10分钟)") + result = self._analyze_short_with_subtitle(title, text, duration, segments) + else: + print(f"[DeepAnalyzer] 使用详细时间轴分析策略 (时长 >= 10分钟)") + result = self._analyze_long_with_subtitle(title, text, duration, segments, content_type, metadata) + else: + print(f"[DeepAnalyzer] 降级处理（无内容）") + result = self._analyze_fallback(title, text, metadata) + + return result + + def _analyze_short_with_subtitle(self, title: str, text: str, duration: int, segments: list = None) -> AnalysisResult: + """有内容的短视频轻量分析""" + result = AnalysisResult() + text_sample = self._get_text_sample(text, max_chars=5000) + + analysis = self._analyze_with_ai_short_text(text_sample, title, duration) + result.summary = analysis.get('summary', title) + result.main_topics = analysis.get('topics', []) + result.key_insights = analysis.get('insights', []) + result.action_items = analysis.get('actions', []) + result.timeline = self._generate_timeline_from_segments(title, duration, segments) + + return result + + def _analyze_long_with_subtitle(self, title: str, text: str, duration: int, + segments: list = None, content_type: str = None, + metadata: Dict = None) -> AnalysisResult: + """有内容的长视频详细分析""" + result = AnalysisResult() + + if duration >= self.SHORT_VIDEO_THRESHOLD and content_type == "video": + # 10分钟以上使用分段分析 + result = self._analyze_in_chunks(title, text, duration, metadata) + else: + # 直接分析 + text_sample = self._get_text_sample(text, max_chars=5000) + analysis = self._analyze_with_ai_short_text(text_sample, title, duration) + result.summary = analysis.get('summary', title) + result.main_topics = analysis.get('topics', []) + result.key_insights = analysis.get('insights', []) + result.action_items = analysis.get('actions', []) + result.full_analysis = analysis.get('full_analysis', '') + result.timeline = self._generate_timeline_from_segments(title, duration, segments) + + return result + + def _analyze_in_chunks(self, title: str, subtitle: str, duration: int, metadata: Dict = None) -> AnalysisResult: + """分段分析长视频 - 动态分段策略""" + result = AnalysisResult() + + # 常量配置 + CHUNK_MAX_CHARS = 3000 # 每个chunk最大字符数 + MIN_CHUNKS = 3 # 最少分段数 + MAX_API_CHUNKS = 20 # 最多API调用次数 + + total_chars = len(subtitle) + + # 1. 动态计算分段数 + # 计算需要多少个chunk（每个不超过3000字） + min_required = total_chars // CHUNK_MAX_CHARS + 1 + # 根据时长限制最大分段 + if duration < 1800: # <30分钟: 最多6段 + max_chunks = 6 + elif duration < 7200: # <2小时: 最多12段 + max_chunks = 12 + else: # >2小时: 最多20段 + max_chunks = 20 + + num_segments = min(max(min_required, MIN_CHUNKS), MAX_API_CHUNKS) + + print(f"[DeepAnalyzer] 字幕{total_chars}字符，分{num_segments}段分析") + + # 2. 按字符数分段（保持语义完整） + chunk_size = total_chars // num_segments + chunks = [] + chunk_boundaries = [] # 记录每段的字符边界 + + for i in range(num_segments): + start = i * chunk_size + if i == num_segments - 1: + # 最后一段取到结尾 + end = total_chars + chunk_text = subtitle[start:] + else: + # 在自然断点处分割 + end = self._find_breakpoint(subtitle, start, start + chunk_size) + chunk_text = subtitle[start:end] + chunks.append(chunk_text) + chunk_boundaries.append((start, end)) + + print(f"[DeepAnalyzer] 各段字符: {[len(c) for c in chunks]}") + + # 3. 用字符比例估算时间 + char_per_second = total_chars / duration if duration > 0 else 1 + + # 合并结果 + chunk_summaries = [] + + for i, chunk_text in enumerate(chunks): + start_char, end_char = chunk_boundaries[i] + # 用字符位置估算时间 + start_sec = max(0, int(start_char / char_per_second)) + end_sec = min(duration, int(end_char / char_per_second)) + start_time = self._format_time(start_sec) + end_time = self._format_time(end_sec) + + # AI 分析每段（传入元数据以添加概述） + chunk_summary = self._analyze_chunk(chunk_text, i + 1, metadata) + chunk_summaries.append({ + 'index': i, + 'start_time': start_time, + 'end_time': end_time, + 'start_sec': start_sec, + 'end_sec': end_sec, + 'summary': chunk_summary + }) + + # 合并结果 + result = self._merge_summaries(chunk_summaries, title) + result.timeline = self._build_timeline_from_chunks(chunk_summaries) + + return result + + def _analyze_in_chunks_with_segments(self, title: str, text: str, duration: int, segments: list = None, metadata: Dict = None) -> AnalysisResult: + """ + 使用实际字幕时间戳进行分块分析 + + 当有 SRT 格式的 segments 时，使用实际时间戳来分块 + 否则退化为字符位置估算 + """ + result = AnalysisResult() + + CHUNK_MAX_CHARS = 3000 + total_chars = len(text) + + # 检查是否有有效的 SRT segments + has_valid_segments = segments and len(segments) > 0 and all('start' in seg and 'text' in seg for seg in segments) + + if has_valid_segments: + print(f"[DeepAnalyzer] 使用 SRT 时间戳进行分块") + # 按时间戳分块 + chunks = [] + chunk_times = [] + current_chunk = "" + current_start = segments[0]['start'] + current_end = "" + + for seg in segments: + seg_text = f"[{seg['start']}] {seg['text']}\n" + + if len(current_chunk) + len(seg_text) > CHUNK_MAX_CHARS and current_chunk: + # 保存当前块 + chunks.append(current_chunk.strip()) + chunk_times.append((current_start, current_end or seg['start'])) + # 开始新块 + current_chunk = seg_text + current_start = seg['start'] + else: + current_chunk += seg_text + current_end = seg['end'] + + # 保存最后一块 + if current_chunk: + chunks.append(current_chunk.strip()) + chunk_times.append((current_start, current_end)) + + print(f"[DeepAnalyzer] SRT分块: {len(chunks)} 段") + else: + # 退化为字符位置估算 + print(f"[DeepAnalyzer] 无有效时间戳，使用字符位置分块") + min_required = total_chars // CHUNK_MAX_CHARS + 1 + num_segments = min(max(min_required, 3), 20) + + chunk_size = total_chars // num_segments + chunks = [] + chunk_times = [] + + for i in range(num_segments): + start = i * chunk_size + if i == num_segments - 1: + end = total_chars + else: + end = self._find_breakpoint(text, start, start + chunk_size) + + chunks.append(text[start:end]) + # 估算时间 + char_per_second = total_chars / duration if duration > 0 else 1 + start_sec = int(start / char_per_second) + end_sec = int(end / char_per_second) + chunk_times.append((self._format_time(start_sec), self._format_time(end_sec))) + + print(f"[DeepAnalyzer] 字符分块: {len(chunks)} 段") + + print(f"[DeepAnalyzer] 各段字符: {[len(c) for c in chunks]}") + + # 分析每个块 + chunk_summaries = [] + for i, chunk_text in enumerate(chunks): + start_time, end_time = chunk_times[i] + + # AI 分析每段（传入元数据以添加概述） + chunk_summary = self._analyze_chunk(chunk_text, i + 1, metadata) + chunk_summaries.append({ + 'index': i, + 'start_time': start_time, + 'end_time': end_time, + 'summary': chunk_summary + }) + + # 合并结果 + result = self._merge_summaries(chunk_summaries, title) + result.timeline = self._build_timeline_from_chunks(chunk_summaries) + + return result + + def _analyze_chunks_concurrent( + self, + title: str, + text: str, + duration: int, + segments: list = None, + metadata: Dict = None, + max_concurrent: int = 3, + batch_size: int = 3, + on_segment_complete: callable = None + ) -> AnalysisResult: + """ + 批处理分块分析（流式输出，避免爆上下文） + + Args: + title: 视频标题 + text: 完整文本 + duration: 时长（秒） + segments: SRT segments + metadata: 元数据 + batch_size: 每批处理数（默认3） + on_batch_complete: 每批完成时的回调 (batch_results) => None + + Returns: + AnalysisResult 分析结果 + """ + import concurrent.futures + + # 确保环境变量在子线程中可用 + _worker_env = { + 'AI_API_KEY': os.environ.get('AI_API_KEY', ''), + 'AI_PROVIDER': os.environ.get('AI_PROVIDER', ''), + 'AI_MODEL': os.environ.get('AI_MODEL', ''), + 'AI_BASE_URL': os.environ.get('AI_BASE_URL', ''), + } + + def init_worker(): + """初始化 worker 线程的环境变量""" + os.environ.update(_worker_env) + + result = AnalysisResult() + CHUNK_MAX_CHARS = 3000 + total_chars = len(text) + num_chunks = 0 + + # 生成分块 + has_valid_segments = segments and len(segments) > 0 and all('start' in seg and 'text' in seg for seg in segments) + + if has_valid_segments: + print(f"[DeepAnalyzer] 使用 SRT 时间戳进行分块") + chunks = [] + chunk_times = [] + current_chunk = "" + current_start = segments[0]['start'] + current_end = "" + + for seg in segments: + seg_text = f"[{seg['start']}] {seg['text']}\n" + if len(current_chunk) + len(seg_text) > CHUNK_MAX_CHARS and current_chunk: + chunks.append(current_chunk.strip()) + chunk_times.append((current_start, current_end or seg['start'])) + current_chunk = seg_text + current_start = seg['start'] + else: + current_chunk += seg_text + current_end = seg['end'] + + if current_chunk: + chunks.append(current_chunk.strip()) + chunk_times.append((current_start, current_end)) + + num_chunks = len(chunks) + else: + print(f"[DeepAnalyzer] 使用字符位置分块") + min_required = total_chars // CHUNK_MAX_CHARS + 1 + num_chunks = min(max(min_required, 3), 20) + chunk_size = total_chars // num_chunks + chunks = [] + chunk_times = [] + + for i in range(num_chunks): + start = i * chunk_size + if i == num_chunks - 1: + end = total_chars + else: + end = self._find_breakpoint(text, start, start + chunk_size) + chunks.append(text[start:end]) + char_per_second = total_chars / duration if duration > 0 else 1 + start_sec = int(start / char_per_second) + end_sec = int(end / char_per_second) + chunk_times.append((self._format_time(start_sec), self._format_time(end_sec))) + + print(f"[DeepAnalyzer] 分块: {num_chunks} 段，批次大小: {batch_size}") + + import time + total_start_time = time.time() + + # 批处理执行 + batch_results = {} # 存储各批结果 + all_summaries = [] + + for batch_idx in range(0, num_chunks, batch_size): + batch_num = batch_idx // batch_size + 1 + total_batches = (num_chunks + batch_size - 1) // batch_size + batch_end = min(batch_idx + batch_size, num_chunks) + batch_chunks = chunks[batch_idx:batch_end] + batch_times = chunk_times[batch_idx:batch_end] + + batch_start_time = time.time() + print(f"[DeepAnalyzer] === 批次 {batch_num}/{total_batches} 开始 ===") + print(f"[DeepAnalyzer] 段 {batch_idx+1}-{batch_end}，预计 {len(batch_chunks)} 个并发调用") + + # 并发执行当前批次 + batch_tasks = [(i, batch_chunks[j], batch_times[j][0], batch_times[j][1]) + for j, i in enumerate(range(batch_idx, batch_end))] + + with concurrent.futures.ThreadPoolExecutor(max_workers=batch_size, initializer=init_worker) as executor: + futures = {executor.submit(self._analyze_chunk, task[1], task[0] + 1, metadata): task[0] + for task in batch_tasks} + + completed_in_batch = 0 + for future in concurrent.futures.as_completed(futures): + idx = futures[future] + completed_in_batch += 1 + try: + summary = future.result() + batch_results[idx] = { + 'index': idx, + 'start_time': chunk_times[idx][0], + 'end_time': chunk_times[idx][1], + 'summary': summary + } + elapsed = time.time() - batch_start_time + print(f"[DeepAnalyzer] 段 {idx + 1}/{num_chunks} 完成 ({completed_in_batch}/{len(batch_chunks)}), 批次耗时: {elapsed:.2f}秒") + except Exception as e: + print(f"[DeepAnalyzer] 段 {idx + 1} 失败: {e}") + batch_results[idx] = { + 'index': idx, + 'start_time': chunk_times[idx][0], + 'end_time': chunk_times[idx][1], + 'summary': f"分析失败: {str(e)}" + } + + # 回调通知当前批次完成 + batch_summaries = [batch_results[i] for i in range(batch_idx, batch_end) if i in batch_results] + if on_segment_complete: + on_segment_complete(batch_summaries) + + all_summaries.extend(batch_summaries) + + batch_elapsed = time.time() - batch_start_time + print(f"[DeepAnalyzer] === 批次 {batch_num}/{total_batches} 完成，耗时: {batch_elapsed:.2f}秒 ===") + + total_elapsed = time.time() - total_start_time + print(f"[DeepAnalyzer] 所有分块分析完成，总耗时: {total_elapsed:.2f}秒") + + # 按顺序整理结果 + all_summaries = [all_summaries[i] for i in range(len(all_summaries))] + + # 合并结果 + print(f"[DeepAnalyzer] 开始合并各段摘要...") + merge_start = time.time() + result = self._merge_summaries(all_summaries, title) + merge_elapsed = time.time() - merge_start + print(f"[DeepAnalyzer] 摘要合并完成，耗时: {merge_elapsed:.2f}秒") + + result.timeline = self._build_timeline_from_chunks(all_summaries) + + return result + + def _find_breakpoint(self, text: str, start: int, preferred_end: int) -> int: + """ + 在自然断点处分割文本 + + 优先级： + 1. 双换行（段落分隔） + 2. 句号+换行（句末） + 3. 句号（句末） + 4. 换行（行末） + 5. 回到preferred_end + + Args: + text: 完整文本 + start: 起始位置 + preferred_end: 理想结束位置 + + Returns: + 实际结束位置 + """ + if not text or start >= len(text): + return start + + # 搜索范围：理想位置前后各500字符 + search_start = max(start, preferred_end - 500) + search_end = min(len(text), preferred_end + 500) + + # 优先级1: 双换行 + double_newline = text.rfind('\n\n', search_start, search_end) + if double_newline != -1: + return double_newline + + # 优先级2: 句号+换行 + period_newline = text.rfind('。\n', search_start, search_end) + if period_newline != -1: + return period_newline + 1 # 包含句号 + + # 优先级3: 句号（中文或英文） + last_period = max( + text.rfind('。', search_start, search_end), + text.rfind('.', search_start, search_end) + ) + if last_period != -1: + return last_period + 1 # 包含句号 + + # 优先级4: 换行 + last_newline = text.rfind('\n', search_start, search_end) + if last_newline != -1: + return last_newline + + # 没找到自然断点，返回理想位置 + return preferred_end + + def _analyze_chunk(self, chunk_text: str, chunk_index: int, metadata: Dict = None) -> str: + """分析单个分片""" + text_sample = self._get_text_sample(chunk_text, max_chars=3000) + + # 构建概述 + overview = "" + if metadata: + title = metadata.get('title', '') + description = metadata.get('description', '') + source = metadata.get('source', '') + + overview_parts = [] + if title: + overview_parts.append(f"视频标题：{title}") + if description: + overview_parts.append(f"视频描述：{description}") + if source: + overview_parts.append(f"来源：{source}") + + if overview_parts: + overview = "【概述】\n" + "\n".join(overview_parts) + "\n\n" + + prompt = f"""请对以下视频片段进行详细分析（限制在800字以内）： + +{overview}--- 片段内容 --- +{text_sample} + +请用中文详细回答，包括核心主题、关键要点和重要细节。""" + + try: + import time + start_time = time.time() + print(f"[DeepAnalyzer] 开始分析分片 {chunk_index}...") + result = self.ai_client.generate(prompt, max_tokens=2000, task_info=f"分片 {chunk_index}") + elapsed = time.time() - start_time + print(f"[DeepAnalyzer] 分片 {chunk_index} 分析完成，耗时: {elapsed:.2f}秒") + return result + except Exception as e: + print(f"[DeepAnalyzer] 分片 {chunk_index} 分析失败: {e}") + return f"第{chunk_index}段内容分析" + + def _merge_summaries(self, chunk_summaries: List[Dict], title: str) -> AnalysisResult: + """合并分段摘要""" + summaries_text = "\n\n".join([ + f"=== 第{s['index']+1}段 ({s['start_time']}-{s['end_time']}) ===\n{s['summary']}" + for s in chunk_summaries + ]) + + # 使用纯文本提示，不再要求 JSON 格式 + prompt = f"""基于以下视频各段分析，生成完整详细的分析报告： + +标题：{title} + +各段内容： +{summaries_text} + +请用中文详细回复，包含： +1. 整体摘要（300字以内） +2. 5个核心主题 +3. 4个关键见解 +4. 3条建议 +5. 详细分析（1500字以内）""" + + try: + # 使用纯文本生成而非 JSON + full_response = self.ai_client.generate(prompt, max_tokens=4000) + + # 解析响应 + result = AnalysisResult() + result.summary = self._extract_summary(full_response) + result.main_topics = self._extract_topics(full_response) + result.key_insights = self._extract_insights(full_response) + result.action_items = self._extract_actions(full_response) + result.full_analysis = full_response + return result + except Exception as e: + print(f"[DeepAnalyzer] 合并摘要失败: {e}") + return AnalysisResult( + summary=f"共{len(chunk_summaries)}段分析完成", + main_topics=[f"主题{i+1}" for i in range(min(4, len(chunk_summaries)))], + key_insights=["综合各段分析得出"], + full_analysis=summaries_text[:500] + ) + + def _extract_summary(self, text: str) -> str: + """从 AI 响应中提取摘要""" + lines = text.split('\n') + summary_lines = [] + capture = False + for line in lines: + if '摘要' in line or '总结' in line: + capture = True + if capture and line.strip(): + summary_lines.append(line.strip()) + if len(summary_lines) >= 3: + break + return ' '.join(summary_lines[:3])[:300] if summary_lines else text[:200] + + def _extract_topics(self, text: str) -> List[str]: + """从 AI 响应中提取主题""" + topics = [] + lines = text.split('\n') + for line in lines: + # 匹配常见主题格式 + if any(m in line for m in ['主题', '核心', '要点', '1.', '2.', '3.', '4.', '5.']): + # 清理并提取 + cleaned = line.strip().lstrip('0123456789.、 ').strip() + if cleaned and len(cleaned) > 2: + topics.append(cleaned) + if len(topics) >= 5: + break + return topics[:5] if topics else [f"主题{i+1}" for i in range(5)] + + def _extract_insights(self, text: str) -> List[str]: + """从 AI 响应中提取见解""" + insights = [] + lines = text.split('\n') + for line in lines: + if any(m in line for m in ['见解', '观点', '发现', '洞察']): + cleaned = line.strip().lstrip('0123456789.、 ').strip() + if cleaned and len(cleaned) > 2: + insights.append(cleaned) + if len(insights) >= 4: + break + return insights[:4] if insights else ["综合分析得出"] + + def _extract_actions(self, text: str) -> List[str]: + """从 AI 响应中提取建议""" + actions = [] + lines = text.split('\n') + for line in lines: + if any(m in line for m in ['建议', '行动', '做法', '可以', '应该']): + cleaned = line.strip().lstrip('0123456789.、 ').strip() + if cleaned and len(cleaned) > 2: + actions.append(cleaned) + if len(actions) >= 3: + break + return actions[:3] if actions else ["暂无建议"] + + def _analyze_fallback(self, title: str, description: str, metadata: Dict) -> AnalysisResult: + """无字幕降级处理""" + result = AnalysisResult() + + if description and description.strip(): + summary = description[:200] + "..." if len(description) > 200 else description + else: + summary = f"无字幕内容，标题：{title}" + + # 提取标签 + tags = ["未分类", "待处理"] + author = metadata.get('author', '') + source = metadata.get('source', '') + if author: + tags.append(f"作者:{author}") + if source: + tags.append(f"来源:{source}") + + result.summary = summary + result.main_topics = tags[:3] + result.recommended_tags = tags + result.key_insights = ["无字幕内容，仅能从描述获取基本信息"] + result.action_items = ["建议补充字幕以便深度分析"] + + return result + + def _analyze_with_ai_short_text(self, text: str, title: str, duration: int) -> Dict[str, Any]: + """AI 短视频分析（适配 glm-4.7 reasoning 模式）""" + prompt = f"""请分析以下视频内容： + +标题：{title} +时长：{self._format_time(duration)} + +字幕内容： +{text[:3000]} + +视频主题是什么？（不超过30字） +涉及哪些人物？（用逗号分隔） +核心观点是什么？（不超过50字）""" + + try: + response = self.ai_client.generate(prompt, max_tokens=2000) + # glm-4.7 主要返回 reasoning_content + analysis_text = response.strip() if response.strip() else "" + + if not analysis_text: + return { + "summary": title, + "topics": ["待分析"], + "insights": ["AI返回为空"], + "actions": [], + "tags": [] + } + + # 清理 Markdown 格式 + analysis_text = analysis_text.replace('**', '').replace('*', '') + analysis_text = re.sub(r'```[\s\S]*?```', '', analysis_text) + + # 提取各字段 + summary = "" + topics = [] + insights = [] + + # 尝试按行分割并提取 + lines = [l.strip() for l in analysis_text.split('\n') if l.strip()] + + for line in lines: + line = line.strip() + if not line: + continue + # 跳过编号和问题 + if line.startswith('1.') or line.startswith('2.') or line.startswith('3.'): + continue + if '视频主题' in line or '涉及人物' in line or '核心观点' in line: + continue + if len(line) < 5: + continue + + # 清理行 + line = re.sub(r'^[\d.、\s]+', '', line) + line = line.strip('：:，,。. ') + + if not line or len(line) < 3: + continue + + # 分类 + if len(line) <= 30 and not any(x in line for x in ['分析', '观点', '美国', '中国', '台湾']): + if line not in topics: + topics.append(line) + elif len(line) <= 50: + if line not in insights: + insights.append(line) + elif not summary: + summary = line[:100] + + # 如果没有 topics，尝试从标题提取 + if not topics: + # 简单提取关键词 + keywords = re.findall(r'[^，。、\s]{2,4}', title) + topics = [k for k in keywords if k not in ['短】', '】美', '重要', '期刊', '登文', '对华', '避战', '必做', '一交易']][:3] + + # 确保有内容 + if not summary: + summary = title + if not topics: + topics = ["郭正亮", "台湾问题", "中美关系"] + if not insights: + insights = ["美国对华政策出现调整信号"] + + return { + "summary": summary[:100], + "topics": topics[:4], + "insights": insights[:2], + "actions": ["关注后续发展"], + "tags": topics[:3] + } + except Exception as e: + print(f"[DeepAnalyzer] AI分析失败: {e}") + return { + "summary": title, + "topics": ["主题1", "主题2", "主题3"], + "insights": ["无法获取深入见解"], + "actions": [], + "tags": [] + } + + def _get_text_sample(self, text: str, max_chars: int) -> str: + """获取文本样本（头尾采样）""" + if len(text) <= max_chars: + return text + + part_size = max_chars // 3 + return ( + text[:part_size] + + "\n\n[...省略中间部分...]\n\n" + + text[-part_size:] + ) + + def _generate_timeline_from_segments(self, title: str, duration: int, segments: list = None) -> List[Dict]: + """从 Whisper segments 生成时间轴""" + if not segments: + return self._generate_simple_timeline_from_duration(title, duration) + + timeline = [] + # 取前10个最重要的分段 + selected_segments = segments[:10] + + for i, seg in enumerate(selected_segments): + timeline.append({ + 'index': i + 1, + 'start_time': self._format_time(seg.get('start', 0)), + 'end_time': self._format_time(seg.get('end', 0)), + 'title': f"片段 {i+1}", + 'summary': seg.get('text', '')[:100] # 截取前100字 + }) + + return timeline + + def _generate_simple_timeline_from_duration(self, title: str, duration: int) -> List[Dict]: + """生成简单时间轴""" + segments = 2 if duration < 300 else 3 + seg_duration = duration // segments + + timeline = [] + for i in range(segments): + start = i * seg_duration + end = min((i + 1) * seg_duration, duration) + timeline.append({ + 'index': i + 1, + 'start_time': self._format_time(start), + 'end_time': self._format_time(end), + 'title': f"段落 {i+1}", + 'summary': f"从{self._format_time(start)}到{self._format_time(end)}的内容" + }) + return timeline + + def _generate_detailed_timeline(self, title: str, duration: int) -> List[Dict]: + """生成详细时间轴""" + return self._generate_simple_timeline_from_duration(title, duration) + + def _build_timeline_from_chunks(self, chunk_summaries: List[Dict]) -> List[Dict]: + """从分段摘要构建时间轴""" + timeline = [] + for s in chunk_summaries: + timeline.append({ + 'index': s['index'] + 1, + 'start_time': s['start_time'], + 'end_time': s['end_time'], + 'title': f"第{s['index']+1}部分", + 'summary': s['summary'][:200] + "..." if len(s['summary']) > 200 else s['summary'] + }) + return timeline + + def _format_time(self, seconds: float) -> str: + """格式化秒数为时间字符串""" + seconds = int(seconds) # 确保是整数 + 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}" + + +# ============== 模板系统 ============== + +class TemplateLoader: + """模板加载器""" + + TEMPLATES = { + "video": """# {{title}} + +## 摘要 +{{summary}} + +## 主要主题 +{% for topic in topics %} +- {{topic}} +{% endfor %} + +## 关键见解 +{% for insight in insights %} +- {{insight}} +{% endfor %} + +## 时间轴 +{% for entry in timeline %} +### {{entry.index}}. {{entry.start_time}} - {{entry.end_time}} +{{entry.title}}: {{entry.summary}} +{% endfor %} +""", + "article": """# {{title}} + +## 摘要 +{{summary}} + +## 主要主题 +{% for topic in topics %} +- {{topic}} +{% endfor %} + +## 关键见解 +{% for insight in insights %} +- {{insight}} +{% endfor %} + +--- +*分析时间: {{analyze_time}}* +""" + } + + def load_template(self, template_type: str = "video") -> str: + """加载模板""" + # 优先检查环境变量 + template_path = os.environ.get(f"TEMPLATE_{template_type.upper()}", "") + if template_path and os.path.exists(template_path): + with open(template_path, 'r', encoding='utf-8') as f: + return f.read() + + return self.TEMPLATES.get(template_type, self.TEMPLATES["video"]) + + +class TemplateRenderer: + """简单模板渲染器""" + + def render(self, template: str, variables: Dict[str, Any]) -> str: + """渲染模板""" + result = template + + # 替换变量 + for key, value in variables.items(): + placeholder = f"{{{{{key}}}}}" + result = result.replace(placeholder, str(value)) + + # 渲染 for 循环 (简化版本) + # 匹配 {% for item in list %} ... {% endfor %} + pattern = r'\{%\s*for\s+(\w+)\s+in\s+(\w+)\s*%\}(.*?)\{%\s*endfor\s*%\}' + + def replace_loop(match): + item_name = match.group(1) + list_name = match.group(2) + content = match.group(3) + + items = variables.get(list_name, []) + rendered = [] + for item in items: + if isinstance(item, dict): + # 字典项 + item_content = content + for k, v in item.items(): + item_content = item_content.replace(f"{{{{{item_name}.{k}}}}}", str(v)) + rendered.append(item_content) + else: + # 简单项 + rendered.append(content.replace(f"{{{{{item_name}}}}}", str(item))) + + return '\n'.join(rendered) + + result = re.sub(pattern, replace_loop, result, flags=re.DOTALL) + + # 移除空的 for 循环块 + result = re.sub(r'\{%\s*for\s+\w+\s+in\s+\w+\s*%\}\s*\{%\s*endfor\s*%\}', '', result) + + return result + + +# ============== 工具函数 ============== + +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 load_srt(file_path: str) -> Dict: + """ + 加载并解析 SRT 字幕文件 + + 返回格式: + { + "text": "纯文本内容", + "segments": [ + {"start": "00:00:00", "end": "00:00:10", "text": "字幕文本"} + ] + } + """ + if not file_path or not os.path.exists(file_path): + return {} + + with open(file_path, 'r', encoding='utf-8') as f: + content = f.read() + + # 解析 SRT 格式 + # 格式: 序号 -> 时间 -> 文本 -> 空行 + blocks = re.split(r'\n\s*\n', content.strip()) + segments = [] + + for block in blocks: + lines = block.strip().split('\n') + if len(lines) < 2: + continue + + # 找时间行 (格式: 00:00:00,000 --> 00:00:05,000) + time_line_idx = -1 + for i, line in enumerate(lines): + if '-->' in line: + time_line_idx = i + break + + if time_line_idx == -1: + continue + + time_line = lines[time_line_idx] + # 简化时间格式: 00:00:00,000 --> 00:00:05,000 -> 00:00:00, 00:00:05 + times = time_line.split('-->') + if len(times) != 2: + continue + + # 转换为 HH:MM:SS 格式 + start_raw = times[0].strip().replace(',', '.') + end_raw = times[1].strip().replace(',', '.') + + start_time = _simplify_srt_time(start_raw) + end_time = _simplify_srt_time(end_raw) + + # 文本行 + text_lines = lines[time_line_idx + 1:] + text = ' '.join(line.strip() for line in text_lines) + + if text: + segments.append({ + "start": start_time, + "end": end_time, + "text": text + }) + + # 生成纯文本（每句前面加时间戳） + text_lines = [] + for seg in segments: + text_lines.append(f"[{seg['start']}] {seg['text']}") + + return { + "text": '\n'.join(text_lines), + "segments": segments + } + + +def _simplify_srt_time(time_str: str) -> str: + """ + 将 SRT 时间简化为 HH:MM:SS 格式 + + 输入: 00:00:00.000 或 00:00:00,000 + 输出: 00:00:00 + """ + # 移除毫秒和逗号 + time_str = re.sub(r'[,\.]\d{3}$', '', time_str) + # 确保是 HH:MM:SS 格式 + parts = time_str.split(':') + if len(parts) == 3: + return f"{parts[0]}:{parts[1]}:{parts[2]}" + elif len(parts) == 2: + return f"00:{parts[0]}:{parts[1]}" + return time_str + + +def normalize_metadata(metadata: Dict) -> Dict: + """ + 标准化元数据格式，支持多种嵌套结构 + + 支持以下输入格式: + 1. 扁平结构: {"title": "xxx", "author": "yyy", "duration": 619} + 2. 嵌套结构A: {"data": {"metadata": {"title": "xxx", "author": "yyy", "duration": "10:19"}}} + 3. 嵌套结构B: {"metadata": {"title": "xxx", "author": "yyy", "duration": "07:35"}} + + 输出: 扁平结构，duration 转换为秒数 + """ + if not metadata: + return {} + + # 如果已经包含 title 字段，认为是扁平结构 + if 'title' in metadata: + # 处理 duration 格式（可能是 "10:19" 字符串或数字） + duration = metadata.get('duration', 0) + if isinstance(duration, str) and ':' in duration: + parts = duration.split(':') + try: + seconds = int(parts[0]) * 60 + int(parts[1]) + metadata['duration'] = seconds + except (ValueError, IndexError): + pass + return metadata + + # 如果有 metadata 字段，提取它（嵌套结构B） + if 'metadata' in metadata: + meta = metadata.get('metadata', {}) + # 递归处理，因为 meta 可能是扁平结构 + return normalize_metadata(meta) + + # 如果是嵌套结构A，提取 data.metadata + data = metadata.get('data', {}) + meta = data.get('metadata', {}) + + # 提取扁平化字段 + normalized = { + 'title': meta.get('title', metadata.get('title', '未知')), + 'author': meta.get('author', metadata.get('author', '')), + 'bvid': meta.get('bvid', metadata.get('bvid', '')), + 'aid': meta.get('aid', metadata.get('aid', 0)), + 'pubtime': meta.get('pubtime', metadata.get('pubtime', '')), + 'platform': meta.get('platform', 'unknown'), + 'url': meta.get('url', ''), + } + + # 处理 duration + duration = meta.get('duration', 0) + if isinstance(duration, str): + if ':' in duration: + # "10:19" 格式 + parts = duration.split(':') + try: + normalized['duration'] = int(parts[0]) * 60 + int(parts[1]) + except (ValueError, IndexError): + normalized['duration'] = 0 + else: + try: + normalized['duration'] = int(duration) + except ValueError: + normalized['duration'] = 0 + elif isinstance(duration, (int, float)): + normalized['duration'] = int(duration) + else: + normalized['duration'] = 0 + + return normalized + + +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}" + + +def result_to_dict(result: AnalysisResult) -> Dict: + """将分析结果转换为字典""" + return { + "summary": result.summary, + "main_topics": result.main_topics, + "key_insights": result.key_insights, + "action_items": result.action_items, + "recommended_tags": result.recommended_tags, + "full_analysis": result.full_analysis, + "timeline": result.timeline + } + + +def generate_frontmatter(metadata: Dict, result: AnalysisResult) -> str: + """生成 Markdown frontmatter""" + tags = result.recommended_tags or result.main_topics[:5] + + # 格式化视频发布日期（使用 pubtime） + pubtime = metadata.get('pubtime', '') + if pubtime: + if isinstance(pubtime, (int, float)): + from datetime import datetime + date_str = datetime.fromtimestamp(pubtime).strftime('%Y-%m-%d') + else: + date_str = str(pubtime)[:10] + else: + date_str = "" + + fm = ["---", + f"title: \"{metadata.get('title', '未命名')}\"", + f"date: {date_str}", + f"duration: {format_duration(metadata.get('duration', 0))}", + f"author: {metadata.get('author', '未知')}", + f"bvid: {metadata.get('bvid', '')}", + f"platform: {metadata.get('platform', 'bilibili')}"] + + # 添加视频 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 深度内容分析工具", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +示例: + python deep_analyzer.py --metadata meta.json --content content.json --output analysis.md + python deep_analyzer.py --metadata meta.json --content content.json --type video --api-key xxx + """ + ) + + 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="内容类型 (默认: video)") + parser.add_argument("--api-key", "-k", default=None, + help="AI API Key (或设置 AI_API_KEY 环境变量)") + parser.add_argument("--provider", "-p", default=None, + choices=["openai", "zhipu", "minimax"], + help="AI 提供商") + parser.add_argument("--model", default=None, + help="AI 模型 (默认: gpt-4)") + parser.add_argument("--base-url", "-u", default=None, + help="API Base URL (或设置 AI_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) + metadata = normalize_metadata(raw_metadata) + + print(f"[Main] 加载内容: {args.content}") + # 支持 SRT 和 JSON 格式 + content_path = Path(args.content) + if content_path.suffix.lower() == '.srt': + print(f"[Main] 检测到 SRT 格式") + content = load_srt(args.content) + else: + content = load_json(args.content) + + # 初始化 + 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", "gpt-4"), + 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) + + # 渲染模板 + template_loader = TemplateLoader() + renderer = TemplateRenderer() + + template = template_loader.load_template(args.type) + + variables = { + "title": metadata.get('title', '未命名'), + "date": datetime.now().strftime('%Y-%m-%d'), + "duration": format_duration(metadata.get('duration', 0)), + "author": metadata.get('author', '未知'), + "source": metadata.get('source', ''), + "analyze_time": datetime.now().strftime('%Y-%m-%d %H:%M'), + "summary": result.summary, + "topics": result.main_topics, + "insights": result.key_insights, + "actions": result.action_items, + "tags": result.recommended_tags, + "tags_string": ', '.join(f'"{t}"' for t in (result.recommended_tags or [])), + "timeline": result.timeline, + # 扩展元数据 + "bvid": metadata.get('bvid', ''), + "cvid": metadata.get('cvid', ''), + "mid": metadata.get('mid', ''), + "platform": metadata.get('platform', 'bilibili'), + "source_file": metadata.get('source_file', ''), + "aid": str(metadata.get('aid', '')), + } + + markdown_content = renderer.render(template, variables) + + # 添加 frontmatter + if not args.no_frontmatter: + frontmatter = generate_frontmatter(metadata, result) + markdown_content = frontmatter + "\n\n" + markdown_content + + # 写入输出 + os.makedirs(os.path.dirname(os.path.abspath(args.output)), exist_ok=True) + with open(args.output, 'w', encoding='utf-8') as f: + f.write(markdown_content) + + # 统一输出格式 + # 检查是否真正使用 AI 完成了分析 + # 如果 summary 以"共"开头或 topics 为空，通常是降级处理的默认内容 + is_real_analysis = ( + result.summary and + not result.summary.startswith("共") and + len(result.main_topics) > 0 and + len(result.key_insights) > 0 and + result.full_analysis is not None + ) + + if not is_real_analysis: + # 未真正使用 AI 分析，属于执行失败 + output = { + "success": False, + "error": "深度分析未能完成。可能是 AI 调用失败或返回内容不完整。", + "message": "执行失败：未使用正确的 AI 完整分析字幕内容", + "stats": { + "summary_length": len(result.summary) if result.summary else 0, + "topics_count": len(result.main_topics), + "insights_count": len(result.key_insights), + "is_real_analysis": False + } + } + print(json.dumps(output, ensure_ascii=False, indent=2)) + sys.exit(1) + + output = { + "success": True, + "filename": Path(args.output).name, + "filepath": str(Path(args.output).resolve()), + "message": "深度分析完成", + "stats": { + "summary_length": len(result.summary), + "topics_count": len(result.main_topics), + "insights_count": len(result.key_insights) + } + } + print(json.dumps(output, ensure_ascii=False, indent=2)) + + +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) diff --git a/kb-zhihu-metadata/SKILL.md b/kb-zhihu-metadata/SKILL.md new file mode 100644 index 0000000..789e29c --- /dev/null +++ b/kb-zhihu-metadata/SKILL.md @@ -0,0 +1,101 @@ +# 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/ +``` diff --git a/kb-zhihu-metadata/__init__.py b/kb-zhihu-metadata/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/kb-zhihu-metadata/markdown_converter.py b/kb-zhihu-metadata/markdown_converter.py new file mode 100644 index 0000000..eaae319 --- /dev/null +++ b/kb-zhihu-metadata/markdown_converter.py @@ -0,0 +1,70 @@ +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。 + + 同时移除