feat(06-01): move zhihu modules to metadata directory

- Move zhihu_extractor.py, zhihu_metadata.py, zhihu_article_crawler.py, markdown_converter.py
- Update metadata/__init__.py to export ZhihuPlaywrightExtractor

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

kb_workflow/metadata/__init__.py

@@ -4,7 +4,9 @@
 """
 
 from .bilibili_extractor import BilibiliExtractor
+from .zhihu_extractor import ZhihuPlaywrightExtractor
 
 __all__ = [
     "BilibiliExtractor",
+    "ZhihuPlaywrightExtractor",
 ]

kb_workflow/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。
+
+        同时移除 <noscript> 标签，因为知乎等平台在 noscript 中包含备用图片，
+        会导致修复后图片重复出现。
+
+        Args:
+            html: 输入的 HTML 字符串
+
+        Returns:
+            处理后的 HTML 字符串
+        """
+        soup = BeautifulSoup(html, 'html.parser')
+
+        # 先移除 noscript 标签，避免图片重复
+        for noscript in soup.find_all('noscript'):
+            noscript.decompose()
+
+        for img in soup.find_all('img'):
+            src = img.get('src', '')
+
+            # 如果 src 不是占位符，保持原样
+            if not src or not src.startswith('data:'):
+                continue
+
+            # 尝试获取真实图片 URL
+            # 优先 data-actualsrc，其次 data-original
+            real_src = img.get('data-actualsrc') or img.get('data-original')
+
+            if real_src:
+                img['src'] = real_src
+
+        return str(soup)

kb_workflow/metadata/zhihu_article_crawler.py

