docs(02): create phase 2 plans for platform skills
This commit is contained in:
@@ -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)*
|
||||
|
||||
@@ -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"
|
||||
---
|
||||
|
||||
<objective>
|
||||
创建统一的 LinkParser 类和幂等检查机制，为后续平台适配器提供链接解析和重复处理防护能力。
|
||||
|
||||
Purpose: 为所有平台提供统一的 URL 解析入口，支持 B站/知乎的内容类型识别和 ID 提取，同时提供幂等检查避免重复处理。
|
||||
Output: LinkParser 类、ParsedLink 数据类、IdempotencyChecker 类
|
||||
</objective>
|
||||
|
||||
<execution_context>
|
||||
@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
|
||||
@/home/starmen1265/.claude/get-shit-done/templates/summary.md
|
||||
</execution_context>
|
||||
|
||||
<context>
|
||||
@.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
|
||||
|
||||
<interfaces>
|
||||
<!-- Phase 1 基础设施 - LinkParser 将复用这些 -->
|
||||
|
||||
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),
|
||||
)
|
||||
```
|
||||
</interfaces>
|
||||
|
||||
<existing_code>
|
||||
<!-- 现有 link-parser.py 的核心模式，需要整合到 kb_pipeline/link_parser.py -->
|
||||
|
||||
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}
|
||||
```
|
||||
</existing_code>
|
||||
</context>
|
||||
|
||||
<tasks>
|
||||
|
||||
<task type="auto" tdd="true">
|
||||
<name>Task 1: Write interface contracts and tests for LinkParser</name>
|
||||
<files>kb_pipeline/link_parser.py, tests/test_pipeline/test_link_parser.py</files>
|
||||
<behavior>
|
||||
- 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（不抛异常）
|
||||
</behavior>
|
||||
<action>
|
||||
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 类型
|
||||
</action>
|
||||
<verify>
|
||||
<automated>pytest tests/test_pipeline/test_link_parser.py -x -v</automated>
|
||||
</verify>
|
||||
<done>LinkParser 能正确解析所有支持的 URL 类型，无法识别的 URL 返回 None</done>
|
||||
</task>
|
||||
|
||||
<task type="auto" tdd="true">
|
||||
<name>Task 2: Write interface contracts and tests for IdempotencyChecker</name>
|
||||
<files>kb_pipeline/idempotency.py, tests/test_pipeline/test_idempotency.py</files>
|
||||
<behavior>
|
||||
- Test 1: 文件存在时 check_exists 返回 (True, [file_list])
|
||||
- Test 2: 文件不存在时 check_exists 返回 (False, [])
|
||||
- Test 3: 部分文件存在时返回正确的存在文件列表
|
||||
</behavior>
|
||||
<action>
|
||||
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 创建临时文件
|
||||
- 测试文件存在/不存在的各种场景
|
||||
- 测试部分文件存在的情况
|
||||
</action>
|
||||
<verify>
|
||||
<automated>pytest tests/test_pipeline/test_idempotency.py -x -v</automated>
|
||||
</verify>
|
||||
<done>IdempotencyChecker 能正确检查文件是否存在，返回准确的文件列表</done>
|
||||
</task>
|
||||
|
||||
<task type="auto">
|
||||
<name>Task 3: Update package exports</name>
|
||||
<files>kb_pipeline/__init__.py</files>
|
||||
<action>
|
||||
更新 kb_pipeline/__init__.py，导出新模块:
|
||||
```python
|
||||
from .link_parser import LinkParser, ParsedLink
|
||||
from .idempotency import IdempotencyChecker
|
||||
|
||||
__all__ = [
|
||||
# ... existing exports ...
|
||||
"LinkParser",
|
||||
"ParsedLink",
|
||||
"IdempotencyChecker",
|
||||
]
|
||||
```
|
||||
</action>
|
||||
<verify>
|
||||
<automated>python -c "from kb_pipeline import LinkParser, ParsedLink, IdempotencyChecker; print('OK')"</automated>
|
||||
</verify>
|
||||
<done>LinkParser, ParsedLink, IdempotencyChecker 可从 kb_pipeline 包根导入</done>
|
||||
</task>
|
||||
|
||||
</tasks>
|
||||
|
||||
<verification>
|
||||
- 所有测试通过: `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`
|
||||
</verification>
|
||||
|
||||
<success_criteria>
|
||||
- [ ] LinkParser 正确解析 B站视频/专栏、知乎问答/专栏 URL
|
||||
- [ ] b23.tv 短链自动解析到真实 URL
|
||||
- [ ] 无法识别的 URL 返回 None（不抛异常）
|
||||
- [ ] IdempotencyChecker 正确检查文件存在性
|
||||
- [ ] 所有测试通过（至少 10 个测试用例）
|
||||
- [ ] 公共 API 可从 kb_pipeline 包根导入
|
||||
</success_criteria>
|
||||
|
||||
<output>
|
||||
After completion, create `.planning/phases/02-platform-skills/02-01-SUMMARY.md`
|
||||
</output>
|
||||
@@ -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"
|
||||
---
|
||||
|
||||
<objective>
|
||||
创建 B站平台适配器，封装视频和专栏的元数据提取与内容获取逻辑，复用现有 BilibiliExtractor 实现。
|
||||
|
||||
Purpose: 为 B站视频和专栏提供统一的元数据提取和内容下载接口，支持错误重试和降级策略。
|
||||
Output: PlatformAdapter 基类、BilibiliAdapter 实现类
|
||||
</objective>
|
||||
|
||||
<execution_context>
|
||||
@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
|
||||
@/home/starmen1265/.claude/get-shit-done/templates/summary.md
|
||||
</execution_context>
|
||||
|
||||
<context>
|
||||
@.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
|
||||
|
||||
<interfaces>
|
||||
<!-- Phase 1 和 Plan 01 提供的接口 -->
|
||||
|
||||
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
|
||||
```
|
||||
</interfaces>
|
||||
|
||||
<existing_code>
|
||||
<!-- 现有 BilibiliExtractor 的核心方法 -->
|
||||
|
||||
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/',
|
||||
}
|
||||
```
|
||||
</existing_code>
|
||||
</context>
|
||||
|
||||
<tasks>
|
||||
|
||||
<task type="auto" tdd="true">
|
||||
<name>Task 1: Write interface contracts for PlatformAdapter base class</name>
|
||||
<files>kb_pipeline/platforms/__init__.py, kb_pipeline/platforms/base.py, tests/test_platforms/__init__.py, tests/test_platforms/conftest.py</files>
|
||||
<behavior>
|
||||
- Test 1: PlatformAdapter 是抽象基类，不能直接实例化
|
||||
- Test 2: 子类必须实现 get_content_type, fetch_metadata, fetch_content, get_output_filename 方法
|
||||
</behavior>
|
||||
<action>
|
||||
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)
|
||||
</action>
|
||||
<verify>
|
||||
<automated>pytest tests/test_platforms/ -x -v -k "base"</automated>
|
||||
</verify>
|
||||
<done>PlatformAdapter 基类定义完成，子类接口明确</done>
|
||||
</task>
|
||||
|
||||
<task type="auto" tdd="true">
|
||||
<name>Task 2: Implement BilibiliAdapter for video metadata</name>
|
||||
<files>kb_pipeline/platforms/bilibili.py, tests/test_platforms/test_bilibili.py</files>
|
||||
<behavior>
|
||||
- Test 1: fetch_metadata("BVxxx", "video") 返回正确的视频元数据
|
||||
- Test 2: 403错误时触发降级策略（API -> Playwright）
|
||||
- Test 3: 网络错误自动重试3次
|
||||
- Test 4: get_output_filename 返回正确格式 "bilibili_BVxxx_metadata.json"
|
||||
</behavior>
|
||||
<action>
|
||||
创建 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 的返回值
|
||||
- 测试视频元数据提取
|
||||
- 测试错误处理和重试机制
|
||||
</action>
|
||||
<verify>
|
||||
<automated>pytest tests/test_platforms/test_bilibili.py::test_video_metadata -x -v</automated>
|
||||
</verify>
|
||||
<done>BilibiliAdapter 能正确提取B站视频元数据，支持错误重试</done>
|
||||
</task>
|
||||
|
||||
<task type="auto" tdd="true">
|
||||
<name>Task 3: Implement BilibiliAdapter for article content</name>
|
||||
<files>kb_pipeline/platforms/bilibili.py, tests/test_platforms/test_bilibili.py</files>
|
||||
<behavior>
|
||||
- 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)
|
||||
</behavior>
|
||||
<action>
|
||||
扩展 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. 扩展测试覆盖专栏场景
|
||||
</action>
|
||||
<verify>
|
||||
<automated>pytest tests/test_platforms/test_bilibili.py::test_article_content -x -v</automated>
|
||||
</verify>
|
||||
<done>BilibiliAdapter 能正确提取B站专栏内容和元数据</done>
|
||||
</task>
|
||||
|
||||
<task type="auto">
|
||||
<name>Task 4: Update package exports</name>
|
||||
<files>kb_pipeline/__init__.py</files>
|
||||
<action>
|
||||
更新 kb_pipeline/__init__.py:
|
||||
```python
|
||||
from .platforms import PlatformAdapter, BilibiliAdapter
|
||||
|
||||
__all__ = [
|
||||
# ... existing exports ...
|
||||
"PlatformAdapter",
|
||||
"BilibiliAdapter",
|
||||
]
|
||||
```
|
||||
</action>
|
||||
<verify>
|
||||
<automated>python -c "from kb_pipeline import BilibiliAdapter; print('OK')"</automated>
|
||||
</verify>
|
||||
<done>BilibiliAdapter 可从 kb_pipeline 包根导入</done>
|
||||
</task>
|
||||
|
||||
</tasks>
|
||||
|
||||
<verification>
|
||||
- 所有测试通过: `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`
|
||||
</verification>
|
||||
|
||||
<success_criteria>
|
||||
- [ ] PlatformAdapter 基类定义清晰，子类接口明确
|
||||
- [ ] BilibiliAdapter 正确提取B站视频元数据
|
||||
- [ ] BilibiliAdapter 正确提取B站专栏内容和元数据
|
||||
- [ ] 内容保存为 Markdown 格式，包含 YAML Frontmatter
|
||||
- [ ] 网络错误自动重试3次
|
||||
- [ ] 所有测试通过（至少 8 个测试用例）
|
||||
</success_criteria>
|
||||
|
||||
<output>
|
||||
After completion, create `.planning/phases/02-platform-skills/02-02-SUMMARY.md`
|
||||
</output>
|
||||
@@ -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"
|
||||
---
|
||||
|
||||
<objective>
|
||||
创建知乎平台适配器，封装问答和专栏的内容获取逻辑，复用现有 ZhihuPlaywrightExtractor 实现。
|
||||
|
||||
Purpose: 为知乎问答和专栏提供统一的内容提取接口，支持 Cookie 认证和错误处理。
|
||||
Output: ZhihuAdapter 实现类
|
||||
</objective>
|
||||
|
||||
<execution_context>
|
||||
@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
|
||||
@/home/starmen1265/.claude/get-shit-done/templates/summary.md
|
||||
</execution_context>
|
||||
|
||||
<context>
|
||||
@.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
|
||||
|
||||
<interfaces>
|
||||
<!-- Phase 1 和 Plan 01/02 提供的接口 -->
|
||||
|
||||
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
|
||||
```
|
||||
</interfaces>
|
||||
|
||||
<existing_code>
|
||||
<!-- 现有 ZhihuPlaywrightExtractor 的核心方法 -->
|
||||
|
||||
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]: ...
|
||||
```
|
||||
</existing_code>
|
||||
</context>
|
||||
|
||||
<tasks>
|
||||
|
||||
<task type="auto" tdd="true">
|
||||
<name>Task 1: Implement ZhihuAdapter for answer content</name>
|
||||
<files>kb_pipeline/platforms/zhihu.py, tests/test_platforms/test_zhihu.py</files>
|
||||
<behavior>
|
||||
- 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"
|
||||
</behavior>
|
||||
<action>
|
||||
创建 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
|
||||
- 测试问答内容提取
|
||||
</action>
|
||||
<verify>
|
||||
<automated>pytest tests/test_platforms/test_zhihu.py::test_answer_content -x -v</automated>
|
||||
</verify>
|
||||
<done>ZhihuAdapter 能正确提取知乎问答内容，Cookie 缺失时返回 AuthError</done>
|
||||
</task>
|
||||
|
||||
<task type="auto" tdd="true">
|
||||
<name>Task 2: Implement ZhihuAdapter for zhuanlan content</name>
|
||||
<files>kb_pipeline/platforms/zhihu.py, tests/test_platforms/test_zhihu.py</files>
|
||||
<behavior>
|
||||
- Test 1: fetch_metadata("xxx", "zhuanlan") 返回正确的专栏元数据
|
||||
- Test 2: fetch_content("xxx", "zhuanlan") 返回 Markdown 格式的文章内容
|
||||
- Test 3: 检测到登录页面时返回 AuthError 并提示更新 Cookie
|
||||
</behavior>
|
||||
<action>
|
||||
扩展 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. 扩展测试覆盖专栏场景
|
||||
</action>
|
||||
<verify>
|
||||
<automated>pytest tests/test_platforms/test_zhihu.py::test_zhuanlan_content -x -v</automated>
|
||||
</verify>
|
||||
<done>ZhihuAdapter 能正确提取知乎专栏内容，检测登录页面时返回 AuthError</done>
|
||||
</task>
|
||||
|
||||
<task type="auto">
|
||||
<name>Task 3: Update package exports</name>
|
||||
<files>kb_pipeline/__init__.py, kb_pipeline/platforms/__init__.py</files>
|
||||
<action>
|
||||
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",
|
||||
]
|
||||
```
|
||||
</action>
|
||||
<verify>
|
||||
<automated>python -c "from kb_pipeline import ZhihuAdapter; print('OK')"</automated>
|
||||
</verify>
|
||||
<done>ZhihuAdapter 可从 kb_pipeline 包根导入</done>
|
||||
</task>
|
||||
|
||||
</tasks>
|
||||
|
||||
<verification>
|
||||
- 所有测试通过: `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`
|
||||
</verification>
|
||||
|
||||
<success_criteria>
|
||||
- [ ] ZhihuAdapter 正确提取知乎问答内容
|
||||
- [ ] ZhihuAdapter 正确提取知乎专栏内容
|
||||
- [ ] Cookie 缺失时返回 AuthError
|
||||
- [ ] 检测到登录页面时返回 AuthError 并提示更新 Cookie
|
||||
- [ ] 内容保存为 Markdown 格式，包含 YAML Frontmatter
|
||||
- [ ] 所有测试通过（至少 6 个测试用例）
|
||||
</success_criteria>
|
||||
|
||||
<output>
|
||||
After completion, create `.planning/phases/02-platform-skills/02-03-SUMMARY.md`
|
||||
</output>
|
||||
Reference in New Issue
Block a user