diff --git a/memory/2026-02-10.md b/memory/2026-02-10.md index 8598568..504a40f 100644 --- a/memory/2026-02-10.md +++ b/memory/2026-02-10.md @@ -1,23 +1,109 @@ # 2026-02-10 -## 知识库技能重构 - B站专栏内容获取 +## 知识库技能重构 - 完整功能 -### kb-bilibili-content 专栏内容获取 +### kb-bilibili-content 专栏内容获取 ✅ -**测试结果：** -``` -标题: [UEPCG][新手向]复刻ChrisMurphy的城市道路建筑生成... -内容长度: 12659 字符 -✅ 获取成功 -``` - -**实现方式：** - 使用 Playwright 异步API爬取专栏页面 - 选择器: `.read-article-holder` - 获取完整文章正文 -### 已提交 +### Git版本管理和清理 ✅ (2026-02-10下午) -- `47355b3` - feat(kb-bilibili-content): Add article content extraction using Playwright -- `56907e3` - refactor(kb-zhihu-metadata): Simplify to only support zvideo and zhuanlan -- 使用playurl API，无需Cookie即可下载视频 +**需求1: Git版本管理** +- Git仓库: `archive/` 目录 +- 配置环境变量: + - `KB_GIT_ROOT` - Git仓库目录 + - `KB_GIT_REMOTE` - 远程仓库地址 + - `KB_GIT_USER` - 用户名 + - `KB_GIT_EMAIL` - 邮箱 +- 命令: `--trigger git-init`, `--trigger git-status` +- 推送重试: 3次 + +**需求2: 订阅数据管理** +- 存储位置: `archive/subscriptions/` +- 文件: `subscriptions.json`, `processed.json` +- Git跟踪历史，防止误删 + +**需求3: 自动清理策略** +| 类型 | 保留 | 清理周期 | +|------|------|---------| +| 分析报告 | ✅ | 永久 | +| 链接信息 | ✅ | 永久 | +| 元数据 | ✅ | 永久 | +| 订阅数据 | ✅ | Git管理 | +| 视频文件 | ❌ | 7天 | +| 截帧文件 | ❌ | 7天 | + +**命令:** +```bash +# 查看状态 +python kb_workflow.py --trigger status + +# 处理待处理内容 +python kb_workflow.py --trigger manual --scope pending --limit 5 + +# 初始化Git +python kb_workflow.py --trigger git-init + +# 清理过期文件 +python kb_workflow.py --trigger cleanup --days 30 +``` + +### 深度分析模板增强 ✅ + +- 完整元数据保留 (bvid, cvid, mid, platform) +- AI生成标签 (5个) +- 模板支持环境变量覆盖 + +--- + +## 待处理事项 + +### 重要 +- [ ] 配置 OpenRouter API +- [ ] 配置阿里云 TTS +- [ ] 🔒 安全巡检报告 + +### 字幕增强（待实现） +- [ ] 视觉大模型方案 (GPT-4V/Claude/GLM-4V) +- [ ] 语音识别方案 (Whisper/ASR) + +--- + +## 已提交Commit + +| Commit | 说明 | +|--------|------| +| `38f1624` | docs: Update TODO | +| `f41fb61` | feat(kb-workflow): Add Git version control and cleanup | +| `8b43416` | feat(kb-deep-analyzer): Enhance template and metadata | +| `f2309bf` | feat(kb-workflow): Complete file storage | +| `3f74bcd` | docs: Add subtitle enhancement tasks | +| `47355b3` | feat(kb-bilibili-content): Article extraction | +| `56907e3` | refactor(kb-zhihu-metadata): Simplify | + +--- + +## 测试要点 + +1. **kb-link-parser** - 链接解析 ✅ +2. **kb-bilibili-metadata** - B站元数据 ✅ +3. **kb-zhihu-metadata** - 知乎元数据 (zvideo/zhuanlan) ⚠️ +4. **kb-bilibili-content** - 内容获取 ✅ +5. **kb-ocr-processor** - OCR识别 ✅ +6. **kb-deep-analyzer** - 深度分析 ✅ (mock模式) +7. **kb-workflow** - 工作流编排 ✅ + +### Git配置 +```bash +export KB_GIT_ROOT="/path/to/archive" +export KB_GIT_REMOTE="git@github.com:user/repo.git" +export KB_GIT_USER="Username" +export KB_GIT_EMAIL="user@example.com" +``` + +### 清理命令 +```bash +python kb_workflow.py --trigger cleanup --days 30 +``` diff --git a/skills/kb-deep-analyzer/deep_analyzer.py b/skills/kb-deep-analyzer/deep_analyzer.py index 41f871d..8697eb6 100644 --- a/skills/kb-deep-analyzer/deep_analyzer.py +++ b/skills/kb-deep-analyzer/deep_analyzer.py @@ -48,22 +48,22 @@ class AnalysisResult: class SimpleAIClient: """简化的 AI 客户端 - 支持多提供商 fallback""" - + # 时长阈值（秒） SHORT_VIDEO_THRESHOLD = 600 # 10分钟 LONG_VIDEO_THRESHOLD = 1800 # 30分钟 - + def __init__(self, api_key: str = None, provider: str = "mock", model: str = "gpt-4"): self.api_key = api_key or os.environ.get("AI_API_KEY", "") self.provider = provider self.model = model self.base_url = os.environ.get("AI_BASE_URL", "") - + def generate(self, prompt: str, max_tokens: int = 1000) -> str: """生成文本""" if not self.api_key or self.provider == "mock": return self._mock_generate(prompt, max_tokens) - + try: if self.provider == "openai": return self._call_openai(prompt, max_tokens) @@ -76,7 +76,7 @@ class SimpleAIClient: except Exception as e: print(f"[AI Client] 调用失败: {e}") return self._mock_generate(prompt, max_tokens) - + def generate_json(self, prompt: str) -> Dict[str, Any]: """生成 JSON 格式响应""" response = self.generate(prompt, max_tokens=2000) @@ -85,59 +85,60 @@ class SimpleAIClient: except json.JSONDecodeError: # 尝试从文本中提取 JSON return self._parse_json_from_text(response) - + def _call_openai(self, prompt: str, max_tokens: int) -> 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=30) response.raise_for_status() return response.json()['choices'][0]['message']['content'] - + def _call_zhipu(self, prompt: str, max_tokens: int) -> str: """调用智谱 AI API""" import requests - base_url = self.base_url or "https://open.bigmodel.cn/api/paas/v4" + 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}] + "messages": [{"role": "user", "content": prompt}], + "max_tokens": max_tokens } - + response = requests.post(url, headers=headers, json=payload, timeout=30) response.raise_for_status() return response.json()['choices'][0]['message']['content'] - + def _call_minimax(self, prompt: str, max_tokens: int) -> 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=30) response.raise_for_status() data = response.json() @@ -145,11 +146,11 @@ class SimpleAIClient: if item.get('type') == 'text': return item.get('text', '') return str(data) - + def _mock_generate(self, prompt: str, max_tokens: int) -> str: """模拟 AI 生成""" prompt_lower = prompt.lower() - + if 'json' in prompt_lower or '请返回' in prompt: # 返回模拟 JSON 响应 if '主题' in prompt or 'topic' in prompt_lower: @@ -167,12 +168,12 @@ class SimpleAIClient: if '标签' in prompt or 'tag' in prompt_lower: return json.dumps({'tags': ['技术', 'AI', '视频分析', '深度学习']}, ensure_ascii=False) return json.dumps({'result': '模拟响应'}, ensure_ascii=False) - + if '摘要' in prompt or 'summary' in prompt_lower: return '这是一段关于内容深度分析的重要讨论，涵盖了多个关键主题和见解。' - + return '模拟AI响应内容' - + def _parse_json_from_text(self, text: str) -> Dict[str, Any]: """从文本中解析 JSON""" json_match = re.search(r'\{[\s\S]*\}', text) @@ -188,23 +189,23 @@ class SimpleAIClient: 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], + + def analyze(self, metadata: Dict[str, Any], content: Dict[str, Any], content_type: str = "video") -> AnalysisResult: """ 分析内容 - + Args: metadata: 元数据字典 content: 内容数据字典（包含 subtitle, description 等） content_type: 内容类型 - + Returns: AnalysisResult 分析结果 """ @@ -212,13 +213,13 @@ class DeepAnalyzer: duration = metadata.get('duration', 0) subtitle = content.get('subtitle', '') description = content.get('description', '') - + print(f"[DeepAnalyzer] 开始分析: {title}") print(f"[DeepAnalyzer] 类型: {content_type}, 时长: {duration}秒") print(f"[DeepAnalyzer] 字幕存在: {bool(subtitle and subtitle.strip())}") - + has_subtitle = bool(subtitle and subtitle.strip()) - + if has_subtitle: if duration < self.SHORT_VIDEO_THRESHOLD: print(f"[DeepAnalyzer] 使用轻量分析策略 (时长 < 10分钟)") @@ -229,28 +230,28 @@ class DeepAnalyzer: else: print(f"[DeepAnalyzer] 降级处理（无字幕）") result = self._analyze_fallback(title, description, metadata) - + return result - + def _analyze_short_with_subtitle(self, title: str, subtitle: str, duration: int) -> AnalysisResult: """有字幕的短视频轻量分析""" result = AnalysisResult() text_sample = self._get_text_sample(subtitle, 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_simple_timeline_from_duration(title, duration) - + return result - - def _analyze_long_with_subtitle(self, title: str, subtitle: str, duration: int, + + def _analyze_long_with_subtitle(self, title: str, subtitle: str, duration: int, content_type: str) -> AnalysisResult: """有字幕的长视频详细分析""" result = AnalysisResult() - + if duration >= self.LONG_VIDEO_THRESHOLD and content_type == "video": # 使用分段分析 result = self._analyze_in_chunks(title, subtitle, duration) @@ -264,13 +265,13 @@ class DeepAnalyzer: result.action_items = analysis.get('actions', []) result.full_analysis = analysis.get('full_analysis', '') result.timeline = self._generate_detailed_timeline(title, duration) - + return result - + def _analyze_in_chunks(self, title: str, subtitle: str, duration: int) -> AnalysisResult: """分段分析长视频""" result = AnalysisResult() - + # 确定分段数 if duration < 3600: # <60分钟 num_segments = 3 @@ -278,20 +279,20 @@ class DeepAnalyzer: num_segments = 4 else: # >120分钟 num_segments = 6 - + # 生成分段摘要 segment_duration = duration // num_segments chunk_summaries = [] - + # 简单按字符数分割字幕 subtitle_chunks = self._chunk_subtitle(subtitle, num_segments) - + for i, chunk_text in enumerate(subtitle_chunks): start_sec = i * segment_duration end_sec = min((i + 1) * segment_duration, duration) start_time = self._format_time(start_sec) end_time = self._format_time(end_sec) - + # AI 分析每段 chunk_summary = self._analyze_chunk(chunk_text, i + 1) chunk_summaries.append({ @@ -300,37 +301,37 @@ class DeepAnalyzer: '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_chunk(self, chunk_text: str, chunk_index: int) -> str: """分析单个分片""" text_sample = self._get_text_sample(chunk_text, max_chars=3000) - + prompt = f"""请对以下视频片段进行简要分析（限制在300字以内）： 片段内容： {text_sample} 请用中文简要回答，包括核心主题和关键要点。""" - + try: return self.ai_client.generate(prompt, max_tokens=500) 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 ]) - + prompt = f"""基于以下视频各段分析，生成完整的分析报告： 标题：{title} @@ -346,7 +347,7 @@ class DeepAnalyzer: "actions": ["建议1", "建议2"], "full_analysis": "详细分析（500字以内）" }}""" - + try: analysis = self.ai_client.generate_json(prompt) result = AnalysisResult() @@ -364,16 +365,16 @@ class DeepAnalyzer: key_insights=["综合各段分析得出"], full_analysis=summaries_text[:500] ) - + 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', '') @@ -382,15 +383,15 @@ class DeepAnalyzer: 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 短视频分析""" prompt = f"""请分析以下视频内容，以JSON格式返回： @@ -409,7 +410,7 @@ class DeepAnalyzer: "actions": ["建议1"], "tags": ["标签1", "标签2", "标签3", "标签4", "标签5"] }}""" - + try: return self.ai_client.generate_json(prompt) except Exception as e: @@ -421,39 +422,39 @@ class DeepAnalyzer: "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 _chunk_subtitle(self, subtitle: str, num_chunks: int) -> List[str]: """简单分割字幕""" if not subtitle: return [""] * num_chunks - + chunk_size = len(subtitle) // num_chunks chunks = [] - + for i in range(num_chunks): start = i * chunk_size end = start + chunk_size if i < num_chunks - 1 else len(subtitle) chunks.append(subtitle[start:end]) - + return chunks - + 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 @@ -466,11 +467,11 @@ class DeepAnalyzer: '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 = [] @@ -483,7 +484,7 @@ class DeepAnalyzer: 'summary': s['summary'][:200] + "..." if len(s['summary']) > 200 else s['summary'] }) return timeline - + def _format_time(self, seconds: int) -> str: """格式化秒数为时间字符串""" h = seconds // 3600 @@ -498,7 +499,7 @@ class DeepAnalyzer: class TemplateLoader: """模板加载器""" - + TEMPLATES = { "video": """# {{title}} @@ -609,7 +610,7 @@ platform: {{platform}} - **原始文件**: {{source_file}} """ } - + def load_template(self, template_type: str = "video") -> str: """加载模板""" # 优先检查环境变量 @@ -617,31 +618,31 @@ platform: {{platform}} 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: @@ -654,14 +655,14 @@ class TemplateRenderer: 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 @@ -671,7 +672,7 @@ 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) @@ -702,15 +703,15 @@ def result_to_dict(result: AnalysisResult) -> Dict: def generate_frontmatter(metadata: Dict, result: AnalysisResult) -> str: """生成 Markdown frontmatter""" tags = result.recommended_tags or result.main_topics[:5] - - fm = ["---", "title: \"" + metadata.get('title', '未命名') + "\"", + + fm = ["---", "title: \"" + metadata.get('title', '未命名') + "\"", f"date: {datetime.now().strftime('%Y-%m-%d')}", f"duration: {format_duration(metadata.get('duration', 0))}", f"author: {metadata.get('author', '未知')}"] - + if tags: fm.append("tags: [" + ", ".join(f'"{t}"' for t in tags) + "]") - + fm.append("---") return '\n'.join(fm) @@ -727,8 +728,8 @@ def main(): python deep_analyzer.py --metadata meta.json --content content.json --type video --api-key xxx """ ) - - parser.add_argument("--metadata", "-m", required=True, + + parser.add_argument("--metadata", "-m", required=True, help="元数据 JSON 文件路径") parser.add_argument("--content", "-c", required=True, help="内容 JSON 文件路径") @@ -738,41 +739,44 @@ def main(): help="内容类型 (默认: video)") parser.add_argument("--api-key", "-k", default=None, help="AI API Key (或设置 AI_API_KEY 环境变量)") - parser.add_argument("--provider", "-p", default="mock", + parser.add_argument("--provider", "-p", default="mock", choices=["openai", "zhipu", "minimax", "mock"], help="AI 提供商 (默认: mock)") parser.add_argument("--model", default="gpt-4", 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}") metadata = load_json(args.metadata) - + print(f"[Main] 加载内容: {args.content}") content = load_json(args.content) - + # 初始化 ai_client = SimpleAIClient( - api_key=args.api_key, - provider=args.provider, - model=args.model + api_key=args.api_key or os.environ.get("AI_API_KEY", ""), + provider=args.provider or os.environ.get("AI_PROVIDER", "mock"), + 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'), @@ -795,19 +799,19 @@ def main(): "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) - + print(f"[Main] 分析完成！输出: {args.output}") print(f"[Main] 摘要: {result.summary[:100]}...") print(f"[Main] 主题数: {len(result.main_topics)}")