@@ -0,0 +1,236 @@
+#!/usr/bin/env python3
+"""
+知乎文章爬取脚本 - 基于 MediaCrawler
+用于爬取知乎专栏文章并导出到知识库格式
+"""
+import argparse
+import json
+import os
+import re
+import sys
+import subprocess
+import shutil
+from pathlib import Path
+from datetime import datetime
+
+# 配置
+MEDIACRAWLER_DIR = Path.home() / "work" / "MediaCrawler"
+DEFAULT_OUTPUT_DIR = Path.home() / "work" / "knowledge-base" / "data" / "zhihu"
+
+
+def load_env():
+    """加载环境变量"""
+    env_file = Path.home() / ".kb_env"
+    if env_file.exists():
+        with open(env_file) as f:
+            for line in f:
+                line = line.strip()
+                if line and not line.startswith("#") and "=" in line:
+                    key, val = line.split("=", 1)
+                    os.environ[key.strip()] = val.strip()
+
+
+def get_zhihu_cookie():
+    """获取知乎 Cookie"""
+    return os.environ.get("ZHIHU_COOKIE", "")
+
+
+def crawl_zhihu_article(article_url: str, output_dir: Path) -> dict:
+    """
+    使用 MediaCrawler 爬取知乎文章
+    
+    Args:
+        article_url: 知乎文章链接
+        output_dir: 输出目录
+    
+    Returns:
+        dict: 爬取结果
+    """
+    # 清理旧数据
+    data_dir = MEDIACRAWLER_DIR / "data" / "zhihu" / "json"
+    data_dir.mkdir(parents=True, exist_ok=True)
+    
+    # 备份现有数据
+    backup_dir = data_dir / "backup"
+    backup_dir.mkdir(exist_ok=True)
+    for f in data_dir.glob("detail_contents_*.json"):
+        shutil.move(str(f), str(backup_dir / f.name))
+    
+    # 配置 MediaCrawler
+    config_file = MEDIACRAWLER_DIR / "config" / "zhihu_config.py"
+    
+    # 读取配置文件并修改
+    with open(config_file) as f:
+        config_content = f.read()
+    
+    # 强制更新配置中的 ID 列表（每次都更新，确保爬取正确的文章）
+    article_id = article_url.split("/")[-1]
+    
+    # 使用正则表达式替换整个 ZHIHU_SPECIFIED_ID_LIST 部分
+    pattern = r'(ZHIHU_SPECIFIED_ID_LIST\s*=\s*\[)([\s\S]*?)(\])'
+    replacement = f'\\1\n    "{article_url}",  # article\n\\3'
+    
+    new_config = re.sub(pattern, replacement, config_content)
+    
+    # 如果正则替换失败（没有匹配到），使用简单替换
+    if new_config == config_content:
+        # 直接替换整个配置项
+        if 'ZHIHU_SPECIFIED_ID_LIST' in config_content:
+            # 找到配置项的范围并替换
+            lines = config_content.split('\n')
+            new_lines = []
+            in_list = False
+            for line in lines:
+                if 'ZHIHU_SPECIFIED_ID_LIST' in line and '=' in line:
+                    new_lines.append(f'ZHIHU_SPECIFIED_ID_LIST = ["{article_url}",  # article')
+                    in_list = True
+                elif in_list and line.strip().startswith(']'):
+                    new_lines.append(line)
+                    in_list = False
+                elif not in_list:
+                    new_lines.append(line)
+            new_config = '\n'.join(new_lines)
+    
+    with open(config_file, "w") as f:
+        f.write(new_config)
+    
+    # 运行 MediaCrawler
+    env = os.environ.copy()
+    cookie = get_zhihu_cookie()
+    if cookie:
+        env["ZHIHU_COOKIE"] = cookie
+    
+    cmd = [
+        "python3", "main.py",
+        "--platform", "zhihu",
+        "--lt", "cookie",
+        "--type", "detail"
+    ]
+    
+    print(f"运行命令: {' '.join(cmd)}")
+    print(f"工作目录: {MEDIACRAWLER_DIR}")
+    
+    result = subprocess.run(
+        cmd,
+        cwd=MEDIACRAWLER_DIR,
+        env=env,
+        capture_output=True,
+        text=True,
+        timeout=120
+    )
+    
+    print(f"返回码: {result.returncode}")
+    if result.returncode != 0:
+        print(f"stderr: {result.stderr[:500]}")
+    
+    # 查找最新的输出文件
+    output_files = sorted(data_dir.glob("detail_contents_*.json"), reverse=True)
+    
+    if output_files:
+        latest_file = output_files[0]
+        with open(latest_file) as f:
+            data = json.load(f)
+        
+        if data:
+            item = data[0]
+            return {
+                "success": True,
+                "title": item.get("title", ""),
+                "content": item.get("content_text", ""),
+                "url": item.get("content_url", ""),
+                "author": item.get("user_nickname", ""),
+                "created_time": item.get("created_time", 0),
+                "voteup": item.get("voteup_count", 0),
+                "comments": item.get("comment_count", 0),
+                "content_type": item.get("content_type", "article"),
+            }
+    
+    return {"success": False, "error": "未获取到数据"}
+
+
+def save_to_markdown(data: dict, output_path: Path) -> str:
+    """保存为 Markdown 格式"""
+    created = datetime.fromtimestamp(data["created_time"]).strftime("%Y-%m-%d %H:%M:%S") if data.get("created_time") else ""
+    
+    content = f"""---
+title: "{data['title']}"
+author: "{data['author']}"
+url: "{data['url']}"
+created: "{created}"
+voteup: {data.get('voteup', 0)}
+comments: {data.get('comments', 0)}
+platform: zhihu
+type: article
+---
+
+# {data['title']}
+
+**作者**: {data['author']}  
+**来源**: [知乎专栏]({data['url']})  
+**发布时间**: {created}  
+**点赞**: {data.get('voteup', 0)}  
+**评论**: {data.get('comments', 0)}
+
+---
+
+{data['content']}
+"""
+    
+    with open(output_path, "w", encoding="utf-8") as f:
+        f.write(content)
+    
+    return str(output_path)
+
+
+def main():
+    parser = argparse.ArgumentParser(description="知乎文章爬取工具 (基于 MediaCrawler)")
+    parser.add_argument("--url", "-u", required=True, help="知乎文章链接")
+    parser.add_argument("--output", "-o", default=str(DEFAULT_OUTPUT_DIR), help="输出目录")
+    
+    args = parser.parse_args()
+    
+    # 加载环境变量
+    load_env()
+    
+    # 确保输出目录存在
+    output_dir = Path(args.output)
+    output_dir.mkdir(parents=True, exist_ok=True)
+    
+    print(f"目标文章: {args.url}")
+    print(f"输出目录: {output_dir}")
+    
+    # 爬取文章
+    print("\n开始爬取...")
+    result = crawl_zhihu_article(args.url, output_dir)
+    
+    if not result.get("success"):
+        print(f"爬取失败: {result.get('error', '未知错误')}")
+        sys.exit(1)
+    
+    print(f"\n爬取成功!")
+    print(f"标题: {result['title']}")
+    print(f"内容长度: {len(result['content'])} 字符")
+    
+    # 生成文件名
+    safe_title = "".join(c for c in result['title'] if c.isalnum() or c in " -_")[:50]
+    filename = f"zhihu_article_{safe_title}.md"
+    output_path = output_dir / filename
+    
+    # 保存为 Markdown
+    save_to_markdown(result, output_path)
+    print(f"\n已保存到: {output_path}")
+    
+    # 输出 JSON 格式
+    print("\n--- JSON 输出 ---")
+    print(json.dumps({
+        "success": True,
+        "type": "article",
+        "title": result["title"],
+        "filepath": str(output_path),
+        "content_length": len(result["content"]),
+        "message": f"爬取成功: {result['title']}"
+    }, ensure_ascii=False, indent=2))
+
+
+if __name__ == "__main__":
+    main()

