diff --git a/.planning/ROADMAP.md b/.planning/ROADMAP.md index b8e9809..dde7a3f 100644 --- a/.planning/ROADMAP.md +++ b/.planning/ROADMAP.md @@ -34,7 +34,7 @@ Decimal phases appear between their surrounding integers in numeric order. Plans: - [ ] 01-01-PLAN.md — State Manager: workflow_state.json 管理 (PIPE-02, REL-01, REL-03) - [ ] 01-02-PLAN.md — Schema Registry: 统一输入输出格式定义 (PIPE-01, PIPE-03, VAL-01, REL-03) -- [ ] 01-03-PLAN.md — Error Handler & Config: 结构化错误处理和统一配置 (REL-02, MNT-02, MNT-03) +- [x] 01-03-PLAN.md — Error Handler & Config: 结构化错误处理和统一配置 (REL-02, MNT-02, MNT-03) ### Phase 2: Platform Skills **Goal**: 集成B站和知乎平台的内容获取能力,形成从链接到原始内容的完整流程 @@ -46,12 +46,12 @@ Plans: 3. 输入知乎问答/专栏链接,自动提取内容和元数据 4. 重复执行同一链接,不产生重复文件(幂等) 5. 403/超时错误触发降级策略,记录失败原因到状态文件 -**Plans**: TBD +**Plans:** 3 plans Plans: -- [ ] 02-01: Link Parser & B站 Metadata 集成 -- [ ] 02-02: B站 Content (视频/专栏) 集成 -- [ ] 02-03: 知乎 Metadata & Content 集成 +- [ ] 02-01-PLAN.md — Link Parser & Idempotency Checker (PLAT-01, PLAT-02, PLAT-03) +- [ ] 02-02-PLAN.md — Bilibili Adapter: 视频元数据 + 专栏内容 (PLAT-01, PLAT-02) +- [ ] 02-03-PLAN.md — Zhihu Adapter: 问答 + 专栏内容 (PLAT-03) ### Phase 3: Processing Skills **Goal**: 集成ASR和OCR处理能力,支持视频和图片的多模态内容提取 @@ -93,12 +93,12 @@ Phases execute in numeric order: 1 → 2 → 3 → 4 | Phase | Plans Complete | Status | Completed | |-------|----------------|--------|-----------| -| 1. Foundation | 2/3 | In Progress | 01-02, 01-01 | -| 2. Platform Skills | 0/3 | Not started | - | +| 1. Foundation | 3/3 | Complete | 01-02, 01-01, 01-03 | +| 2. Platform Skills | 0/3 | Ready | - | | 3. Processing Skills | 0/2 | Not started | - | | 4. Analysis & Orchestration | 0/3 | Not started | - | --- *Roadmap created: 2026-03-13* -*Last updated: 2026-03-13 - Phase 1 in progress (2/3 plans complete)* +*Last updated: 2026-03-13 - Phase 2 planned (3 plans ready)* *Granularity: coarse (4 phases)* diff --git a/.planning/phases/02-platform-skills/02-01-PLAN.md b/.planning/phases/02-platform-skills/02-01-PLAN.md new file mode 100644 index 0000000..64b1306 --- /dev/null +++ b/.planning/phases/02-platform-skills/02-01-PLAN.md @@ -0,0 +1,239 @@ +--- +phase: 02-platform-skills +plan: 01 +type: execute +wave: 1 +depends_on: [] +files_modified: + - kb_pipeline/link_parser.py + - kb_pipeline/idempotency.py + - kb_pipeline/__init__.py + - tests/test_pipeline/test_link_parser.py + - tests/test_pipeline/test_idempotency.py +autonomous: true +requirements: [PLAT-01, PLAT-02, PLAT-03] +user_setup: [] + +must_haves: + truths: + - "输入B站视频URL,解析出 platform=bilibili, type=video, id=BVxxx" + - "输入知乎专栏URL,解析出 platform=zhihu, type=zhuanlan, id=xxx" + - "输入b23.tv短链,自动解析到真实URL并提取ID" + - "输入无法识别的URL,返回None和错误信息(不抛异常)" + - "幂等检查能正确判断文件是否已存在" + artifacts: + - path: "kb_pipeline/link_parser.py" + provides: "统一链接解析器" + exports: ["LinkParser", "ParsedLink"] + - path: "kb_pipeline/idempotency.py" + provides: "幂等检查机制" + exports: ["IdempotencyChecker"] + - path: "tests/test_pipeline/test_link_parser.py" + provides: "LinkParser 单元测试" + min_lines: 50 + - path: "tests/test_pipeline/test_idempotency.py" + provides: "幂等检查单元测试" + min_lines: 30 + key_links: + - from: "kb_pipeline/link_parser.py" + to: "kb_pipeline/config.py" + via: "get_settings for base_path" + pattern: "from.*config.*import" + - from: "kb_pipeline/idempotency.py" + to: "kb_pipeline/link_parser.py" + via: "ParsedLink for content_id" + pattern: "from.*link_parser.*import" +--- + + +创建统一的 LinkParser 类和幂等检查机制,为后续平台适配器提供链接解析和重复处理防护能力。 + +Purpose: 为所有平台提供统一的 URL 解析入口,支持 B站/知乎的内容类型识别和 ID 提取,同时提供幂等检查避免重复处理。 +Output: LinkParser 类、ParsedLink 数据类、IdempotencyChecker 类 + + + +@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md +@/home/starmen1265/.claude/get-shit-done/templates/summary.md + + + +@.planning/PROJECT.md +@.planning/ROADMAP.md +@.planning/STATE.md +@.planning/phases/02-platform-skills/02-CONTEXT.md +@.planning/phases/02-platform-skills/02-RESEARCH.md +@.planning/phases/02-platform-skills/02-VALIDATION.md + + + + +From kb_pipeline/schemas.py: +```python +class StepInput(BaseModel): + content_id: str = Field(..., min_length=1, description="内容唯一标识") + platform: Literal["bilibili", "zhihu"] = Field(..., description="平台名称") + +class StepStatus(str, Enum): + PENDING = "pending" + COMPLETED = "completed" + FAILED = "failed" +``` + +From kb_pipeline/config.py: +```python +class KBSettings(BaseSettings): + base_path: str = Field(..., description="知识库根目录") + archive_path: str = Field(..., description="归档目录") + +@lru_cache +def get_settings() -> KBSettings: + """获取配置单例""" +``` + +From kb_pipeline/errors.py: +```python +class KBError(Exception): + def __init__(self, message, error_type, step_name, input_summary, suggestion, cause=None) + +class ValidationError(KBError): + def __init__(self, step_name, input_summary, validation_errors=None, cause=None) +``` + +From kb_pipeline/retry.py: +```python +retry_on_network_error = retry( + stop=stop_after_attempt(3), + wait=wait_exponential(multiplier=1, min=2, max=10), + retry=retry_if_exception_type(NetworkError), +) +``` + + + + + +From kb-link-parser/link-parser.py: +```python +class LinkParser: + BILIBILI_VIDEO_PATTERN = re.compile(r'https?://(?:www\.)?bilibili\.com/video/(BV[a-zA-Z0-9]+)') + BILIBILI_ARTICLE_PATTERN = re.compile(r'https?://(?:www\.)?bilibili\.com/(?:article|read)/([a-zA-Z0-9]+)') + ZHIHAU_ANSWER_PATTERN = re.compile(r'https?://(?:www\.)?zhihu\.com/question/\d+/answer/(\d+)') + ZHIHAU_ZHUANLAN_PATTERN = re.compile(r'https?://zhuanlan\.zhihu\.com/p/(\d+)') + + def resolve_shortlink(self, url: str) -> Optional[str]: + # 使用 requests.head(url, allow_redirects=True, timeout=10) + + def parse(self, url: str) -> Dict[str, Any]: + # 返回 {platform, type, id, original_url, resolved_url, success, error} +``` + + + + + + + Task 1: Write interface contracts and tests for LinkParser + kb_pipeline/link_parser.py, tests/test_pipeline/test_link_parser.py + + - Test 1: 解析B站视频URL返回正确的 ParsedLink (platform=bilibili, type=video, id=BVxxx) + - Test 2: 解析B站专栏URL返回正确的 ParsedLink (platform=bilibili, type=article, id=cvxxx) + - Test 3: 解析知乎问答URL返回正确的 ParsedLink (platform=zhihu, type=answer, id=xxx) + - Test 4: 解析知乎专栏URL返回正确的 ParsedLink (platform=zhihu, type=zhuanlan, id=xxx) + - Test 5: 解析b23.tv短链返回解析后的真实URL + - Test 6: 无法识别的URL返回None(不抛异常) + + + 1. 创建 kb_pipeline/link_parser.py: + - 定义 ParsedLink dataclass,包含: platform, content_type, content_id, original_url, resolved_url + - 定义 LinkParser 类,实现 parse(url) -> Optional[ParsedLink] + - 支持 URL 类型: + * B站视频: bilibili.com/video/BVxxx + * B站专栏: bilibili.com/read/cvxxx + * 知乎问答: zhihu.com/question/xxx/answer/xxx + * 知乎专栏: zhuanlan.zhihu.com/p/xxx + - 短链支持: b23.tv 使用 requests.head(allow_redirects=True) 解析 + - 错误处理: 无法识别返回 None,不抛异常 + + 2. 创建 tests/test_pipeline/test_link_parser.py: + - 使用 pytest 和 parametrize 测试各种 URL 格式 + - Mock 短链解析以避免真实网络请求 + - 覆盖所有 CONTEXT.md 定义的支持 URL 类型 + + + pytest tests/test_pipeline/test_link_parser.py -x -v + + LinkParser 能正确解析所有支持的 URL 类型,无法识别的 URL 返回 None + + + + Task 2: Write interface contracts and tests for IdempotencyChecker + kb_pipeline/idempotency.py, tests/test_pipeline/test_idempotency.py + + - Test 1: 文件存在时 check_exists 返回 (True, [file_list]) + - Test 2: 文件不存在时 check_exists 返回 (False, []) + - Test 3: 部分文件存在时返回正确的存在文件列表 + + + 1. 创建 kb_pipeline/idempotency.py: + - 定义 IdempotencyChecker 类 + - 实现 check_exists(output_dir: Path, content_id: str, platform: str) -> Tuple[bool, list[Path]] + - 检查文件: + * 元数据: {platform}_{content_id}_metadata.json + * 内容: {platform}_{content_id}_content.md + - 使用 pathlib.Path 进行路径操作 + + 2. 创建 tests/test_pipeline/test_idempotency.py: + - 使用 tmp_path fixture 创建临时文件 + - 测试文件存在/不存在的各种场景 + - 测试部分文件存在的情况 + + + pytest tests/test_pipeline/test_idempotency.py -x -v + + IdempotencyChecker 能正确检查文件是否存在,返回准确的文件列表 + + + + Task 3: Update package exports + kb_pipeline/__init__.py + + 更新 kb_pipeline/__init__.py,导出新模块: + ```python + from .link_parser import LinkParser, ParsedLink + from .idempotency import IdempotencyChecker + + __all__ = [ + # ... existing exports ... + "LinkParser", + "ParsedLink", + "IdempotencyChecker", + ] + ``` + + + python -c "from kb_pipeline import LinkParser, ParsedLink, IdempotencyChecker; print('OK')" + + LinkParser, ParsedLink, IdempotencyChecker 可从 kb_pipeline 包根导入 + + + + + +- 所有测试通过: `pytest tests/test_pipeline/test_link_parser.py tests/test_pipeline/test_idempotency.py -v` +- 导入测试: `python -c "from kb_pipeline import LinkParser, IdempotencyChecker"` +- 覆盖率: `pytest tests/test_pipeline/ --cov=kb_pipeline --cov-report=term-missing` + + + +- [ ] LinkParser 正确解析 B站视频/专栏、知乎问答/专栏 URL +- [ ] b23.tv 短链自动解析到真实 URL +- [ ] 无法识别的 URL 返回 None(不抛异常) +- [ ] IdempotencyChecker 正确检查文件存在性 +- [ ] 所有测试通过(至少 10 个测试用例) +- [ ] 公共 API 可从 kb_pipeline 包根导入 + + + +After completion, create `.planning/phases/02-platform-skills/02-01-SUMMARY.md` + diff --git a/.planning/phases/02-platform-skills/02-02-PLAN.md b/.planning/phases/02-platform-skills/02-02-PLAN.md new file mode 100644 index 0000000..c5da558 --- /dev/null +++ b/.planning/phases/02-platform-skills/02-02-PLAN.md @@ -0,0 +1,336 @@ +--- +phase: 02-platform-skills +plan: 02 +type: execute +wave: 1 +depends_on: [02-01] +files_modified: + - kb_pipeline/platforms/__init__.py + - kb_pipeline/platforms/base.py + - kb_pipeline/platforms/bilibili.py + - kb_pipeline/__init__.py + - tests/test_platforms/__init__.py + - tests/test_platforms/conftest.py + - tests/test_platforms/test_bilibili.py +autonomous: true +requirements: [PLAT-01, PLAT-02] +user_setup: [] + +must_haves: + truths: + - "输入BV号,获取B站视频元数据(标题、作者、时长、描述等)" + - "输入CV号,获取B站专栏元数据和正文内容" + - "元数据保存为结构化JSON文件" + - "内容保存为Markdown文件(YAML Frontmatter + 正文)" + - "403错误触发降级策略(API -> Playwright)" + - "网络错误自动重试3次" + artifacts: + - path: "kb_pipeline/platforms/base.py" + provides: "平台适配器基类" + exports: ["PlatformAdapter"] + - path: "kb_pipeline/platforms/bilibili.py" + provides: "B站平台适配器" + exports: ["BilibiliAdapter"] + min_lines: 100 + - path: "tests/test_platforms/test_bilibili.py" + provides: "B站适配器测试" + min_lines: 80 + key_links: + - from: "kb_pipeline/platforms/bilibili.py" + to: "kb_pipeline/retry.py" + via: "retry_on_network_error decorator" + pattern: "@retry_on_network_error" + - from: "kb_pipeline/platforms/bilibili.py" + to: "kb_pipeline/errors.py" + via: "NetworkError, AuthError" + pattern: "raise NetworkError" + - from: "kb_pipeline/platforms/bilibili.py" + to: "kb-bilibili-metadata/bilibili_extractor.py" + via: "BilibiliExtractor class" + pattern: "from kb_bilibili_metadata" +--- + + +创建 B站平台适配器,封装视频和专栏的元数据提取与内容获取逻辑,复用现有 BilibiliExtractor 实现。 + +Purpose: 为 B站视频和专栏提供统一的元数据提取和内容下载接口,支持错误重试和降级策略。 +Output: PlatformAdapter 基类、BilibiliAdapter 实现类 + + + +@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md +@/home/starmen1265/.claude/get-shit-done/templates/summary.md + + + +@.planning/PROJECT.md +@.planning/ROADMAP.md +@.planning/STATE.md +@.planning/phases/02-platform-skills/02-CONTEXT.md +@.planning/phases/02-platform-skills/02-RESEARCH.md + + + + +From kb_pipeline/link_parser.py (Plan 01): +```python +@dataclass +class ParsedLink: + platform: str # "bilibili" | "zhihu" + content_type: str # "video" | "article" | "answer" | "zhuanlan" + content_id: str # BV号、CV号等 + original_url: str + resolved_url: str + +class LinkParser: + def parse(self, url: str) -> Optional[ParsedLink]: +``` + +From kb_pipeline/schemas.py: +```python +class StepInput(BaseModel): + content_id: str + platform: Literal["bilibili", "zhihu"] + +class StepOutput(BaseModel): + success: bool + error_message: Optional[str] + completed_at: datetime +``` + +From kb_pipeline/errors.py: +```python +class NetworkError(KBError): + def __init__(self, step_name, input_summary, status_code=None, cause=None) + +class AuthError(KBError): + def __init__(self, step_name, input_summary, auth_type=None, cause=None) +``` + +From kb_pipeline/retry.py: +```python +retry_on_network_error = retry( + stop=stop_after_attempt(3), + wait=wait_exponential(multiplier=1, min=2, max=10), +) +``` + +From kb_pipeline/config.py: +```python +class KBSettings(BaseSettings): + base_path: str + archive_path: str +``` + + + + + +From kb-bilibili-metadata/bilibili_extractor.py: +```python +class BilibiliExtractor: + def extract_video_metadata(self, bvid: str) -> Dict[str, Any]: + # 返回 {metadata: {...}, stats: {...}, description: str} + + def extract_article_metadata(self, cvid: str) -> Dict[str, Any]: + # 返回 {cvid, title, author, publish_time, ...} + + async def _extract_video_by_api(self, bvid: str) -> Dict[str, Any]: + # 使用 httpx 调用 API + +DEFAULT_HEADERS = { + 'User-Agent': 'Mozilla/5.0 ...', + 'Referer': 'https://www.bilibili.com/', +} +``` + + + + + + + Task 1: Write interface contracts for PlatformAdapter base class + kb_pipeline/platforms/__init__.py, kb_pipeline/platforms/base.py, tests/test_platforms/__init__.py, tests/test_platforms/conftest.py + + - Test 1: PlatformAdapter 是抽象基类,不能直接实例化 + - Test 2: 子类必须实现 get_content_type, fetch_metadata, fetch_content, get_output_filename 方法 + + + 1. 创建 kb_pipeline/platforms/__init__.py: + - 导出 PlatformAdapter, BilibiliAdapter + + 2. 创建 kb_pipeline/platforms/base.py: + ```python + from abc import ABC, abstractmethod + from typing import Dict, Any, Optional + from pathlib import Path + + class PlatformAdapter(ABC): + """平台适配器基类""" + + @abstractmethod + def get_content_type(self, url: str) -> str: + """识别内容类型 (video/article/answer/zhuanlan)""" + pass + + @abstractmethod + async def fetch_metadata(self, content_id: str, content_type: str) -> Dict[str, Any]: + """获取元数据""" + pass + + @abstractmethod + async def fetch_content(self, content_id: str, content_type: str) -> str: + """获取正文内容(Markdown格式)""" + pass + + @abstractmethod + def get_output_filename(self, content_id: str, content_type: str, suffix: str) -> str: + """生成输出文件名""" + pass + ``` + + 3. 创建测试目录和 fixtures: + - tests/test_platforms/__init__.py (空文件) + - tests/test_platforms/conftest.py: 定义共享 fixtures (mock_api_response, tmp_content_dir) + + + pytest tests/test_platforms/ -x -v -k "base" + + PlatformAdapter 基类定义完成,子类接口明确 + + + + Task 2: Implement BilibiliAdapter for video metadata + kb_pipeline/platforms/bilibili.py, tests/test_platforms/test_bilibili.py + + - Test 1: fetch_metadata("BVxxx", "video") 返回正确的视频元数据 + - Test 2: 403错误时触发降级策略(API -> Playwright) + - Test 3: 网络错误自动重试3次 + - Test 4: get_output_filename 返回正确格式 "bilibili_BVxxx_metadata.json" + + + 创建 kb_pipeline/platforms/bilibili.py: + ```python + from kb_pipeline.platforms.base import PlatformAdapter + from kb_pipeline.errors import NetworkError, AuthError + from kb_pipeline.retry import retry_on_network_error + from kb_bilibili_metadata.bilibili_extractor import BilibiliExtractor + + class BilibiliAdapter(PlatformAdapter): + PLATFORM = "bilibili" + + def __init__(self, headless: bool = True, timeout: int = 30): + self.extractor = BilibiliExtractor(headless=headless, timeout=timeout) + + def get_content_type(self, url: str) -> str: + # 从 URL 识别 video 或 article + + @retry_on_network_error + async def fetch_metadata(self, content_id: str, content_type: str) -> Dict[str, Any]: + if content_type == "video": + return self.extractor.extract_video_metadata(content_id) + elif content_type == "article": + return self.extractor.extract_article_metadata(content_id) + raise ValueError(f"Unsupported content type: {content_type}") + + async def fetch_content(self, content_id: str, content_type: str) -> str: + # 视频内容获取留给 Phase 3 (ASR) + # 专栏内容需要额外实现 + pass + + def get_output_filename(self, content_id: str, content_type: str, suffix: str) -> str: + return f"bilibili_{content_id}_{suffix}" + ``` + + 创建 tests/test_platforms/test_bilibili.py: + - 使用 mock 模拟 BilibiliExtractor 的返回值 + - 测试视频元数据提取 + - 测试错误处理和重试机制 + + + pytest tests/test_platforms/test_bilibili.py::test_video_metadata -x -v + + BilibiliAdapter 能正确提取B站视频元数据,支持错误重试 + + + + Task 3: Implement BilibiliAdapter for article content + kb_pipeline/platforms/bilibili.py, tests/test_platforms/test_bilibili.py + + - Test 1: fetch_metadata("cvxxx", "article") 返回正确的专栏元数据 + - Test 2: fetch_content("cvxxx", "article") 返回 Markdown 格式的正文 + - Test 3: Markdown 包含 YAML Frontmatter (title, author, source_url, platform, content_type, fetched_at) + + + 扩展 kb_pipeline/platforms/bilibili.py: + 1. 实现 fetch_content 方法: + - 使用 BilibiliExtractor 获取专栏内容 + - 转换为 Markdown 格式 + - 添加 YAML Frontmatter + + 2. 添加 format_content_markdown 辅助函数: + ```python + import yaml + from datetime import datetime + + def format_content_markdown(metadata: dict, content: str, platform: str, content_type: str) -> str: + frontmatter = { + "title": metadata.get("title", ""), + "author": metadata.get("author", ""), + "source_url": metadata.get("url", ""), + "platform": platform, + "content_type": content_type, + "fetched_at": datetime.now().isoformat(), + } + return f"---\n{yaml.dump(frontmatter, allow_unicode=True)}---\n\n{content}" + ``` + + 3. 扩展测试覆盖专栏场景 + + + pytest tests/test_platforms/test_bilibili.py::test_article_content -x -v + + BilibiliAdapter 能正确提取B站专栏内容和元数据 + + + + Task 4: Update package exports + kb_pipeline/__init__.py + + 更新 kb_pipeline/__init__.py: + ```python + from .platforms import PlatformAdapter, BilibiliAdapter + + __all__ = [ + # ... existing exports ... + "PlatformAdapter", + "BilibiliAdapter", + ] + ``` + + + python -c "from kb_pipeline import BilibiliAdapter; print('OK')" + + BilibiliAdapter 可从 kb_pipeline 包根导入 + + + + + +- 所有测试通过: `pytest tests/test_platforms/test_bilibili.py -v` +- 导入测试: `python -c "from kb_pipeline import BilibiliAdapter"` +- 覆盖率: `pytest tests/test_platforms/ --cov=kb_pipeline/platforms --cov-report=term-missing` + + + +- [ ] PlatformAdapter 基类定义清晰,子类接口明确 +- [ ] BilibiliAdapter 正确提取B站视频元数据 +- [ ] BilibiliAdapter 正确提取B站专栏内容和元数据 +- [ ] 内容保存为 Markdown 格式,包含 YAML Frontmatter +- [ ] 网络错误自动重试3次 +- [ ] 所有测试通过(至少 8 个测试用例) + + + +After completion, create `.planning/phases/02-platform-skills/02-02-SUMMARY.md` + diff --git a/.planning/phases/02-platform-skills/02-03-PLAN.md b/.planning/phases/02-platform-skills/02-03-PLAN.md new file mode 100644 index 0000000..87c5db8 --- /dev/null +++ b/.planning/phases/02-platform-skills/02-03-PLAN.md @@ -0,0 +1,302 @@ +--- +phase: 02-platform-skills +plan: 03 +type: execute +wave: 2 +depends_on: [02-01, 02-02] +files_modified: + - kb_pipeline/platforms/zhihu.py + - kb_pipeline/platforms/__init__.py + - kb_pipeline/__init__.py + - tests/test_platforms/test_zhihu.py +autonomous: true +requirements: [PLAT-03] +user_setup: + - service: zhihu + why: "知乎内容获取需要登录 Cookie" + env_vars: + - name: KB_ZHIHU_COOKIE + source: "浏览器开发者工具 -> Application -> Cookies -> zhihu.com -> 复制完整 Cookie 字符串" + dashboard_config: + - task: "获取知乎 Cookie" + location: "登录知乎后,在浏览器开发者工具中获取 Cookie" + +must_haves: + truths: + - "输入知乎问答URL,获取问题标题、作者、回答内容" + - "输入知乎专栏URL,获取文章标题、作者、正文内容" + - "内容保存为Markdown文件(YAML Frontmatter + 正文)" + - "Cookie缺失时返回 AuthError 并提示配置" + - "检测到登录页面时返回 AuthError 并提示更新 Cookie" + artifacts: + - path: "kb_pipeline/platforms/zhihu.py" + provides: "知乎平台适配器" + exports: ["ZhihuAdapter"] + min_lines: 100 + - path: "tests/test_platforms/test_zhihu.py" + provides: "知乎适配器测试" + min_lines: 60 + key_links: + - from: "kb_pipeline/platforms/zhihu.py" + to: "kb_pipeline/config.py" + via: "get_settings().zhihu_cookie" + pattern: "get_settings().zhihu_cookie" + - from: "kb_pipeline/platforms/zhihu.py" + to: "kb-zhihu-metadata/zhihu_extractor.py" + via: "ZhihuPlaywrightExtractor class" + pattern: "from kb_zhihu_metadata" + - from: "kb_pipeline/platforms/zhihu.py" + to: "kb_pipeline/errors.py" + via: "AuthError for cookie issues" + pattern: "raise AuthError" +--- + + +创建知乎平台适配器,封装问答和专栏的内容获取逻辑,复用现有 ZhihuPlaywrightExtractor 实现。 + +Purpose: 为知乎问答和专栏提供统一的内容提取接口,支持 Cookie 认证和错误处理。 +Output: ZhihuAdapter 实现类 + + + +@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md +@/home/starmen1265/.claude/get-shit-done/templates/summary.md + + + +@.planning/PROJECT.md +@.planning/ROADMAP.md +@.planning/STATE.md +@.planning/phases/02-platform-skills/02-CONTEXT.md +@.planning/phases/02-platform-skills/02-RESEARCH.md + + + + +From kb_pipeline/platforms/base.py (Plan 02): +```python +class PlatformAdapter(ABC): + @abstractmethod + def get_content_type(self, url: str) -> str: pass + + @abstractmethod + async def fetch_metadata(self, content_id: str, content_type: str) -> Dict[str, Any]: pass + + @abstractmethod + async def fetch_content(self, content_id: str, content_type: str) -> str: pass + + @abstractmethod + def get_output_filename(self, content_id: str, content_type: str, suffix: str) -> str: pass +``` + +From kb_pipeline/config.py: +```python +class KBSettings(BaseSettings): + zhihu_cookie: SecretStr = Field(default=SecretStr(""), description="知乎 Cookie") +``` + +From kb_pipeline/errors.py: +```python +class AuthError(KBError): + def __init__(self, step_name, input_summary, auth_type=None, cause=None): + # suggestion: "检查 API Token 或 Cookie 是否有效" +``` + +From kb_pipeline/link_parser.py (Plan 01): +```python +@dataclass +class ParsedLink: + platform: str + content_type: str # "answer" | "zhuanlan" + content_id: str +``` + + + + + +From kb-zhihu-metadata/zhihu_extractor.py: +```python +class ZhihuPlaywrightExtractor: + 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=True, timeout=30, cookie=None): + self.cookie = cookie + self.browser = PlaywrightBrowser(headless=headless, timeout=timeout) + + def extract(self, url: str) -> Optional[Dict[str, Any]]: + # 返回 {type, title, author, content, content_markdown, publish_time, ...} + + def _extract_answer_content(self) -> Dict[str, Any]: ... + def _extract_zhuanlan_content(self) -> Dict[str, Any]: ... +``` + + + + + + + Task 1: Implement ZhihuAdapter for answer content + kb_pipeline/platforms/zhihu.py, tests/test_platforms/test_zhihu.py + + - Test 1: fetch_metadata("xxx", "answer") 返回正确的问答元数据 + - Test 2: fetch_content("xxx", "answer") 返回 Markdown 格式的回答内容 + - Test 3: Cookie 缺失时返回 AuthError + - Test 4: get_output_filename 返回正确格式 "zhihu_xxx_metadata.json" + + + 创建 kb_pipeline/platforms/zhihu.py: + ```python + from kb_pipeline.platforms.base import PlatformAdapter + from kb_pipeline.errors import AuthError, NetworkError + from kb_pipeline.config import get_settings + from kb_zhihu_metadata.zhihu_extractor import ZhihuPlaywrightExtractor + + class ZhihuAdapter(PlatformAdapter): + PLATFORM = "zhihu" + + def __init__(self, headless: bool = True, timeout: int = 30): + self.headless = headless + self.timeout = timeout + self._cookie = None + + @property + def cookie(self) -> str: + if self._cookie is None: + settings = get_settings() + self._cookie = settings.zhihu_cookie.get_secret_value() + return self._cookie + + def get_content_type(self, url: str) -> str: + # 从 URL 识别 answer 或 zhuanlan + if "answer" in url: + return "answer" + elif "zhuanlan" in url: + return "zhuanlan" + return "unknown" + + async def fetch_metadata(self, content_id: str, content_type: str) -> Dict[str, Any]: + if not self.cookie: + raise AuthError( + step_name="zhihu_metadata", + input_summary=f"content_id={content_id}", + auth_type="zhihu_cookie", + ) + # 使用 ZhihuPlaywrightExtractor 提取 + pass + + async def fetch_content(self, content_id: str, content_type: str) -> str: + # 返回 Markdown 格式内容 + pass + + def get_output_filename(self, content_id: str, content_type: str, suffix: str) -> str: + return f"zhihu_{content_id}_{suffix}" + ``` + + 创建 tests/test_platforms/test_zhihu.py: + - Mock ZhihuPlaywrightExtractor 的返回值 + - 测试 Cookie 缺失时的 AuthError + - 测试问答内容提取 + + + pytest tests/test_platforms/test_zhihu.py::test_answer_content -x -v + + ZhihuAdapter 能正确提取知乎问答内容,Cookie 缺失时返回 AuthError + + + + Task 2: Implement ZhihuAdapter for zhuanlan content + kb_pipeline/platforms/zhihu.py, tests/test_platforms/test_zhihu.py + + - Test 1: fetch_metadata("xxx", "zhuanlan") 返回正确的专栏元数据 + - Test 2: fetch_content("xxx", "zhuanlan") 返回 Markdown 格式的文章内容 + - Test 3: 检测到登录页面时返回 AuthError 并提示更新 Cookie + + + 扩展 kb_pipeline/platforms/zhihu.py: + 1. 实现 zhuanlan 类型的 fetch_metadata 和 fetch_content + 2. 添加登录页面检测逻辑: + ```python + def _detect_login_page(self, result: Dict) -> bool: + """检测是否被重定向到登录页面""" + if not result: + return False + title = result.get("title", "") + content = result.get("content", "") + return "登录" in title or len(content) < 100 + ``` + 3. 登录页面检测时抛出 AuthError: + ```python + if self._detect_login_page(result): + raise AuthError( + step_name="zhihu_content", + input_summary=f"content_id={content_id}", + auth_type="zhihu_cookie", + cause=Exception("Cookie 可能已过期,请更新 KB_ZHIHU_COOKIE") + ) + ``` + + 4. 扩展测试覆盖专栏场景 + + + pytest tests/test_platforms/test_zhihu.py::test_zhuanlan_content -x -v + + ZhihuAdapter 能正确提取知乎专栏内容,检测登录页面时返回 AuthError + + + + Task 3: Update package exports + kb_pipeline/__init__.py, kb_pipeline/platforms/__init__.py + + 1. 更新 kb_pipeline/platforms/__init__.py: + ```python + from .base import PlatformAdapter + from .bilibili import BilibiliAdapter + from .zhihu import ZhihuAdapter + + __all__ = ["PlatformAdapter", "BilibiliAdapter", "ZhihuAdapter"] + ``` + + 2. 更新 kb_pipeline/__init__.py: + ```python + from .platforms import PlatformAdapter, BilibiliAdapter, ZhihuAdapter + + __all__ = [ + # ... existing exports ... + "PlatformAdapter", + "BilibiliAdapter", + "ZhihuAdapter", + ] + ``` + + + python -c "from kb_pipeline import ZhihuAdapter; print('OK')" + + ZhihuAdapter 可从 kb_pipeline 包根导入 + + + + + +- 所有测试通过: `pytest tests/test_platforms/test_zhihu.py -v` +- 导入测试: `python -c "from kb_pipeline import ZhihuAdapter"` +- 覆盖率: `pytest tests/test_platforms/ --cov=kb_pipeline/platforms --cov-report=term-missing` + + + +- [ ] ZhihuAdapter 正确提取知乎问答内容 +- [ ] ZhihuAdapter 正确提取知乎专栏内容 +- [ ] Cookie 缺失时返回 AuthError +- [ ] 检测到登录页面时返回 AuthError 并提示更新 Cookie +- [ ] 内容保存为 Markdown 格式,包含 YAML Frontmatter +- [ ] 所有测试通过(至少 6 个测试用例) + + + +After completion, create `.planning/phases/02-platform-skills/02-03-SUMMARY.md` +