自动同步: 2026-03-22 02:10

skills/kb_workflow/deep_analyzer/__init__.py

@@ -37,20 +37,25 @@ class AnalysisResult:
 class SimpleAIClient:
     """AI 客户端"""
 
-    def __init__(self, api_key: str = None, provider: str = None, model: str = None, base_url: str = None):
+    def __init__(self, api_key: str = None, provider: str = None, model: str = None, base_url: str = None, disable_reasoning: bool = False):
         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
+        self.disable_reasoning = disable_reasoning
 
-    def generate(self, prompt: str, max_tokens: int = 16000) -> str:
+    def generate(self, prompt: str, max_tokens: int = 16000, disable_reasoning: bool = None) -> 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)
 
+        # 确定 disable_reasoning 的值：优先使用参数，否则使用实例变量
+        if disable_reasoning is None:
+            disable_reasoning = self.disable_reasoning
+
         if not api_key:
             raise ValueError("AI_API_KEY 未设置")
 
@@ -63,7 +68,7 @@ class SimpleAIClient:
                 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)
+                    result = self._call_zhipu(prompt, max_tokens, api_key, model, base_url, disable_reasoning)
                 elif provider == "minimax":
                     result = self._call_minimax(prompt, max_tokens, api_key, model, base_url)
                 else:
@@ -88,35 +93,51 @@ class SimpleAIClient:
         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:
+    def _call_zhipu(self, prompt: str, max_tokens: int, api_key: str, model: str, base_url: str, disable_reasoning: bool = False) -> str:
         import requests
         import sys
 
         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}"}
-        # GLM-4.5/4.6/4.7/5 默认启用thinking mode，使用 thinking.type 参数禁用
+        # GLM-4.5/4.6/4.7/5 默认启用thinking mode
+        # 通过 disable_reasoning 参数控制是否禁用思考模式
         payload = {
             "model": model,
             "messages": [{"role": "user", "content": prompt}],
-            "max_tokens": max_tokens,
-            "thinking": {
-                "type": "disabled"  # 禁用思考模式，获取直接响应
-            }
+            "max_tokens": max_tokens
         }
+
+        # 如果明确要求禁用思考模式，添加 thinking 参数
+        if disable_reasoning:
+            payload["thinking"] = {"type": "disabled"}
         response = requests.post(url, headers=headers, json=payload, timeout=600)
         response.raise_for_status()
         data = response.json()
         message = data.get('choices', [{}])[0].get('message', {})
+
+        # 调试：打印完整的 API 响应结构
+        print(f"[DEBUG] disable_reasoning={disable_reasoning}", file=sys.stderr)
+        print(f"[DEBUG] message keys: {list(message.keys())}", file=sys.stderr)
+        if 'content' in message:
+            print(f"[DEBUG] content 长度: {len(message['content'])}, 前200字符: {message['content'][:200]}", file=sys.stderr)
+        if 'reasoning_content' in message:
+            print(f"[DEBUG] reasoning_content 长度: {len(message['reasoning_content'])}, 前200字符: {message['reasoning_content'][:200]}", file=sys.stderr)
+
+        # 返回 content 字段
+        # 当禁用思考模式时，content 包含直接答案
+        # 当启用思考模式时，content 可能包含思考过程和分析
         content = message.get('content', '')
 
         if not content:
-            # 如果仍然为空，检查是否有reasoning_content（用于调试）
+            # 如果 content 为空，检查是否有 reasoning_content
             reasoning = message.get('reasoning_content', '')
             if reasoning:
-                print(f"[DEBUG] Zhipu API返回空的content，但有reasoning_content。", file=sys.stderr)
-                print(f"[DEBUG] 这可能表示thinking.type参数未生效或模型不支持。", file=sys.stderr)
+                print(f"[DEBUG] Zhipu API返回空的content，使用reasoning_content。", file=sys.stderr)
                 print(f"[DEBUG] reasoning_content前200字符: {reasoning[:200]}", file=sys.stderr)
+                content = reasoning
+
+        if not content:
             raise ValueError(f"Zhipu API返回空内容。请检查API配置和模型参数。响应数据: {data}")
 
         return content

skills/kb_workflow/pipeline/processors/archive.py

@@ -7,6 +7,7 @@ to the archive directory organized by platform.
 from __future__ import annotations
 
 import logging
+import re
 import shutil
 from datetime import datetime
 from pathlib import Path
@@ -65,9 +66,14 @@ class ArchiveProcessor(BaseProcessor):
             platform_dir = archive_base / input_data.platform
             platform_dir.mkdir(parents=True, exist_ok=True)
 
-            # 文件命名: {YYYYMMDD}_{content_id}_analysis.md
+            # 从源文件中提取标题（用于文件命名）
+            title = self._extract_title_from_file(Path(input_data.analysis_file))
+            # 清理标题：移除特殊字符，限制长度
+            safe_title = self._sanitize_title(title) if title else input_data.content_id
+
+            # 文件命名: {YYYYMMDD}_{content_id}_{title}_analysis.md
             timestamp = datetime.now().strftime("%Y%m%d")
-            archive_filename = f"{timestamp}_{input_data.content_id}_analysis.md"
+            archive_filename = f"{timestamp}_{input_data.content_id}_{safe_title}_analysis.md"
             archive_path = platform_dir / archive_filename
 
             # Step 3: 检查重复
@@ -147,6 +153,64 @@ class ArchiveProcessor(BaseProcessor):
         matches = list(platform_dir.glob(pattern))
         return matches[0] if matches else None
 