kb_workflow/metadata/zhihu_extractor.py

@@ -0,0 +1,479 @@
+"""
+知乎 Playwright 内容提取器
+使用 Playwright 浏览器自动化工具抓取知乎内容
+"""
+
+import re
+import sys
+import os
+from typing import Dict, Optional, Any
+from urllib.parse import urlparse
+
+# 添加当前目录到 Python 路径以支持绝对导入
+current_dir = os.path.dirname(os.path.abspath(__file__))
+if current_dir not in sys.path:
+    sys.path.insert(0, current_dir)
+
+try:
+    from .markdown_converter import MarkdownConverter
+except ImportError:
+    from markdown_converter import MarkdownConverter
+
+
+class PlaywrightBrowser:
+    """Playwright 浏览器管理器"""
+
+    def __init__(self, headless: bool = True, timeout: int = 30):
+        self.headless = headless
+        self.timeout = timeout * 1000
+        self.browser = None
+        self.context = None
+        self.page = None
+
+    def start(self, cookie: str = None):
+        """启动浏览器"""
+        from playwright.sync_api import sync_playwright
+        self.playwright = sync_playwright().start()
+        self.browser = self.playwright.chromium.launch(headless=self.headless)
+        
+        # 创建浏览器上下文
+        context_options = {
+            'user_agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36'
+        }
+        
+        self.context = self.browser.new_context(**context_options)
+        
+        # 如果提供了 Cookie，设置 Cookie
+        if cookie:
+            self._add_cookies(cookie)
+        
+        self.page = self.context.new_page()
+        self.page.set_default_timeout(self.timeout)
+    
+    def _add_cookies(self, cookie_str: str):
+        """添加 Cookie 到浏览器上下文"""
+        cookies = []
+        for item in cookie_str.split(';'):
+            item = item.strip()
+            if '=' in item:
+                name, value = item.split('=', 1)
+                cookies.append({
+                    'name': name.strip(),
+                    'value': value.strip(),
+                    'domain': '.zhihu.com',
+                    'path': '/',
+                    'secure': True,
+                    'httpOnly': False,
+                    'sameSite': 'Lax'
+                })
+        self.context.add_cookies(cookies)
+
+    def close(self):
+        """关闭浏览器"""
+        if self.page:
+            self.page.close()
+        if self.context:
+            self.context.close()
+        if self.browser:
+            self.browser.close()
+        if hasattr(self, 'playwright'):
+            self.playwright.stop()
+
+    def goto(self, url: str, wait_until: str = "networkidle"):
+        """导航到指定URL"""
+        return self.page.goto(url, wait_until=wait_until)
+
+    def wait_for_selector(self, selector: str, timeout: int = None):
+        """等待选择器出现"""
+        timeout_ms = timeout * 1000 if timeout else self.timeout
+        return self.page.wait_for_selector(selector, timeout=timeout_ms)
+
+    def query_selector(self, selector: str):
+        """查询选择器"""
+        return self.page.query_selector(selector)
+
+    def query_selector_all(self, selector: str):
+        """查询所有匹配的选择器"""
+        return self.page.query_selector_all(selector)
+
+    def evaluate(self, script: str):
+        """执行 JavaScript 代码"""
+        return self.page.evaluate(script)
+
+
+class ZhihuPlaywrightExtractor:
+    """知乎内容提取器 - 使用 Playwright"""
+
+    URL_PATTERNS = {
+        'zvideo': r'zhihu\.com/zvideo/(\d+)',
+        'zhuanlan': r'zhuanlan\.zhihu\.com/p/(\d+)',
+        'answer': r'zhihu\.com/question/\d+/answer/(\d+)',
+        'question': r'zhihu\.com/question/(\d+)',
+    }
+
+    def __init__(self, headless: bool = True, timeout: int = 30, cookie: str = None):
+        """
+        初始化提取器
+
+        Args:
+            headless: 是否使用无头浏览器
+            timeout: 超时时间（秒）
+            cookie: 知乎 Cookie 字符串（可选）
+        """
+        self.cookie = cookie
+        self.browser = PlaywrightBrowser(headless=headless, timeout=timeout)
+        self.markdown_converter = MarkdownConverter()
+
+    def identify_url_type(self, url: str) -> Optional[str]:
+        """
+        识别 URL 类型
+
+        Returns:
+            'answer', 'article', 'question' 或 None
+        """
+        for url_type, pattern in self.URL_PATTERNS.items():
+            if re.search(pattern, url):
+                return url_type
+        return None
+
+    def extract_id(self, url: str) -> Optional[str]:
+        """
+        从 URL 中提取内容 ID
+
+        Returns:
+            内容 ID 或 None
+        """
+        url_type = self.identify_url_type(url)
+        if url_type:
+            pattern = self.URL_PATTERNS[url_type]
+            match = re.search(pattern, url)
+            if match:
+                return match.group(1)
+        return None
+
+    def _extract_text_content(self, element) -> str:
+        """提取元素的文本内容"""
+        if element is None:
+            return ''
+        return element.inner_text() if hasattr(element, 'inner_text') else ''
+
+    def _extract_html_content(self, element) -> str:
+        """提取元素的 HTML 内容"""
+        if element is None:
+            return ''
+        return element.inner_html() if hasattr(element, 'inner_html') else ''
+
+    def _extract_zvideo_content(self) -> Dict[str, Any]:
+        """提取知乎视频内容"""
+        result = {
+            'type': 'zvideo',
+            'title': '',
+            'author': '',
+            'content': '',
+            'content_markdown': '',
+            'publish_time': '',
+            'vote_up': 0,
+        }
+
+        # 打印页面标题用于调试
+        page_title = self.browser.page.title()
+        print(f"页面标题: {page_title}")
+
+        # 提取视频标题
+        title_elem = self.browser.query_selector('.VideoPlaylistInfo-title')
+        if title_elem:
+            result['title'] = self._extract_text_content(title_elem)
+        
+        # 备用：尝试其他选择器
+        if not result['title']:
+            title_elem = self.browser.query_selector('h1')
+            if title_elem:
+                result['title'] = self._extract_text_content(title_elem)
+
+        # 提取作者信息
+        author_elem = self.browser.query_selector('.AuthorInfo-name')
+        if author_elem:
+            result['author'] = self._extract_text_content(author_elem)
+
+        # 提取视频描述
+        desc_elem = self.browser.query_selector('.Video-desc')
+        if desc_elem:
+            html_content = self._extract_html_content(desc_elem)
+            result['content'] = html_content
+            result['content_markdown'] = self.markdown_converter.convert(html_content)
+
+        # 提取发布时间
+        time_elem = self.browser.query_selector('.Video-meta time')
+        if time_elem:
+            result['publish_time'] = time_elem.get_attribute('datetime') or self._extract_text_content(time_elem)
+
+        # 提取点赞数
+        vote_elem = self.browser.query_selector('.VideoLikeButton-count')
+        if vote_elem:
+            vote_text = self._extract_text_content(vote_elem)
+            try:
+                result['vote_up'] = int(vote_text.replace(',', ''))
+            except (ValueError, AttributeError):
+                pass
+
+        return result
+
+    def _extract_zhuanlan_content(self) -> Dict[str, Any]:
+        """提取知乎专栏文章内容"""
+        result = {
+            'type': 'zhuanlan',
+            'title': '',
+            'author': '',
+            'content': '',
+            'content_markdown': '',
+            'publish_time': '',
+            'vote_up': 0,
+        }
+
+        # 打印页面标题用于调试
+        page_title = self.browser.page.title()
+        print(f"页面标题: {page_title}")
+
+        # 提取文章标题
+        title_elem = self.browser.query_selector('.Post-Title')
+        if title_elem:
+            result['title'] = self._extract_text_content(title_elem)
+
+        # 备用
+        if not result['title']:
+            title_elem = self.browser.query_selector('h1')
+            if title_elem:
+                result['title'] = self._extract_text_content(title_elem)
+
+        # 提取作者信息
+        author_elem = self.browser.query_selector('.AuthorInfo-name')
+        if author_elem:
+            result['author'] = self._extract_text_content(author_elem)
+
+        # 提取文章内容
+        content_elem = self.browser.query_selector('.Post-RichText')
+        if content_elem:
+            html_content = self._extract_html_content(content_elem)
+            result['content'] = html_content
+            result['content_markdown'] = self.markdown_converter.convert(html_content)
+
+        # 提取发布时间
+        time_elem = self.browser.query_selector('.ContentItem-time')
+        if time_elem:
+            result['publish_time'] = time_elem.get_attribute('data-tooltip') or self._extract_text_content(time_elem)
+
+        # 提取点赞数
+        vote_elem = self.browser.query_selector('.VoteButton--up .VoteButton-label')
+        if vote_elem:
+            vote_text = self._extract_text_content(vote_elem)
+            try:
+                result['vote_up'] = int(vote_text.replace(',', ''))
+            except (ValueError, AttributeError):
+                pass
+
+        return result
+
+    def _extract_answer_content(self) -> Dict[str, Any]:
+        """提取知乎回答内容"""
+        result = {
+            'type': 'answer',
+            'title': '',  # 问题标题
+            'question_id': '',
+            'answer_id': '',
+            'author': '',
+            'content': '',
+            'content_markdown': '',
+            'publish_time': '',
+            'vote_up': 0,
+        }
+
+        # 打印页面标题用于调试
+        page_title = self.browser.page.title()
+        print(f"页面标题: {page_title}")
+
+        # 提取问题标题
+        question_title_elem = self.browser.query_selector('.QuestionHeader-title')
+        if not question_title_elem:
+            question_title_elem = self.browser.query_selector('.QuestionHeader-content h1')
+        if question_title_elem:
+            result['title'] = self._extract_text_content(question_title_elem)
+
+        # 提取回答内容
+        content_elem = self.browser.query_selector('.RichText')
+        if not content_elem:
+            content_elem = self.browser.query_selector('.AnswerItem-richText')
+        if not content_elem:
+            content_elem = self.browser.query_selector('[class*="RichText"]')
+        if content_elem:
+            html_content = self._extract_html_content(content_elem)
+            result['content'] = html_content
+            result['content_markdown'] = self.markdown_converter.convert(html_content)
+
+        # 提取作者信息
+        author_elem = self.browser.query_selector('.AuthorInfo-name')
+        if author_elem:
+            result['author'] = self._extract_text_content(author_elem)
+
+        # 提取发布时间
+        time_elem = self.browser.query_selector('.AnswerItem-time')
+        if not time_elem:
+            time_elem = self.browser.query_selector('.ContentItem-time')
+        if time_elem:
+            result['publish_time'] = time_elem.get_attribute('data-tooltip') or self._extract_text_content(time_elem)
+
+        # 提取点赞数
+        vote_elem = self.browser.query_selector('.VoteButton--up .VoteButton-label')
+        if not vote_elem:
+            vote_elem = self.browser.query_selector('.VoteButton .Count')
+        if vote_elem:
+            vote_text = self._extract_text_content(vote_elem)
+            try:
+                result['vote_up'] = int(vote_text.replace(',', ''))
+            except (ValueError, AttributeError):
+                pass
+
+        return result
+
+    def _extract_question_content(self) -> Dict[str, Any]:
+        """提取知乎问题内容（包含所有回答）"""
+        result = {
+            'type': 'question',
+            'title': '',
+            'question_id': '',
+            'author': '',
+            'content': '',
+            'content_markdown': '',
+            'publish_time': '',
+            'answer_count': 0,
+        }
+
+        # 打印页面标题用于调试
+        page_title = self.browser.page.title()
+        print(f"页面标题: {page_title}")
+
+        # 提取问题标题
+        title_elem = self.browser.query_selector('.QuestionHeader-title')
+        if not title_elem:
+            title_elem = self.browser.query_selector('.QuestionHeader-content h1')
+        if title_elem:
+            result['title'] = self._extract_text_content(title_elem)
+
+        # 提取问题描述
+        desc_elem = self.browser.query_selector('.QuestionHeader-detail')
+        if not desc_elem:
+            desc_elem = self.browser.query_selector('.QuestionRichText')
+        if desc_elem:
+            html_content = self._extract_html_content(desc_elem)
+            result['content'] = html_content
+            result['content_markdown'] = self.markdown_converter.convert(html_content)
+
+        # 提取回答数量
+        answer_count_elem = self.browser.query_selector('.List-headerText span')
+        if answer_count_elem:
+            count_text = self._extract_text_content(answer_count_elem)
+            try:
+                # 提取数字
+                import re
+                match = re.search(r'(\d+)', count_text)
+                if match:
+                    result['answer_count'] = int(match.group(1))
+            except (ValueError, AttributeError):
+                pass
+
+        return result
+
+    def extract(self, url: str) -> Optional[Dict[str, Any]]:
+        """
+        从 URL 提取知乎内容
+
+        Args:
+            url: 知乎 URL
+
+        Returns:
+            包含提取内容的字典，结构如下:
+            {
+                'type': 'zvideo' | 'zhuanlan',
+                'title': str,
+                'author': str,
+                'content': str (HTML),
+                'content_markdown': str (Markdown),
+                'publish_time': str,
+                'vote_up': int,
+            }
+            如果提取失败返回 None
+        """
+        try:
+            # 识别 URL 类型
+            url_type = self.identify_url_type(url)
+            if not url_type:
+                print(f"无法识别 URL 类型: {url}")
+                return None
+
+            print(f"URL 类型: {url_type}")
+            print(f"启动浏览器...")
+
+            # 启动浏览器（传入 Cookie）
+            self.browser.start(cookie=self.cookie)
+
+            print(f"访问页面: {url}")
+            # 访问页面
+            self.browser.goto(url, wait_until="domcontentloaded")
+
+            print("等待页面加载...")
+            import time
+            time.sleep(3)  # 等待 JavaScript 执行
+
+            # 根据类型提取内容
+            if url_type == 'zvideo':
+                result = self._extract_zvideo_content()
+            elif url_type == 'zhuanlan':
+                result = self._extract_zhuanlan_content()
+            elif url_type == 'answer':
+                result = self._extract_answer_content()
+            elif url_type == 'question':
+                result = self._extract_question_content()
+            else:
+                print(f"不支持的类型: {url_type}")
+                return None
+
+            # 添加额外信息
+            result['url'] = url
+            result['content_id'] = self.extract_id(url)
+
+            print(f"提取成功! 标题: {result.get('title', '')[:50]}...")
+            return result
+
+        except Exception as e:
+            print(f"提取知乎内容时出错: {e}")
+            import traceback
+            traceback.print_exc()
+            return None
+        finally:
+            # 关闭浏览器
+            self.browser.close()
+
+    def __enter__(self):
+        """上下文管理器入口"""
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        """上下文管理器出口"""
+        self.browser.close()
+        return False
+
+
+# 便捷函数
+def extract_zhihu_content(url: str, headless: bool = True, timeout: int = 30) -> Optional[Dict[str, Any]]:
+    """
+    便捷函数：提取知乎内容
+
+    Args:
+        url: 知乎 URL
+        headless: 是否使用无头浏览器
+        timeout: 超时时间（秒）
+
+    Returns:
+        提取的内容字典
+    """
+    with ZhihuPlaywrightExtractor(headless=headless, timeout=timeout) as extractor:
+        return extractor.extract(url)

