19 KiB
Phase 2: Platform Skills - Research
Researched: 2026-03-13 Domain: Python / B站知乎API集成 / 内容获取管道 Confidence: HIGH
Summary
Phase 2 需要集成B站和知乎平台的内容获取能力,形成从链接到原始内容的完整流程。项目已有完整的平台API实现(kb-bilibili-metadata、kb-zhihu-metadata),以及链接解析器(kb-link-parser)。Phase 1 已建立了核心管道基础设施(schemas.py、errors.py、config.py、retry.py)。
Primary recommendation: 复用现有平台提取器实现,通过统一 LinkParser 类整合链接解析,使用 Phase 1 的 StepInput/StepOutput 模式包装现有功能,确保幂等检查和网络重试机制。
<user_constraints>
User Constraints (from CONTEXT.md)
Locked Decisions
链接解析策略
- 解析器结构: 统一 LinkParser 类 — 一个模块处理所有平台,通过正则/模式匹配区分
- 支持 URL 类型:
- B站视频: bilibili.com/video/BVxxx, b23.tv/xxx(短链解析)
- B站专栏: bilibili.com/read/cvxxx
- 知乎问答: zhihu.com/question/xxx/answer/xxx
- 知乎专栏: zhuanlan.zhihu.com/p/xxx
- ID 提取: BV号 + AV号(同时支持,自动转换)
- 短链支持: 支持 b23.tv 短链解析,获取真实 URL
- 错误处理: 无法识别的 URL 返回 None + 错误信息(不抛异常)
元数据输出格式
- B站视频字段: 标题、作者、时长、发布时间、描述、封面、链接、ID(BV号)
- 知乎字段: 标题、作者、发布时间、内容摘要、链接、ID
- 文件命名: 平台_ID.json(如 bilibili_BV1xx.json, zhihu_12345.json)
- 存储位置: 内容目录下(与内容文件同目录)
内容输出格式
- Markdown 结构: YAML Frontmatter + 正文内容
- Frontmatter 字段: title, author, source_url, platform, content_type, fetched_at
- 图片处理: 保留原始链接(不下载到本地)
- 格式保留: 基础格式(标题、段落、列表、代码块)
- 不保留: 表格、数学公式、LaTeX(Phase 2 不处理)
幂等检查机制
- 检查时机: 链接解析后 — 解析成功后立即检查目录是否存在
- 判断依据: 检查输出文件是否存在(元数据 JSON + 内容 Markdown)
- 重复处理: 询问用户 — 发现已存在文件时询问是否重新下载
Claude's Discretion
- LinkParser 的正则表达式具体实现
- 文件名清理规则(特殊字符处理)
- 网络请求超时时间和重试次数(复用 Phase 1 的 retry 机制)
Deferred Ideas (OUT OF SCOPE)
- B站/知乎用户主页跟踪 — 未来功能
- 批量链接处理 — 当前只处理单个链接
- 进度显示 — v2 考虑
</user_constraints>
<phase_requirements>
Phase Requirements
| ID | Description | Research Support |
|---|---|---|
| PLAT-01 | B站视频处理 - 支持 B站视频的元数据提取、字幕下载、语音识别、深度分析 | 复用 kb-bilibili-metadata/bilibili_extractor.py 的 BilibiliExtractor 类,支持视频元数据和专栏提取 |
| PLAT-02 | B站专栏处理 - 支持 B站专栏的文字提取和深度分析 | 复用 kb-bilibili-content/bilibili_content.py 的 extract_article() 方法 |
| PLAT-03 | 知乎问答/专栏处理 - 支持知乎问答和专栏的文字提取和深度分析 | 复用 kb-zhihu-metadata/zhihu_extractor.py 的 ZhihuPlaywrightExtractor 类 |
</phase_requirements>
Standard Stack
Core
| Library | Version | Purpose | Why Standard |
|---|---|---|---|
| pydantic | v2 | Schema 验证 | Phase 1 已采用,类型安全,自动验证 |
| pydantic-settings | v2 | 配置管理 | Phase 1 已采用,统一从 ~/.kb_env 加载 |
| tenacity | 8.x | 重试机制 | Phase 1 已采用,指数退避,装饰器模式 |
| httpx | 0.27+ | HTTP 客户端 | 现有代码已使用,支持异步,比 requests 更现代 |
| aiohttp | 3.9+ | 异步 HTTP | 现有代码已使用,用于 B站 API 调用 |
Supporting
| Library | Version | Purpose | When to Use |
|---|---|---|---|
| playwright | 1.40+ | 浏览器自动化 | 知乎内容提取、B站专栏全文获取 |
| yt-dlp | latest | 视频下载 | B站视频下载(替代 bilibili-dl) |
| requests | 2.31+ | 同步 HTTP | 短链解析、简单 API 调用 |
Alternatives Considered
| Instead of | Could Use | Tradeoff |
|---|---|---|
| playwright | selenium | Playwright 更现代,API 更友好,异步支持更好 |
| httpx | requests | httpx 支持异步,更适合高并发场景;requests 更简单稳定 |
Installation:
pip install pydantic pydantic-settings tenacity httpx aiohttp playwright yt-dlp requests
playwright install chromium # Playwright 浏览器安装
Architecture Patterns
Recommended Project Structure
kb_pipeline/
├── __init__.py # 导出公共 API
├── schemas.py # StepInput/StepOutput/WorkflowState(Phase 1 已完成)
├── errors.py # KBError 体系(Phase 1 已完成)
├── config.py # KBSettings 配置(Phase 1 已完成)
├── retry.py # retry_on_network_error(Phase 1 已完成)
├── state.py # WorkflowStateManager(Phase 1 已完成)
├── link_parser.py # 统一链接解析器(Phase 2 新增)
├── platforms/ # 平台适配器(Phase 2 新增)
│ ├── __init__.py
│ ├── base.py # 平台适配器基类
│ ├── bilibili.py # B站适配器
│ └── zhihu.py # 知乎适配器
└── content_fetcher.py # 内容获取协调器(Phase 2 新增)
Pattern 1: 平台适配器模式
What: 为每个平台创建统一的适配器接口,封装元数据提取和内容获取逻辑 When to use: 需要支持多个内容平台时 Example:
# Source: 项目设计模式
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional
from kb_pipeline.schemas import StepInput, StepOutput
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) -> str:
"""生成输出文件名"""
pass
Pattern 2: 链接解析统一入口
What: 统一 LinkParser 类处理所有平台的 URL 解析 When to use: 任何需要从 URL 提取平台和内容 ID 的场景 Example:
# Source: 基于现有 kb-link-parser/link-parser.py 重构
import re
from typing import Optional, Dict, Any
from dataclasses import dataclass
@dataclass
class ParsedLink:
"""解析后的链接信息"""
platform: str # "bilibili" | "zhihu"
content_type: str # "video" | "article" | "answer" | "zhuanlan"
content_id: str # BV号、CV号、知乎ID等
original_url: str # 原始 URL
resolved_url: str # 解析后的完整 URL(短链展开后)
class LinkParser:
"""统一链接解析器"""
# B站正则模式
BILIBILI_PATTERNS = {
"video": re.compile(r'bilibili\.com/video/(BV[a-zA-Z0-9]+)'),
"article": re.compile(r'bilibili\.com/read/cv(\d+)'),
"short": re.compile(r'b23\.tv/([a-zA-Z0-9]+)'),
}
# 知乎正则模式
ZHIHU_PATTERNS = {
"answer": re.compile(r'zhihu\.com/question/\d+/answer/(\d+)'),
"zhuanlan": re.compile(r'zhuanlan\.zhihu\.com/p/(\d+)'),
"zvideo": re.compile(r'zhihu\.com/zvideo/(\d+)'),
}
def parse(self, url: str) -> Optional[ParsedLink]:
"""解析 URL,返回结构化信息或 None"""
# 1. 先处理短链
resolved = self._resolve_short_link(url)
# 2. 匹配平台模式
for platform, patterns in [("bilibili", self.BILIBILI_PATTERNS),
("zhihu", self.ZHIHU_PATTERNS)]:
for content_type, pattern in patterns.items():
match = pattern.search(resolved)
if match:
return ParsedLink(
platform=platform,
content_type=content_type,
content_id=match.group(1),
original_url=url,
resolved_url=resolved
)
return None # 无法识别,返回 None(不抛异常)
def _resolve_short_link(self, url: str) -> str:
"""解析短链"""
# 复用现有 link_parser_shortlink.py 的逻辑
if "b23.tv" in url:
try:
import requests
resp = requests.head(url, allow_redirects=True, timeout=10)
return resp.url
except Exception:
pass
return url
Pattern 3: 幂等检查机制
What: 在执行前检查输出文件是否已存在 When to use: 避免重复下载和处理 Example:
# Source: CONTEXT.md 决策
from pathlib import Path
from typing import Tuple
class IdempotencyChecker:
"""幂等检查器"""
def check_exists(self, output_dir: Path, content_id: str) -> Tuple[bool, list[Path]]:
"""检查文件是否已存在
Returns:
(exists, files): 是否存在,存在的文件列表
"""
metadata_file = output_dir / f"{content_id}_metadata.json"
content_file = output_dir / f"{content_id}_content.md"
existing = []
if metadata_file.exists():
existing.append(metadata_file)
if content_file.exists():
existing.append(content_file)
return len(existing) > 0, existing
def prompt_redownload(self, existing_files: list[Path]) -> bool:
"""询问用户是否重新下载
Returns:
True 表示重新下载,False 表示跳过
"""
file_list = "\n".join(f" - {f.name}" for f in existing_files)
print(f"内容已存在:\n{file_list}")
response = input("是否重新下载?[y/N]: ").strip().lower()
return response == 'y'
Anti-Patterns to Avoid
- 每个平台单独的 CLI 入口: 不要为每个平台创建独立的命令行工具,应统一到工作流管道
- 硬编码文件路径: 使用 KBSettings 配置,不要硬编码输出目录
- 忽略网络错误: 必须使用 retry_on_network_error 装饰器处理网络请求
- 同步阻塞调用: 元数据获取和内容下载应使用 async/await
Don't Hand-Roll
| Problem | Don't Build | Use Instead | Why |
|---|---|---|---|
| BV/AV号转换 | 自己实现算法 | bilibili-api 库的转换函数 | 算法复杂,易出错 |
| 短链解析 | HTTP 重定向处理 | requests.head(allow_redirects=True) | 自动处理重定向链 |
| HTML 转 Markdown | 正则替换 | markdown_converter.py 已有实现 | 处理各种 HTML 边界情况 |
| 浏览器自动化 | selenium 脚本 | Playwright + 现有提取器 | 现有代码已完整实现 |
Key insight: 现有代码库已有完整的平台 API 实现,Phase 2 的核心工作是整合而非重写。
Common Pitfalls
Pitfall 1: B站 API 403 错误
What goes wrong: 直接调用 B站 API 返回 403 Forbidden Why it happens: B站有反爬机制,需要正确的 User-Agent 和 Referer How to avoid: 使用现有 BilibiliExtractor 的 DEFAULT_HEADERS Warning signs: API 返回 403,或返回空数据
Pitfall 2: 知乎需要登录
What goes wrong: 知乎内容无法获取,返回登录页面 Why it happens: 知乎对未登录用户限制内容访问 How to avoid: 从 KBSettings 获取 zhihu_cookie,传递给 Playwright 浏览器 Warning signs: 内容为空,页面标题包含"登录"
Pitfall 3: 短链解析超时
What goes wrong: b23.tv 短链解析卡住或超时 Why it happens: 网络问题或短链服务不可用 How to avoid: 设置合理的 timeout(10秒),失败时返回原始 URL Warning signs: 请求超时,网络异常
Pitfall 4: 文件名包含特殊字符
What goes wrong: 标题包含 / : * ? " < > | 等字符导致文件创建失败 Why it happens: 操作系统文件系统限制 How to avoid: 实现文件名清理函数,移除或替换非法字符 Warning signs: FileNotFoundError,PermissionError
Pitfall 5: 幂等检查遗漏
What goes wrong: 重复执行产生重复文件或覆盖已有内容 Why it happens: 忘记在执行前检查文件是否存在 How to avoid: 在 LinkParser 解析成功后立即检查,遵循 CONTEXT.md 的决策 Warning signs: workflow_state.json 中步骤状态与实际文件不一致
Code Examples
B站视频元数据获取
# Source: kb-bilibili-metadata/bilibili_extractor.py
from kb_bilibili_metadata.bilibili_extractor import BilibiliExtractor
async def fetch_bilibili_video(bvid: str) -> dict:
"""获取 B站视频元数据"""
extractor = BilibiliExtractor(headless=True, timeout=30)
result = extractor.extract_video_metadata(bvid)
# 提取核心字段
return {
"title": result["metadata"]["title"],
"author": result["metadata"]["author"],
"duration": result["metadata"]["duration"],
"pubtime": result["metadata"]["pubtime"],
"description": result["description"],
"stats": result["stats"],
"bvid": result["metadata"]["bvid"],
"url": result["metadata"]["url"],
}
知乎内容提取
# Source: kb-zhihu-metadata/zhihu_extractor.py
from kb_zhihu_metadata.zhihu_extractor import ZhihuPlaywrightExtractor
from kb_pipeline.config import get_settings
async def fetch_zhihu_content(url: str) -> dict:
"""获取知乎内容"""
settings = get_settings()
cookie = settings.zhihu_cookie.get_secret_value()
extractor = ZhihuPlaywrightExtractor(
headless=True,
timeout=30,
cookie=cookie if cookie else None
)
result = extractor.extract(url)
if not result:
return None
return {
"type": result["type"],
"title": result["title"],
"author": result["author"],
"content": result["content_markdown"],
"publish_time": result["publish_time"],
"url": url,
"content_id": result["content_id"],
}
内容输出格式
# Source: CONTEXT.md 决策
from datetime import datetime
import yaml
def format_content_markdown(metadata: dict, content: str) -> str:
"""生成标准 Markdown 格式(YAML Frontmatter + 正文)"""
frontmatter = {
"title": metadata["title"],
"author": metadata["author"],
"source_url": metadata["url"],
"platform": metadata["platform"],
"content_type": metadata["content_type"],
"fetched_at": datetime.now().isoformat(),
}
return f"---\n{yaml.dump(frontmatter, allow_unicode=True)}---\n\n{content}"
State of the Art
| Old Approach | Current Approach | When Changed | Impact |
|---|---|---|---|
| 各平台独立脚本 | 统一管道模型 | Phase 1 | 可复用基础设施,统一错误处理 |
| 手动处理短链 | 自动解析 + 回退 | 现有 | 提升用户体验,减少手动操作 |
| 同步阻塞请求 | 异步并发 | 现有 | 提升性能,支持并发处理 |
Deprecated/outdated:
- bilibili-dl: 已被 yt-dlp 替代,yt-dlp 支持更多平台且维护活跃
- selenium: 已被 Playwright 替代,Playwright API 更现代
Open Questions
-
Playwright 浏览器安装
- What we know: Playwright 需要单独安装浏览器(
playwright install chromium) - What's unclear: 是否需要在首次运行时自动检测并提示安装
- Recommendation: 在 README 中说明安装步骤,运行时检测并给出友好提示
- What we know: Playwright 需要单独安装浏览器(
-
Cookie 过期处理
- What we know: 知乎需要 Cookie 才能获取完整内容
- What's unclear: Cookie 过期时的错误提示和刷新机制
- Recommendation: 检测登录页面特征,返回 AuthError 并提示更新 Cookie
Validation Architecture
Test Framework
| Property | Value |
|---|---|
| Framework | pytest 8.x |
| Config file | pytest.ini |
| Quick run command | pytest tests/test_pipeline/test_link_parser.py -x -v |
| Full suite command | pytest tests/ -v --cov=kb_pipeline |
Phase Requirements → Test Map
| Req ID | Behavior | Test Type | Automated Command | File Exists? |
|---|---|---|---|---|
| PLAT-01 | B站视频元数据提取 | unit | pytest tests/test_platforms/test_bilibili.py::test_video_metadata -x |
Wave 0 |
| PLAT-01 | B站视频内容下载 | integration | pytest tests/test_platforms/test_bilibili.py::test_video_content -x |
Wave 0 |
| PLAT-02 | B站专栏内容提取 | unit | pytest tests/test_platforms/test_bilibili.py::test_article_content -x |
Wave 0 |
| PLAT-03 | 知乎问答内容提取 | unit | pytest tests/test_platforms/test_zhihu.py::test_answer_content -x |
Wave 0 |
| PLAT-03 | 知乎专栏内容提取 | unit | pytest tests/test_platforms/test_zhihu.py::test_zhuanlan_content -x |
Wave 0 |
| PLAT-01/02/03 | 链接解析 | unit | pytest tests/test_pipeline/test_link_parser.py -x |
Wave 0 |
| PLAT-01/02/03 | 幂等检查 | unit | pytest tests/test_pipeline/test_idempotency.py -x |
Wave 0 |
Sampling Rate
- Per task commit:
pytest tests/test_pipeline/test_link_parser.py tests/test_platforms/ -x - Per wave merge:
pytest tests/ -v --cov=kb_pipeline --cov-report=term-missing - Phase gate: Full suite green before
/gsd:verify-work
Wave 0 Gaps
tests/test_pipeline/test_link_parser.py— LinkParser 单元测试tests/test_pipeline/test_idempotency.py— 幂等检查测试tests/test_platforms/__init__.py— 测试模块初始化tests/test_platforms/conftest.py— 平台测试共享 fixturestests/test_platforms/test_bilibili.py— B站适配器测试tests/test_platforms/test_zhihu.py— 知乎适配器测试- Mock fixtures: B站/知乎 API 响应数据
Sources
Primary (HIGH confidence)
- 现有代码:
kb-bilibili-metadata/bilibili_extractor.py- B站 API 调用实现 - 现有代码:
kb-zhihu-metadata/zhihu_extractor.py- 知乎 Playwright 提取器 - 现有代码:
kb-link-parser/link-parser.py- 链接解析器 - 现有代码:
kb_pipeline/schemas.py- Phase 1 Schema 定义
Secondary (MEDIUM confidence)
- Phase 1 实现:
kb_pipeline/errors.py,kb_pipeline/config.py,kb_pipeline/retry.py - 现有代码:
kb_workflow/kb_workflow.py- 子技能映射和内容类型判断
Tertiary (LOW confidence)
- N/A - 本阶段研究基于现有代码库,无外部来源
Metadata
Confidence breakdown:
- Standard stack: HIGH - 基于 Phase 1 已采用的技术栈和现有代码
- Architecture: HIGH - 现有代码已有成熟实现,主要是整合工作
- Pitfalls: HIGH - 基于现有代码中已处理的问题和 CONTEXT.md 决策
Research date: 2026-03-13 Valid until: 30 days - Python 和相关库相对稳定
Research complete. Ready for planning phase.