+    def _extract_title_from_file(self, file_path: Path) -> Optional[str]:
+        """从 Markdown 文件的 frontmatter 中提取标题
+
+        Args:
+            file_path: Markdown 文件路径
+
+        Returns:
+            标题字符串，如果未找到返回 None
+        """
+        try:
+            content = file_path.read_text(encoding='utf-8')
+
+            # 查找 frontmatter 中的 title 字段
+            # 格式：title: "标题内容" 或 title: 标题内容
+            title_match = re.search(r'^title:\s*(?:"([^"]*)"|(.+))$', content, re.MULTILINE)
+
+            if title_match:
+                # 优先使用引号内的内容，否则使用第二个捕获组
+                title = title_match.group(1) or title_match.group(2)
+                return title.strip() if title else None
+
+            return None
+        except Exception as e:
+            logger.warning(f"Failed to extract title from {file_path}: {e}")
+            return None
+
+    def _sanitize_title(self, title: str) -> str:
+        """清理标题用于文件命名
+
+        移除特殊字符，限制长度，转换为安全的文件名。
+
+        Args:
+            title: 原始标题
+
+        Returns:
+            清理后的标题
+        """
+        # 移除或替换不适合文件名的字符
+        # 保留中文、字母、数字、下划线、连字符
+        # 替换所有特殊字符（包括标点符号和常见符号）
+        safe = re.sub(r'[\\/:*?"<>|，。！？、；：""\'\'\s@#$%&*+\-=\[\]{}()<>]+', '_', title)
+
+        # 移除连续的下划线
+        safe = re.sub(r'_+', '_', safe)
+
+        # 去除首尾下划线和空格
+        safe = safe.strip('_')
+
+        # 限制长度（避免文件名过长）
+        if len(safe) > 50:
+            safe = safe[:50]
+
+        # 如果清理后为空，返回默认值
+        if not safe:
+            safe = "untitled"
+
+        return safe
+
     def _ask_overwrite(self, existing_file: Path, is_non_interactive: bool = False) -> bool:
         """询问用户是否覆盖已存在的文件
 

skills/kb_workflow/pipeline/processors/srt.py

@@ -166,7 +166,7 @@ class SRTProcessor(BaseProcessor):
             metadata_file: 元数据文件路径
 
         Returns:
-            元数据字典
+            元数据字典（扁平化后的）
         """
         path = Path(metadata_file)
         if not path.exists():
@@ -174,7 +174,21 @@ class SRTProcessor(BaseProcessor):
             return {}
 
         with open(path, 'r', encoding='utf-8') as f:
-            return json.load(f)
+            raw_metadata = json.load(f)
+
+        # 扁平化嵌套结构（与 AnalyzeProcessor._normalize_metadata 一致）
+        if not raw_metadata:
+            return {}
+
+        # 如果已经是扁平结构（顶层有 title）
+        if 'title' in raw_metadata:
+            return raw_metadata
+
+        # 处理嵌套结构 {metadata: {...}, stats: {...}}
+        if 'metadata' in raw_metadata:
+            return raw_metadata['metadata'].copy()
+
+        return raw_metadata
 
     def _extract_tags(self, paragraphs: List[str]) -> List[str]:
         """从段落中提取标签
@@ -216,16 +230,7 @@ class SRTProcessor(BaseProcessor):
         # 构建内容
         content_parts = [frontmatter, ""]
 
-        # 添加摘要（可选）
-        if paragraphs:
-            content_parts.append("## 摘要")
-            content_parts.append("")
-            # 使用第一段作为简单摘要
-            first_para = paragraphs[0][:200] if len(paragraphs[0]) > 200 else paragraphs[0]
-            content_parts.append(first_para + ("..." if len(paragraphs[0]) > 200 else ""))
-            content_parts.append("")
-
-        # 添加内容
+        # 直接添加内容，不生成摘要部分
         content_parts.append("## 内容")
         content_parts.append("")
         for para in paragraphs:
@@ -271,7 +276,12 @@ class SRTProcessor(BaseProcessor):
         # 时长（视频）
         duration = metadata.get('duration', 0)
         if duration:
-            duration_str = self._format_duration(duration)
+            # 如果 duration 已经是字符串格式（如 "05:42"），直接使用
+            # 否则（整数秒数）调用 _format_duration 转换
+            if isinstance(duration, str):
+                duration_str = duration
+            else:
+                duration_str = self._format_duration(duration)
             lines.append(f'duration: {duration_str}')
 
         # BVID

skills/kb_workflow/srt/corrector.py

@@ -26,7 +26,11 @@ class AIClientProtocol(Protocol):
 
 # 纠错 Prompt 模板
 CORRECTION_PROMPT = """你是一个文本校对专家。以下是语音识别生成的字幕，可能包含识别错误。
-请只纠正明显的错误（如同音字、漏字、错字），不要改变原意和风格。
+
+请执行以下任务：
+1. 纠正明显的错误（如同音字、漏字、错字）
+2. 添加适当的标点符号（逗号、句号、顿号等）
+3. 不要改变原意和风格
 
 原文：
 {text}
@@ -100,7 +104,8 @@ class ASRCorrector:
         """
         try:
             prompt = CORRECTION_PROMPT.format(text=text)
-            result = self.ai_client.generate(prompt, max_tokens=len(text) + 500)
+            # ASR 纠错需要直接结果，不需要 AI 的分析过程，因此禁用思考模式
+            result = self.ai_client.generate(prompt, max_tokens=len(text) + 500, disable_reasoning=True)
             return result.strip()
         except Exception as e:
             logger.warning(f"ASR correction failed: {e}, returning original text")