kb_workflow/metadata/zhihu_metadata.py

@@ -0,0 +1,206 @@
+#!/usr/bin/env python3
+"""
+知乎元数据获取工具
+
+支持以下内容类型：
+- zvideo: 知乎视频 (https://www.zhihu.com/zvideo/xxx)
+- zhuanlan: 知乎专栏文章 (https://zhuanlan.zhihu.com/p/xxx)
+- answer: 知乎回答 (https://www.zhihu.com/question/xxx/answer/yyy)
+- question: 知乎问题 (https://www.zhihu.com/question/xxx)
+"""
+
+import argparse
+import json
+import sys
+import os
+from typing import Dict, Optional, Any
+
+# 添加当前目录到 Python 路径
+current_dir = os.path.dirname(os.path.abspath(__file__))
+if current_dir not in sys.path:
+    sys.path.insert(0, current_dir)
+
+from zhihu_extractor import ZhihuPlaywrightExtractor
+
+
+def extract_zvideo(video_id: str, headless: bool = True, cookie: str = None) -> Dict[str, Any]:
+    """
+    获取知乎视频元数据
+
+    Args:
+        video_id: 视频 ID (数字字符串)
+        headless: 是否使用无头浏览器
+        cookie: 知乎 Cookie（可选）
+
+    Returns:
+        视频元数据字典
+    """
+    url = f"https://www.zhihu.com/zvideo/{video_id}"
+    return _extract_metadata(url, "zvideo", headless, cookie)
+
+
+def extract_zhuanlan(article_id: str, headless: bool = True, cookie: str = None) -> Dict[str, Any]:
+    """
+    获取知乎专栏文章元数据
+
+    Args:
+        article_id: 文章 ID (数字字符串)
+        headless: 是否使用无头浏览器
+        cookie: 知乎 Cookie（可选）
+
+    Returns:
+        文章元数据字典
+    """
+    url = f"https://zhuanlan.zhihu.com/p/{article_id}"
+    return _extract_metadata(url, "zhuanlan", headless, cookie)
+
+
+def extract_answer(answer_id: str, headless: bool = True, cookie: str = None) -> Dict[str, Any]:
+    """
+    获取知乎回答元数据
+
+    Args:
+        answer_id: 回答 ID (数字字符串)
+        headless: 是否使用无头浏览器
+        cookie: 知乎 Cookie（可选）
+
+    Returns:
+        回答元数据字典
+    """
+    # 回答需要构建包含问题ID的URL，但answer_id本身足以识别
+    # 使用一个占位符问题ID，因为提取器会从URL中提取answer_id
+    url = f"https://www.zhihu.com/question/0/answer/{answer_id}"
+    return _extract_metadata(url, "answer", headless, cookie)
+
+
+def extract_question(question_id: str, headless: bool = True, cookie: str = None) -> Dict[str, Any]:
+    """
+    获取知乎问题元数据
+
+    Args:
+        question_id: 问题 ID (数字字符串)
+        headless: 是否使用无头浏览器
+        cookie: 知乎 Cookie（可选）
+
+    Returns:
+        问题元数据字典
+    """
+    url = f"https://www.zhihu.com/question/{question_id}"
+    return _extract_metadata(url, "question", headless, cookie)
+
+
+def _extract_metadata(url: str, content_type: str, headless: bool, cookie: str = None) -> Dict[str, Any]:
+    """
+    内部函数：提取元数据
+
+    Args:
+        url: 知乎 URL
+        content_type: 内容类型 (zvideo/zhuanlan)
+        headless: 是否使用无头浏览器
+        cookie: 知乎 Cookie（可选）
+
+    Returns:
+        元数据字典
+    """
+    extractor = ZhihuPlaywrightExtractor(headless=headless, cookie=cookie)
+    result = extractor.extract(url)
+
+    if result:
+        # 标准化输出格式
+        metadata = {
+            "type": content_type,
+            "id": result.get("content_id", ""),
+            "url": result.get("url", url),
+            "title": result.get("title", ""),
+            "author": result.get("author", ""),
+            "publish_time": result.get("publish_time", ""),
+            "vote_up": result.get("vote_up", 0),
+            "content_markdown": result.get("content_markdown", ""),
+            "raw_html": result.get("content", ""),
+        }
+        return metadata
+    else:
+        return {
+            "type": content_type,
+            "error": "Failed to extract content",
+            "url": url
+        }
+
+
+def main():
+    """命令行入口"""
+    parser = argparse.ArgumentParser(
+        description="知乎元数据获取工具",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog="""
+支持的类型:
+  zvideo   - 知乎视频 (https://www.zhihu.com/zvideo/123456)
+  zhuanlan - 知乎专栏文章 (https://zhuanlan.zhihu.com/p/123456)
+  answer   - 知乎回答 (https://www.zhihu.com/question/xxx/answer/yyy)
+  question - 知乎问题 (https://www.zhihu.com/question/xxx)
+
+示例:
+  python zhihu_metadata.py --id "123456" --type zvideo
+  python zhihu_metadata.py --id "123456" --type zhuanlan
+  python zhihu_metadata.py --id "123456" --type answer
+  python zhihu_metadata.py --id "123456" --type question
+        """
+    )
+
+    parser.add_argument(
+        "--id", "-i",
+        required=True,
+        help="知乎内容 ID (视频ID/文章ID/回答ID/问题ID)"
+    )
+
+    parser.add_argument(
+        "--type", "-t",
+        required=True,
+        choices=["zvideo", "zhuanlan", "answer", "question"],
+        help="内容类型: zvideo=视频, zhuanlan=专栏文章, answer=回答, question=问题"
+    )
+
+    parser.add_argument(
+        "--output", "-o",
+        default=None,
+        help="输出文件路径 (默认: stdout)"
+    )
+
+    parser.add_argument(
+        "--headless",
+        action="store_true",
+        default=True,
+        help="使用无头浏览器模式 (默认: True)"
+    )
+
+    args = parser.parse_args()
+
+    # 获取知乎 Cookie（优先使用环境变量，其次使用参数）
+    cookie = os.environ.get('ZHIHU_COOKIE')
+    
+    # 提取元数据（传入 Cookie）
+    if args.type == "zvideo":
+        result = extract_zvideo(args.id, headless=args.headless, cookie=cookie)
+    elif args.type == "zhuanlan":
+        result = extract_zhuanlan(args.id, headless=args.headless, cookie=cookie)
+    elif args.type == "answer":
+        result = extract_answer(args.id, headless=args.headless, cookie=cookie)
+    elif args.type == "question":
+        result = extract_question(args.id, headless=args.headless, cookie=cookie)
+    else:
+        print(f"不支持的类型: {args.type}")
+        sys.exit(1)
+
+    # 输出结果
+    json_output = json.dumps(result, ensure_ascii=False, indent=2)
+
+    if args.output:
+        with open(args.output, 'w', encoding='utf-8') as f:
+            f.write(json_output)
+        print(f"结果已保存到: {args.output}")
+    else:
+        print(json_output)
+
+
+if __name__ == "__main__":
+    main()