Interview Radar

InterviewRadar · 面试雷达 Skill

把「简历 + 模糊岗位」变成一份基于真实面经内容的个性化备考包。你(agent)负责推理判断;scripts/ 下的 Python 脚本负责确定性的脏活。两者通过 corpus_cache/ 里的 JSON 文件交互。

输入

• 简历:PDF、图片或文本文件路径。
• 模糊岗位:一个方向,例如"AI 应用开发"(不是具体的 JD)。

工具(用包内 venv 运行:.venv/bin/python)

• scripts/resume_extract.py → extract_resume(path) -> ResumeExtraction{text, needs_vision, asset_path}
• scripts/connectors/github.py → GithubConnector(repo_raw_urls).search(queries) -> SearchResult
• scripts/connectors/nowcoder.py → NowCoderConnector(post_urls).search(queries) -> SearchResult
• scripts/connectors/xiaohongshu.py → XiaohongshuConnector(export_path=..., driver=..., enable_image_ocr=...).search(queries) -> SearchResult(二选一:export_path 读预生成的 JSON;driver 自动跑 MediaCrawler;enable_image_ocr=True 为 deep 模式,图片会下载/OCR 后作为主正文;False 为 fast 模式,只读标题、正文 caption、标签和时间戳)
• scripts/scrape/mediacrawler_driver.py → MediaCrawlerDriver(home=None).scrape_xhs(keywords, login_type="qrcode"|"cookie") -> Path,驱动模式——shell out 调本机已装的 MediaCrawler(默认从 $MEDIACRAWLER_HOME 或 ~/.mediacrawler/ 找)。推荐 cookie 登录:在 MediaCrawler config/base_config.py 里设置 LOGIN_TYPE = "cookie" 和 COOKIES = "web_session=<value>"。
• scripts/scrape/normalize_xhs.py → normalize(notes) -> list[dict](CLI:python -m scripts.scrape.normalize_xhs <in.json> -o <out.json>),把 MediaCrawler 原生输出归一化为 XiaohongshuConnector 的输入。仅手动模式用得到;driver 模式连接器内部自动调用。
• scripts/ocr/extract.py → extract_text_from_image(path, engine=None, min_confidence=0.6) -> OcrResult{text, confidence, needs_vision}
• scripts/ocr/xhs_images.py → 下载小红书 image_list,默认探测 RapidOCR,把分页 OCR 合并进 RawPost.content_text/raw_text;低质量时设 needs_vision_fallback=True
• scripts/corpus/store.py → save_raw_posts / load_raw_posts / save_questions / load_questions
• scripts/corpus/recency.py → filter_recent(posts, window_days=730, today=None) -> list[RawPost]
• scripts/corpus/dedupe_rank.py → dedupe_and_rank(questions) -> list[Question]
• 数据模型在 scripts/models.py;结构说明见 assets/schema.md。

工作流

0. 准备(仅当启用小红书源)。 两种模式二选一,见 docs/setup/mediacrawler.md:

• driver 模式(推荐):用户一次性装 MediaCrawler 并登录。优先用 cookie 模式:从正常浏览器复制 web_session,写入 MediaCrawler config.COOKIES,然后用 XiaohongshuConnector(driver=MediaCrawlerDriver(), login_type="cookie") 自动跑采集。二维码模式仍可用,但更容易触发风控。
• 手动模式:用户每次自己跑 MediaCrawler + normalize_xhs.py,把 corpus_cache/xhs_export.json 喂给 XiaohongshuConnector(export_path=...)。
• 读取深度必须说明:小红书很多面经受文字区限制,完整题目在图片里。fast 模式只读标题/正文/标签,适合验证召回;正式备考包优先用 deep 模式(enable_image_ocr=True)读取图片 OCR。若因速度或依赖问题使用 fast,必须在输出中明确写“未读取图片 OCR,可能漏掉图片里的完整题目”。

文本/牛客/GitHub 源不需要这一步。

1. 简历理解。 调用 extract_resume。若 needs_vision 为真,就用你自己的视觉能力直接读这张图片/PDF。产出结构化摘要:技能、项目(每个项目用到的技术)、关键术语。

2. 种子查询生成。 用你自己的领域知识,从「岗位方向 + 简历」推导出种子查询。这是领域无关的:无论什么领域(市场、量化、后端、设计……)你本来就知道相关的岗位别名和底层技能/话题,当场生成即可。不要依赖任何预设词表。 种子来自两处:(a) 岗位方向隐含的相关岗位别名;(b) 从简历里抽出的具体技能/项目/关键词。优先用底层技能/话题词,而不是岗位名——它们更稳定、召回更好。

3. 迭代检索。 源的优先级:牛客 + 小红书(主力,带时间戳)> GitHub(补充,常过时)。

3a. URL 发现(每轮先做)。 用你的搜索能力(WebSearch 或等价工具)对当前的种子查询跑一遍,收集候选 URL。按域名分桶:

• nowcoder.com/discuss/<post_id> → 进 NowCoderConnector(post_urls=...)
• xiaohongshu.com/explore/<note_id> → 不能直抓;如果启用了小红书源,把当前的关键词丢给 XiaohongshuConnector(driver=MediaCrawlerDriver()).search([keywords]),它会自动 shell out 跑 MediaCrawler。若未启用,记下笔记 ID 让用户按 docs/setup/mediacrawler.md 配置
• github.com/<owner>/<repo>/blob/<branch>/<path> 或 raw.githubusercontent.com/... → 转 raw URL → 进 GithubConnector(repo_raw_urls=...)
• 其他公开正文页(知乎 article、CSDN 文章、个人博客、woshipm/uisdc 等):用 WebFetch 拉回正文,自己手工构造 RawPost(source="webfetch:<domain>", post_type="text", raw_text=<正文>, posted_at=<页面可见日期或 None>),和 connector 结果一起 save_raw_posts

显式排除:聚合/listing 页(只接 article 页);403/需登录的页面(不浪费 fetch 配额,记下来当作知识缺口告诉用户)。

3b. 调 connectors + 收割。 把分好桶的 URL 喂给对应 connector,结果用 save_raw_posts 落盘。读取结果,收割真实出现的岗位名 / 标签 / 高频术语,用收割到的词跑下一轮 3a,直到不再冒出新词。若某 connector 返回 status="degraded"(例如牛客需要 cookie、小红书需要先跑 MediaCrawler、或消息含 selector 表示 HTML 漂移),把它需要的东西告诉用户;主力源降级会显著影响时效性,必须明确提示用户,不要默默用 GitHub 凑数。

GitHub 调用必带 `relevance_hints`(不要传 `None` 或 `[]`)。 GitHub 仓库里夹带大量算法/八股,不过滤会严重污染语料。GithubConnector(repo_raw_urls=..., relevance_hints=<当前一轮的术语/岗位别名>)。冷启动(第一轮还没有收割结果)时,直接把步骤 2 的种子查询当 hints 传进去——任何时候都要传非空 list。命中规则:子串、大小写不敏感,只要正文里出现任一 hint 就保留。

Human-in-the-loop: 在最后一轮之前,把你从真实数据里发现的方向/术语展示给用户,让他增删/纠偏。

4. 内容级相关性判定。 通过读帖子内容对照用户岗位 + 简历来判断每条是否相关——不是靠帖子的岗位名是否匹配某张预设表。

5. 题目抽取。 文本类 RawPost 直接读。图片类 RawPost(小红书,`post_type="image"`):正式包默认用 deep 模式读 content_text/raw_text,它优先来自图片 OCR;标题、caption、tags 只在 locator_text 里做召回/定位。若 needs_vision_fallback=True,用你自己的视觉能力按 asset_paths 顺序补读图片。若本次为了速度用了 fast 模式(enable_image_ocr=False),只能把标题/正文/标签作为线索,不得声称覆盖了图片里的完整面经题。把读到的题目转成标准化的 Question,图片来源的设 modality_origin="ocr" 或 "vision"。用 save_questions 落盘。

5b. 时效过滤。 在抽取出题目之前,先用 filter_recent(raw_posts) 把超过约两年的帖子丢掉(默认窗口 730 天;posted_at 为 None 的无日期帖子保留)。时效性是硬需求——过时的面经没有价值。

6. 去重 & 排序。 跑 dedupe_and_rank(load_questions(...)) 并保存排序结果。排序同时考虑频次和时效(近期题加权更高),这就是高频题集。

7. 项目锚定推理。 对每道高频题,检查它能否挂到简历里的某个项目/技能上。能挂上就构造 FollowUpChain(种子题 → 个性化追问,is_grounded=true)。每个追问都必须能追溯到(简历某项目/技能)+(某条真实爬到的题);追溯不上就设 is_grounded=false,当普通八股题保留。不要凭空编追问。

8. 备考包。 严格按下面的固定模板写 Markdown,保存到 corpus_cache/prep_package.md 并展示给用户。不要自由发挥改大结构;若某部分数据不足,写清"数据不足/未覆盖",不要编。

备考包固定模板

# {目标岗位}岗位备考包 — {候选人姓名或"候选人"}

生成日期:{YYYY-MM-DD}
目标岗位:{用户给定岗位 + 从真实数据收割到的岗位别名}

## 1. 你的候选人定位

用 1 段话定义候选人的面试包装方向。格式:

> {一句候选人定位}

简历里最强的三个证据:

1. **{项目/经历 1}**
   - {证据点 1}
   - {证据点 2}
   - {适合回答什么面试问题}

2. **{项目/经历 2}**
   - ...

3. **{项目/经历 3}**
   - ...

## 2. 岗位 Gap 分析

| 维度 | 当前简历表现 | 面试风险 | 准备建议 |
|---|---|---|---|
| {能力维度} | {基于简历的证据} | {真实面经会追问的风险} | {可执行补强建议} |

至少覆盖:技术理解、产品基本功、业务 sense、数据/实验、落地可信度。非 AI 岗位时把维度换成该岗位的核心能力。

## 3. 真实数据来源概况

本次召回:

- {来源 1}:{数量/状态/保存路径}
- {来源 2}:{数量/状态/保存路径}

数据缺口:

- {降级源、OCR/反爬/时间戳缺口等}

## 4. 高频题 Top {N}

### {序号}. {题目}

来源:
- {source title 或 source type}:`{url}`
- ...

回答要点/回答框架:

- {要点 1}
- {要点 2}
- {要点 3}

可挂简历锚点:

> {把这题连接到候选人某个项目/技能的一段话}

如果适合表格解释,可以用小表格;否则用 bullet。每题必须至少有 1 个真实来源。没有来源的题不能进 Top 高频题。

## 5. 个性化项目追问链

### 链 {序号}:{主题} → {简历项目/经历}

种子题:{来自第 4 节的高频题}

追问:

1. {追问 1}
2. {追问 2}
3. {追问 3}
4. {追问 4}
5. {追问 5}

准备重点:

- {如何准备真实例子/图/指标}
- {面试时要强调的产品视角}

每条追问必须同时能追溯到"真实面经题目"和"简历项目/技能";追溯不上就不要写成个性化追问。

## 6. 你的 60-90 秒自我介绍草稿

写 2-3 段中文口语稿。必须包含:

- 背景
- 目标岗位动机
- 2-3 个最强项目证据
- 候选人的差异化定位

## 7. 一周冲刺计划

### Day 1:{主题}

- {行动项}
- {行动项}

...

### Day 7:{主题}

- {行动项}
- {行动项}

## 8. 建议你立刻补强的简历表述

### {项目/经历 1}

补:

> {可直接放进简历/面试话术的改写}

### {项目/经历 2}

补:

> ...

## 9. 来源列表

按来源类型分组列出代表来源:

小红书:
- `{url}` — {一句说明}

牛客/网页/GitHub:
- `{url}` — {一句说明}

## 10. 面试前速查清单

- 60 秒自我介绍
- {3 个最能证明岗位匹配度的项目/经历证据}
- {3 个真实 trade-off / 失败 / 返工案例}
- {5 个目标岗位必会高频题;例如 AI 产品岗可写 RAG、Agent、实验、指标、产品设计,其他岗位按真实面经替换}
- {4 个高质量反问}
- {可以展示的作品、流程图、数据、文档或代码证据}

模板约束:

• 全文中文,但技术名词可保留英文。
• 第 4 节的题目必须来自 Question 或明确可追溯的 RawPost.content_text/raw_text;小红书 caption/tag 只作为 locator_text 辅助,不能单独生成题目。
• 高频题优先按 dedupe_and_rank 结果排序;人工调整时只能因岗位相关性或简历匹配度调整,并说明依据。
• 来源 URL 必须真实存在于 source_refs 或 RawPost.url;不要写"综合资料"这类不可追溯来源。
• 自我介绍和简历表述可以做表达优化,但事实必须来自简历或真实面经。
• 第 10 节必须按用户目标岗位动态生成,不要写死为 AI 产品题;AI 产品只是示例。

约束

• 所有面向用户的产出一律用中文(备考包、题目、追问、分析)——面经源是中文。
• 当前源:牛客 + 小红书(主力,带时间戳)+ GitHub(补充)。
• 小红书走 MediaCrawler 采集导出(用户预先离线跑一次,流程见 docs/setup/mediacrawler.md),OCR 采用混合策略(粗 OCR + 视觉回退);MediaCrawler 仅供个人、非商业用途。
• 时效性是硬需求:只保留近两年的面经,排序向近期加权。
• 后续版本用到的第三方爬虫(如 MediaCrawler)仅供个人、非商业用途。
• 可追溯优先于流畅度:绝不编造无法追溯到真实数据的题目或追问。
• 若 connector 返回 degraded 且消息含 selector,说明源站点 HTML 改了选择器;到对应 scripts/connectors/<name>.py 顶部注释看当前假设,核对真实 HTML 后更新选择器并补 fixture。

Data Structures

• RawPost { source, url, post_type(text|image|mixed), raw_text, locator_text, content_text, image_ocr_text|null, needs_vision_fallback, extraction_quality, posted_at(ISO date|null), asset_paths[], comments[] }

One scraped unit (a question-like line, post, or image). posted_at is the source post date (ISO YYYY-MM-DD) or null for undated sources. Produced by connectors. Filtered to the recency window by corpus/recency.py. raw_text is the current best primary content for legacy pipeline compatibility; locator_text is source metadata for retrieval/filtering.

• Question { text, source_refs[], freq, latest_posted_at(ISO date|null), role_tags[], topic, modality_origin(text|ocr|vision) }

A normalized interview question. latest_posted_at is the most recent date among merged duplicates. Produced by the agent's extraction step from RawPosts, then merged/ranked (by frequency AND recency) by corpus/dedupe_rank.py.

• FollowUpChain { seed_question, resume_anchor, followups[], is_grounded }

A personalized follow-up chain. Produced by the agent's project-anchoring step. is_grounded=false means it degraded to a plain 八股 question (no resume anchor found).

Persistence: normalized JSON under corpus_cache/ via corpus/store.py.

Image posts & OCR

Image-based posts (小红书) use post_type="image". The connector downloads images to corpus_cache/assets/xhs/{note_id}/, runs OCR in image order, and stores the merged page text in image_ocr_text and content_text. Captions/tags stay in locator_text so extraction reads the image content first. Low-quality OCR sets needs_vision_fallback=true.

免责声明 / Disclaimer

中文

本项目 InterviewRadar 是一个面向个人学习的开源工具,使用前请阅读以下条款:

1. 用途限定

• 本项目仅供个人学习、面试准备、技术研究使用。
• 严禁用于商业用途,包括但不限于:对外销售、付费咨询、二次开发为收费产品。
• 严禁用于任何违反目标平台服务条款的批量采集行为。

2. 数据来源责任

本项目从公开网络资源中拉取面经内容,包括但不限于牛客网、GitHub、技术博客等。

• 拉取的内容版权归原作者所有。
• 本项目仅提供工具能力,不存储任何爬取的原始数据;所有数据存于用户本地 corpus_cache/ 目录。
• 使用本工具拉取数据的合规性,由用户自行承担。建议:
• 控制请求频次,不要对源站点造成压力
• 遵守每个平台的 robots.txt 和服务条款
• 不要将拉取的数据二次分发

3. 第三方依赖说明

• 本项目通过适配器与 MediaCrawler 集成以支持小红书等平台采集。MediaCrawler 项目同样要求仅供个人非商业用途。
• 用户需自行安装并合规使用 MediaCrawler;本项目不直接 vendor 任何爬虫代码。

4. AI 生成内容

本项目使用 LLM 处理拉取到的素材,生成的备考包(prep_package.md)中:

• 题目来自真实公开面经,但经过 LLM 加工,可能存在改写或推理偏差
• 项目追问由 LLM 基于用户简历推理生成,不保证一定会被实际面试官问到

5. 风险自担

本工具按"原样"提供,作者不对任何使用后果负责,包括但不限于:

• 拉取数据时遇到的法律风险
• LLM 生成内容的事实错误
• 备考包的应试效果

如不接受以上条款,请立即停止使用本项目。

---

English

InterviewRadar is an open-source tool intended for personal study only.

• Personal, non-commercial use only. No reselling, paid consulting, or repackaging into paid products.
• You are responsible for the compliance of any data you scrape using this tool. Respect the robots.txt and ToS of each source platform.
• Third-party scraping integration (MediaCrawler) is also restricted to personal non-commercial use; install and operate it under its own terms.
• Generated interview prep packages are LLM-processed and may contain inaccuracies. Do not treat them as authoritative.
• Provided AS-IS with no warranty. The author is not liable for any consequences of use.

By using this project you accept the terms above.

Interview Intelligence Skill — MVP Vertical Slice Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Build a runnable Claude Code Agent Skill that, given a resume (PDF/image) and a fuzzy role, retrieves interview questions from GitHub interview repos, dedupes/ranks them, and lets the agent produce a personalized prep package with project-anchored follow-ups.

Architecture: Deterministic dirty work (fetch, parse, persist, dedupe/rank, resume text extraction) lives in tested Python scripts. Judgment/reasoning (resume understanding, iterative vocabulary harvesting, content-relevance, project anchoring, prep authoring) is performed by the agent following SKILL.md. Python tools and the agent communicate through normalized JSON files in corpus_cache/.

Tech Stack: Python 3.11, pytest, pypdf, requests. (Embeddings/semantic dedupe and the 牛客/小红书 connectors + OCR are deferred to Plans 2 and 3.)

Scope note: This plan is the MVP slice. 牛客 connector (Plan 2) and 小红书 + hybrid OCR (Plan 3) are out of scope here. The connector interface in Task 4 is designed so those drop in later without changing callers.

---

Task 1: Project scaffold

Files:

• Create: interview-intelligence/requirements.txt
• Create: interview-intelligence/pytest.ini
• Create: interview-intelligence/scripts/__init__.py
• Create: interview-intelligence/scripts/connectors/__init__.py
• Create: interview-intelligence/scripts/corpus/__init__.py
• Create: interview-intelligence/tests/__init__.py

• [ ] Step 1: Create requirements.txt

pypdf==4.3.1
requests==2.32.3
pytest==8.3.2

• [ ] Step 2: Create pytest.ini

[pytest]
testpaths = tests
python_files = test_*.py

• [ ] Step 3: Create empty package markers

Create each __init__.py listed above as an empty file.

• [ ] Step 4: Install deps

Run: cd interview-intelligence && python3 -m venv .venv && .venv/bin/pip install -r requirements.txt Expected: installs without error.

• [ ] Step 5: Commit

git add interview-intelligence/requirements.txt interview-intelligence/pytest.ini interview-intelligence/scripts interview-intelligence/tests
git commit -m "chore: scaffold interview-intelligence skill package"

---

Task 2: Data models

Files:

• Create: interview-intelligence/scripts/models.py
• Test: interview-intelligence/tests/test_models.py

• [ ] Step 1: Write the failing test

from scripts.models import RawPost, Question, FollowUpChain


def test_rawpost_roundtrips_through_dict():
    post = RawPost(
        source="github",
        url="https://example.com/p1",
        post_type="text",
        raw_text="What is MCP?",
        asset_paths=[],
        comments=["see docs"],
    )
    assert RawPost.from_dict(post.to_dict()) == post


def test_question_roundtrips_through_dict():
    q = Question(
        text="What is MCP?",
        source_refs=["https://example.com/p1"],
        freq=2,
        role_tags=["agent"],
        topic="protocols",
        modality_origin="text",
    )
    assert Question.from_dict(q.to_dict()) == q


def test_followupchain_roundtrips_through_dict():
    chain = FollowUpChain(
        seed_question="What is MCP?",
        resume_anchor="skill-driven project",
        followups=["How does your skill engine work?"],
        is_grounded=True,
    )
    assert FollowUpChain.from_dict(chain.to_dict()) == chain

• [ ] Step 2: Run test to verify it fails

Run: cd interview-intelligence && .venv/bin/pytest tests/test_models.py -v Expected: FAIL with "No module named 'scripts.models'".

• [ ] Step 3: Write minimal implementation

from dataclasses import dataclass, field, asdict


@dataclass
class RawPost:
    source: str
    url: str
    post_type: str  # text | image | mixed
    raw_text: str
    asset_paths: list[str] = field(default_factory=list)
    comments: list[str] = field(default_factory=list)

    def to_dict(self) -> dict:
        return asdict(self)

    @classmethod
    def from_dict(cls, d: dict) -> "RawPost":
        return cls(**d)


@dataclass
class Question:
    text: str
    source_refs: list[str] = field(default_factory=list)
    freq: int = 1
    role_tags: list[str] = field(default_factory=list)
    topic: str = ""
    modality_origin: str = "text"  # text | ocr | vision

    def to_dict(self) -> dict:
        return asdict(self)

    @classmethod
    def from_dict(cls, d: dict) -> "Question":
        return cls(**d)


@dataclass
class FollowUpChain:
    seed_question: str
    resume_anchor: str
    followups: list[str] = field(default_factory=list)
    is_grounded: bool = False

    def to_dict(self) -> dict:
        return asdict(self)

    @classmethod
    def from_dict(cls, d: dict) -> "FollowUpChain":
        return cls(**d)

• [ ] Step 4: Run test to verify it passes

Run: cd interview-intelligence && .venv/bin/pytest tests/test_models.py -v Expected: 3 passed.

• [ ] Step 5: Commit

git add interview-intelligence/scripts/models.py interview-intelligence/tests/test_models.py
git commit -m "feat: add RawPost/Question/FollowUpChain data models"

---

Task 3: Corpus store (JSON persistence)

Files:

• Create: interview-intelligence/scripts/corpus/store.py
• Test: interview-intelligence/tests/test_store.py

• [ ] Step 1: Write the failing test

from scripts.models import RawPost, Question
from scripts.corpus.store import (
    save_raw_posts, load_raw_posts, save_questions, load_questions,
)


def test_raw_posts_save_and_load(tmp_path):
    posts = [RawPost("github", "u1", "text", "Q1"), RawPost("github", "u2", "text", "Q2")]
    path = tmp_path / "raw.json"
    save_raw_posts(posts, path)
    assert load_raw_posts(path) == posts


def test_questions_save_and_load(tmp_path):
    qs = [Question("Q1", ["u1"]), Question("Q2", ["u2"], freq=3)]
    path = tmp_path / "q.json"
    save_questions(qs, path)
    assert load_questions(path) == qs

• [ ] Step 2: Run test to verify it fails

Run: cd interview-intelligence && .venv/bin/pytest tests/test_store.py -v Expected: FAIL with "No module named 'scripts.corpus.store'".

• [ ] Step 3: Write minimal implementation

import json
from pathlib import Path

from scripts.models import RawPost, Question


def save_raw_posts(posts: list[RawPost], path) -> None:
    Path(path).parent.mkdir(parents=True, exist_ok=True)
    data = [p.to_dict() for p in posts]
    Path(path).write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")


def load_raw_posts(path) -> list[RawPost]:
    data = json.loads(Path(path).read_text(encoding="utf-8"))
    return [RawPost.from_dict(d) for d in data]


def save_questions(questions: list[Question], path) -> None:
    Path(path).parent.mkdir(parents=True, exist_ok=True)
    data = [q.to_dict() for q in questions]
    Path(path).write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")


def load_questions(path) -> list[Question]:
    data = json.loads(Path(path).read_text(encoding="utf-8"))
    return [Question.from_dict(d) for d in data]

• [ ] Step 4: Run test to verify it passes

Run: cd interview-intelligence && .venv/bin/pytest tests/test_store.py -v Expected: 2 passed.

• [ ] Step 5: Commit

git add interview-intelligence/scripts/corpus/store.py interview-intelligence/tests/test_store.py
git commit -m "feat: add JSON corpus store for raw posts and questions"

---

Task 4: Connector interface

Files:

• Create: interview-intelligence/scripts/connectors/base.py
• Test: interview-intelligence/tests/test_base_connector.py

• [ ] Step 1: Write the failing test

from scripts.models import RawPost
from scripts.connectors.base import Connector, SearchResult


def test_searchresult_holds_status_and_posts():
    posts = [RawPost("github", "u1", "text", "Q1")]
    r = SearchResult(posts=posts, status="ok", message="")
    assert r.posts == posts
    assert r.status == "ok"


def test_searchresult_degraded_factory_has_no_posts():
    r = SearchResult.degraded("nowcoder", "needs cookie")
    assert r.posts == []
    assert r.status == "degraded"
    assert "cookie" in r.message


def test_connector_is_abstract():
    class Dummy(Connector):
        name = "dummy"

        def search(self, queries):
            return SearchResult(posts=[], status="ok", message="")

    assert Dummy().search(["x"]).status == "ok"

• [ ] Step 2: Run test to verify it fails

Run: cd interview-intelligence && .venv/bin/pytest tests/test_base_connector.py -v Expected: FAIL with "No module named 'scripts.connectors.base'".

• [ ] Step 3: Write minimal implementation

from abc import ABC, abstractmethod
from dataclasses import dataclass, field

from scripts.models import RawPost


@dataclass
class SearchResult:
    posts: list[RawPost] = field(default_factory=list)
    status: str = "ok"  # ok | degraded | error
    message: str = ""

    @classmethod
    def degraded(cls, source: str, message: str) -> "SearchResult":
        return cls(posts=[], status="degraded", message=f"[{source}] {message}")


class Connector(ABC):
    name: str = "base"

    @abstractmethod
    def search(self, queries: list[str]) -> SearchResult:
        """Run queries against the source and return normalized RawPosts."""
        raise NotImplementedError

• [ ] Step 4: Run test to verify it passes

Run: cd interview-intelligence && .venv/bin/pytest tests/test_base_connector.py -v Expected: 3 passed.

• [ ] Step 5: Commit

git add interview-intelligence/scripts/connectors/base.py interview-intelligence/tests/test_base_connector.py
git commit -m "feat: add Connector interface and SearchResult with degrade factory"

---

Task 5: GitHub connector — markdown parsing

Files:

• Create: interview-intelligence/scripts/connectors/github.py
• Test: interview-intelligence/tests/test_github_connector.py

The parser splits an interview-repo markdown file into candidate question posts. It treats markdown headings (#..######) and list bullets (-, *, or 1.) that end in a question mark OR contain interview-question keywords as separate RawPosts. Network fetch is a thin wrapper tested separately by monkeypatching.

• [ ] Step 1: Write the failing test

from scripts.connectors.github import extract_posts_from_markdown, GithubConnector


SAMPLE_MD = """# Agent 面经
## 一面
- 说明 MCP 和 Skill 的区别
- 什么是 RAG？
随便一句不是题目的话。
### 项目相关
1. 介绍一下你的 agent 项目架构
"""


def test_extract_picks_question_like_lines():
    posts = extract_posts_from_markdown(SAMPLE_MD, "https://example.com/repo")
    texts = [p.raw_text for p in posts]
    assert "说明 MCP 和 Skill 的区别" in texts
    assert "什么是 RAG？" in texts
    assert "介绍一下你的 agent 项目架构" in texts
    assert "随便一句不是题目的话。" not in texts


def test_extract_sets_source_and_url():
    posts = extract_posts_from_markdown(SAMPLE_MD, "https://example.com/repo")
    assert all(p.source == "github" for p in posts)
    assert all(p.url == "https://example.com/repo" for p in posts)
    assert all(p.post_type == "text" for p in posts)


def test_connector_search_uses_injected_fetcher():
    conn = GithubConnector(
        repo_raw_urls=["https://example.com/repo"],
        fetcher=lambda url: SAMPLE_MD,
    )
    result = conn.search(["agent"])
    assert result.status == "ok"
    assert any("RAG" in p.raw_text for p in result.posts)


def test_connector_degrades_on_fetch_error():
    def boom(url):
        raise RuntimeError("network down")

    conn = GithubConnector(repo_raw_urls=["https://example.com/repo"], fetcher=boom)
    result = conn.search(["agent"])
    assert result.status == "degraded"
    assert result.posts == []

• [ ] Step 2: Run test to verify it fails

Run: cd interview-intelligence && .venv/bin/pytest tests/test_github_connector.py -v Expected: FAIL with "No module named 'scripts.connectors.github'".

• [ ] Step 3: Write minimal implementation

import re
from collections.abc import Callable

import requests

from scripts.connectors.base import Connector, SearchResult
from scripts.models import RawPost

_KEYWORDS = ("介绍", "说明", "区别", "原理", "什么是", "如何", "为什么", "解释")
_HEADING = re.compile(r"^#{1,6}\s+(.*)$")
_BULLET = re.compile(r"^(?:[-*]|\d+\.)\s+(.*)$")


def _is_question_like(text: str) -> bool:
    t = text.strip()
    if not t:
        return False
    if t.endswith("?") or t.endswith("？"):
        return True
    return any(k in t for k in _KEYWORDS)


def extract_posts_from_markdown(md_text: str, url: str) -> list[RawPost]:
    posts: list[RawPost] = []
    for line in md_text.splitlines():
        m = _HEADING.match(line) or _BULLET.match(line)
        candidate = m.group(1).strip() if m else line.strip()
        if _is_question_like(candidate):
            posts.append(RawPost(source="github", url=url, post_type="text", raw_text=candidate))
    return posts


def _default_fetcher(url: str) -> str:
    resp = requests.get(url, timeout=30)
    resp.raise_for_status()
    return resp.text


class GithubConnector(Connector):
    name = "github"

    def __init__(self, repo_raw_urls: list[str], fetcher: Callable[[str], str] | None = None):
        self.repo_raw_urls = repo_raw_urls
        self.fetcher = fetcher or _default_fetcher

    def search(self, queries: list[str]) -> SearchResult:
        posts: list[RawPost] = []
        try:
            for url in self.repo_raw_urls:
                posts.extend(extract_posts_from_markdown(self.fetcher(url), url))
        except Exception as exc:  # noqa: BLE001 - degrade, never crash the pipeline
            return SearchResult.degraded(self.name, f"fetch failed: {exc}")
        return SearchResult(posts=posts, status="ok", message=f"{len(posts)} posts")

Note: queries is accepted for interface symmetry; GitHub repos are pulled whole and filtered downstream by the agent's content-relevance step (the repo URLs themselves are the targeting).

• [ ] Step 4: Run test to verify it passes

Run: cd interview-intelligence && .venv/bin/pytest tests/test_github_connector.py -v Expected: 4 passed.

• [ ] Step 5: Commit

git add interview-intelligence/scripts/connectors/github.py interview-intelligence/tests/test_github_connector.py
git commit -m "feat: add GitHub interview-repo connector with degrade-on-error"

---

Task 6: Dedupe & rank

Files:

• Create: interview-intelligence/scripts/corpus/dedupe_rank.py
• Test: interview-intelligence/tests/test_dedupe_rank.py

MVP dedupe is normalized-text based (lowercase, strip punctuation/whitespace). Embedding-based semantic dedupe is deferred — normalize is the single seam where it would be swapped in later. Duplicates merge: freq sums, source_refs/role_tags union (order-preserving). Output is sorted by freq descending, ties broken by first-seen order.

• [ ] Step 1: Write the failing test

from scripts.models import Question
from scripts.corpus.dedupe_rank import normalize, dedupe_and_rank


def test_normalize_strips_case_punctuation_whitespace():
    assert normalize("  What is  MCP??  ") == normalize("what is mcp")


def test_dedupe_merges_and_sums_freq():
    qs = [
        Question("What is MCP?", ["u1"], role_tags=["agent"]),
        Question("what is  mcp", ["u2"], role_tags=["llm"]),
        Question("What is RAG?", ["u3"]),
    ]
    out = dedupe_and_rank(qs)
    assert len(out) == 2
    top = out[0]
    assert top.freq == 2
    assert top.source_refs == ["u1", "u2"]
    assert top.role_tags == ["agent", "llm"]


def test_rank_sorts_by_freq_desc():
    qs = [
        Question("rare", ["a"]),
        Question("common", ["b"]),
        Question("common", ["c"]),
    ]
    out = dedupe_and_rank(qs)
    assert out[0].text == "common"
    assert out[0].freq == 2
    assert out[1].text == "rare"

• [ ] Step 2: Run test to verify it fails

Run: cd interview-intelligence && .venv/bin/pytest tests/test_dedupe_rank.py -v Expected: FAIL with "No module named 'scripts.corpus.dedupe_rank'".

• [ ] Step 3: Write minimal implementation

import re

from scripts.models import Question

_PUNCT = re.compile(r"[^\w一-鿿]+")


def normalize(text: str) -> str:
    t = text.strip().lower()
    t = _PUNCT.sub(" ", t)
    return " ".join(t.split())


def _union(into: list[str], extra: list[str]) -> None:
    for item in extra:
        if item not in into:
            into.append(item)


def dedupe_and_rank(questions: list[Question]) -> list[Question]:
    merged: dict[str, Question] = {}
    order: list[str] = []
    for q in questions:
        key = normalize(q.text)
        if key not in merged:
            merged[key] = Question(
                text=q.text,
                source_refs=list(q.source_refs),
                freq=q.freq,
                role_tags=list(q.role_tags),
                topic=q.topic,
                modality_origin=q.modality_origin,
            )
            order.append(key)
        else:
            m = merged[key]
            m.freq += q.freq
            _union(m.source_refs, q.source_refs)
            _union(m.role_tags, q.role_tags)
    ranked = sorted(order, key=lambda k: -merged[k].freq)
    return [merged[k] for k in ranked]

• [ ] Step 4: Run test to verify it passes

Run: cd interview-intelligence && .venv/bin/pytest tests/test_dedupe_rank.py -v Expected: 3 passed.

• [ ] Step 5: Commit

git add interview-intelligence/scripts/corpus/dedupe_rank.py interview-intelligence/tests/test_dedupe_rank.py
git commit -m "feat: add normalized-text dedupe and frequency ranking"

---

Task 7: Resume extraction

Files:

• Create: interview-intelligence/scripts/resume_extract.py
• Test: interview-intelligence/tests/test_resume_extract.py
• Test fixture: interview-intelligence/tests/fixtures/sample_resume.pdf

PDFs are extracted to text with pypdf. Image resumes (.png/.jpg/.jpeg/.webp) cannot be parsed deterministically — return needs_vision=True with the asset path so the agent reads the image directly (mirrors the hybrid-OCR fallback policy). A PDF that yields little/no text (scanned image PDF) also sets needs_vision=True.

• [ ] Step 1: Create the PDF fixture

Run (generates a one-page text PDF without extra deps via pypdf's writer is insufficient, so use reportlab-free approach with a committed fixture):

cd interview-intelligence && .venv/bin/python -c "
from pypdf import PdfWriter
# pypdf cannot draw text; create fixture via a minimal embedded PDF string instead
import pathlib, base64
# Minimal single-page PDF containing the text 'Skill driven agent project Python RAG'
pdf = b'''%PDF-1.4
1 0 obj<</Type/Catalog/Pages 2 0 R>>endobj
2 0 obj<</Type/Pages/Kids[3 0 R]/Count 1>>endobj
3 0 obj<</Type/Page/Parent 2 0 R/MediaBox[0 0 300 144]/Contents 4 0 R/Resources<</Font<</F1 5 0 R>>>>>>endobj
4 0 obj<</Length 74>>stream
BT /F1 12 Tf 20 100 Td (Skill driven agent project Python RAG) Tj ET
endstream endobj
5 0 obj<</Type/Font/Subtype/Type1/BaseFont/Helvetica>>endobj
xref
0 6
0000000000 65535 f 
0000000009 00000 n 
0000000052 00000 n 
0000000101 00000 n 
0000000229 00000 n 
0000000353 00000 n 
trailer<</Size 6/Root 1 0 R>>
startxref
422
%%EOF'''
pathlib.Path('tests/fixtures').mkdir(parents=True, exist_ok=True)
pathlib.Path('tests/fixtures/sample_resume.pdf').write_bytes(pdf)
print('wrote fixture')
"

Expected: prints wrote fixture. Then verify it is readable: .venv/bin/python -c "from pypdf import PdfReader; print(repr(PdfReader('tests/fixtures/sample_resume.pdf').pages[0].extract_text()))" — expected output contains Skill driven agent project Python RAG. If extraction returns empty, regenerate; do not proceed until the fixture extracts text.

• [ ] Step 2: Write the failing test

from pathlib import Path

from scripts.resume_extract import extract_resume, ResumeExtraction

FIXTURE = Path(__file__).parent / "fixtures" / "sample_resume.pdf"


def test_pdf_extraction_returns_text():
    result = extract_resume(FIXTURE)
    assert isinstance(result, ResumeExtraction)
    assert "Skill" in result.text
    assert result.needs_vision is False


def test_image_resume_flags_vision(tmp_path):
    img = tmp_path / "resume.png"
    img.write_bytes(b"\x89PNG\r\n\x1a\n")
    result = extract_resume(img)
    assert result.needs_vision is True
    assert result.asset_path == str(img)
    assert result.text == ""


def test_empty_pdf_flags_vision(tmp_path):
    blank = tmp_path / "blank.pdf"
    blank.write_bytes((FIXTURE.read_bytes().replace(b"Skill driven agent project Python RAG", b" ")))
    result = extract_resume(blank)
    assert result.needs_vision is True

• [ ] Step 3: Run test to verify it fails

Run: cd interview-intelligence && .venv/bin/pytest tests/test_resume_extract.py -v Expected: FAIL with "No module named 'scripts.resume_extract'".

• [ ] Step 4: Write minimal implementation

from dataclasses import dataclass
from pathlib import Path

from pypdf import PdfReader

_IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".webp", ".bmp"}
_MIN_TEXT_CHARS = 10


@dataclass
class ResumeExtraction:
    text: str
    needs_vision: bool
    asset_path: str


def extract_resume(path) -> ResumeExtraction:
    p = Path(path)
    ext = p.suffix.lower()
    if ext in _IMAGE_EXTS:
        return ResumeExtraction(text="", needs_vision=True, asset_path=str(p))
    if ext == ".pdf":
        reader = PdfReader(str(p))
        text = "\n".join((page.extract_text() or "") for page in reader.pages).strip()
        if len(text) < _MIN_TEXT_CHARS:
            return ResumeExtraction(text="", needs_vision=True, asset_path=str(p))
        return ResumeExtraction(text=text, needs_vision=False, asset_path=str(p))
    # Plain text / markdown resumes
    text = p.read_text(encoding="utf-8", errors="ignore").strip()
    needs_vision = len(text) < _MIN_TEXT_CHARS
    return ResumeExtraction(text=text, needs_vision=needs_vision, asset_path=str(p))

• [ ] Step 5: Run test to verify it passes

Run: cd interview-intelligence && .venv/bin/pytest tests/test_resume_extract.py -v Expected: 3 passed.

• [ ] Step 6: Commit

git add interview-intelligence/scripts/resume_extract.py interview-intelligence/tests/test_resume_extract.py interview-intelligence/tests/fixtures/sample_resume.pdf
git commit -m "feat: add resume extraction with vision fallback for images/scanned PDFs"

---

Task 8: Full test run + reference docs

Files:

• Create: interview-intelligence/assets/schema.md
• Create: interview-intelligence/references/role_taxonomy.md

• [ ] Step 1: Run the whole suite green

Run: cd interview-intelligence && .venv/bin/pytest -v Expected: all tests from Tasks 2–7 pass (15 tests).

• [ ] Step 2: Write assets/schema.md

# Data Structures

- **RawPost** `{ source, url, post_type(text|image|mixed), raw_text, asset_paths[], comments[] }`
  One scraped unit (a question-like line, post, or image). Produced by connectors.
- **Question** `{ text, source_refs[], freq, role_tags[], topic, modality_origin(text|ocr|vision) }`
  A normalized interview question. Produced by the agent's extraction step from RawPosts,
  then merged/ranked by `corpus/dedupe_rank.py`.
- **FollowUpChain** `{ seed_question, resume_anchor, followups[], is_grounded }`
  A personalized follow-up chain. Produced by the agent's project-anchoring step.
  `is_grounded=false` means it degraded to a plain 八股 question (no resume anchor found).

Persistence: normalized JSON under `corpus_cache/` via `corpus/store.py`.

• [ ] Step 3: Write references/role_taxonomy.md

# Role Taxonomy (seed only)

A *starting* alias list for AI-application roles. This is a SEED for the agent's first-pass
queries, NOT the source of truth. The real role vocabulary is discovered iteratively by
harvesting actual role names/tags from first-pass scrape results (see SKILL.md step 3).

- AI 应用开发 / AI 应用工程师
- AI 创新应用 / AI 产品研发
- Agent 开发 / 智能体应用
- AI 研究开发 / LLM 应用开发
- 大模型应用 / 生成式 AI 工程师

Common underlying skill/topic seeds (more stable than role names):
agent, RAG, MCP, prompt engineering, LLM 应用, 向量检索, function calling, 微调, 评测.

• [ ] Step 4: Commit

git add interview-intelligence/assets/schema.md interview-intelligence/references/role_taxonomy.md
git commit -m "docs: add data-structure schema and role-taxonomy seed reference"

---

Task 9: SKILL.md orchestration

Files:

• Create: interview-intelligence/SKILL.md

This is the agent-facing orchestration contract. No code/tests — it is authored prose that wires the Python tools into the 8-step workflow, GitHub-only for the MVP.

• [ ] Step 1: Write SKILL.md

````markdown --- name: interview-intelligence description: Use when a user wants to prepare for interviews by uploading a resume (PDF/image) and naming a fuzzy target role (e.g. "AI 应用开发"). Broad-net retrieves real interview questions from GitHub interview repos, dedupes/ranks them, and produces a personalized prep package with project-anchored follow-ups. V1 source is GitHub only. ---

Interview Intelligence Skill

Turn a resume + a fuzzy role into a personalized interview prep package grounded in real interview-experience (面经) content. You (the agent) do the reasoning; Python scripts under scripts/ do the deterministic work. Communicate through JSON in corpus_cache/.

Inputs

• Resume: a PDF, image, or text file path.
• Fuzzy role: a direction like "AI 应用开发" (NOT a specific JD).

Tools (run with the package venv: .venv/bin/python)

• scripts/resume_extract.py → extract_resume(path) -> ResumeExtraction{text, needs_vision, asset_path}
• scripts/connectors/github.py → GithubConnector(repo_raw_urls).search(queries) -> SearchResult
• scripts/corpus/store.py → save_raw_posts/load_raw_posts/save_questions/load_questions
• scripts/corpus/dedupe_rank.py → dedupe_and_rank(questions) -> list[Question]
• Data models in scripts/models.py; structures documented in assets/schema.md.

Workflow

1. Resume understanding. Call extract_resume. If needs_vision is true, read the image/PDF yourself with your vision capability. Produce a structured summary: skills, projects (with the techniques each project used), and notable keywords.

2. Seed query generation. From the role direction + the resume's skills/topics, build SEED queries from underlying skills/topics (agent, RAG, MCP, LLM 应用, …), NOT a guessed role-name list. Use references/role_taxonomy.md only as a starting hint.

3. Iterative retrieval (GitHub, V1). Pick relevant interview repos and pass their raw markdown URLs to GithubConnector(repo_raw_urls).search(seed_queries). Save the returned posts with save_raw_posts. Read the results and HARVEST the real role names / tags / recurring terms that actually appear. Re-run with repos/terms the harvest surfaced. Repeat until no new vocabulary emerges. If a connector returns status="degraded", tell the user what it needs and continue with what you have — never block the pipeline. Human-in-the-loop: before the final pass, show the user the directions/terms you discovered from real data and let them add/remove/steer.

4. Content-semantic relevance. Decide each post's relevance by reading its content against the user's role + resume — NOT by whether a role name matched a preset list.

5. Question extraction. Convert relevant RawPosts into normalized Questions (set modality_origin). Save with save_questions.

6. Dedupe & rank. Run dedupe_and_rank(load_questions(...)) and save the ranked result. This is the high-frequency question set.

7. Project-anchored reasoning. For each top question, check whether it connects to a resume project/skill. If yes, build a FollowUpChain (seed → personalized follow-ups, is_grounded=true). Every follow-up MUST trace to (a resume project/skill) + (a real scraped question) — if you cannot ground it, set is_grounded=false and keep it as a plain 八股 question. Do NOT fabricate follow-ups.

8. Prep package. Write a Markdown package: role analysis, gap analysis, high-frequency 八股 questions (with source links), personalized project follow-up chains, and reference approaches. Save it to corpus_cache/prep_package.md and show it to the user.

Constraints

• GitHub is the only source in V1 (牛客/小红书 + OCR come in later plans).
• Third-party scrapers used in later versions (e.g. MediaCrawler) are for personal,

non-commercial use only.

• Grounding over fluency: never invent follow-ups or questions not traceable to real data.

````

• [ ] Step 2: Sanity-check the skill loads (front-matter valid)

Run: cd interview-intelligence && .venv/bin/python -c "import re,sys; t=open('SKILL.md').read(); assert t.startswith('---'); fm=t.split('---')[1]; assert 'name:' in fm and 'description:' in fm; print('SKILL.md front-matter OK')" Expected: prints SKILL.md front-matter OK.

• [ ] Step 3: Commit

git add interview-intelligence/SKILL.md
git commit -m "feat: add SKILL.md orchestration for interview-intelligence MVP"

---

Self-Review

Spec coverage (MVP scope):

• Resume understanding (image/PDF, vision fallback) → Task 7 + SKILL step 1. ✓
• Seed query generation (skills/topics, not role names) → SKILL step 2 + Task 8 taxonomy. ✓
• Iterative data-driven retrieval + vocabulary harvest + HITL → SKILL step 3. ✓
• Content-semantic relevance → SKILL step 4. ✓
• Connector interface + degrade → Task 4; GitHub connector → Task 5. ✓
• Question extraction + normalized records → SKILL step 5 + Task 2 models. ✓
• Dedupe & rank → Task 6. ✓
• Project-anchored reasoning + grounding validation → SKILL step 7 + FollowUpChain (Task 2). ✓
• Prep package output → SKILL step 8. ✓
• Corpus persistence → Task 3. ✓
• Deferred (correctly out of MVP scope): 牛客 connector (Plan 2), 小红书 + hybrid OCR (Plan 3),

embedding-based semantic dedupe, ASR sources, interactive mock.

Placeholder scan: No TBD/TODO; every code step has complete code; commands have expected output.

Type consistency: RawPost/Question/FollowUpChain fields are identical across Tasks 2, 3, 5, 6, and SKILL.md. SearchResult{posts,status,message} consistent in Tasks 4 and 5. extract_resume→ResumeExtraction{text,needs_vision,asset_path} consistent in Task 7 and SKILL.md. dedupe_and_rank / normalize names consistent in Task 6 and SKILL.md.

Interview Intelligence — 牛客 Connector + Recency Implementation Plan (Plan 2)

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Add timeliness to the skill — extract post dates, filter 面经 to the last ~2 years, rank by frequency-and-recency — and add a 牛客 NowCoder connector (the primary timestamped source) that parses post HTML into dated RawPosts with graceful degradation.

Architecture: Builds directly on the V1 package at interview-intelligence/. Adds a posted_at date to RawPost and latest_posted_at to Question, a new corpus/recency.py filter, upgrades corpus/dedupe_rank.py to weight recency, and adds connectors/nowcoder.py (parses recorded sample HTML offline; live fetch is an injectable, degrade-safe wrapper — no real login-wall scraping implemented here). All reasoning stays in SKILL.md.

Tech Stack: Python 3.11, pytest, pypdf, requests, beautifulsoup4 (new — HTML parsing for 牛客 pages).

Scope note: This plan does NOT implement real cookie-based 牛客 login/anti-bot scraping (deferred; the connector exposes an injectable fetcher and degrades cleanly when none is wired). 小红书 + hybrid OCR remain in Plan 3. The recency window is fixed at 2 years (730 days).

Prerequisite: V1 is merged to main (interview-intelligence/scripts/models.py has RawPost/Question; corpus/dedupe_rank.py has normalize/dedupe_and_rank; connectors/base.py has Connector/SearchResult). Work on a branch off main.

---

Task 1: Add beautifulsoup4 dependency

Files:

• Modify: interview-intelligence/requirements.txt

• [ ] Step 1: Append the dependency

Add this line to interview-intelligence/requirements.txt (keep the existing three lines):

beautifulsoup4==4.12.3

The full file must then read:

pypdf==4.3.1
requests==2.32.3
pytest==8.3.2
beautifulsoup4==4.12.3

• [ ] Step 2: Install it

Run: cd interview-intelligence && .venv/bin/pip install -r requirements.txt Expected: installs beautifulsoup4 (and soupsieve) with no error.

• [ ] Step 3: Verify import

Run: cd interview-intelligence && .venv/bin/python -c "import bs4; print(bs4.__version__)" Expected: prints 4.12.3.

• [ ] Step 4: Commit

git add interview-intelligence/requirements.txt
git commit -m "chore: add beautifulsoup4 for HTML parsing"

---

Task 2: Add posted_at to RawPost and latest_posted_at to Question

Files:

• Modify: interview-intelligence/scripts/models.py
• Test: interview-intelligence/tests/test_models.py (add cases)

Both new fields are optional (str | None = None), so existing positional construction in other tests/code keeps working. Dates are ISO YYYY-MM-DD strings or None.

• [ ] Step 1: Add failing tests (append to tests/test_models.py)

def test_rawpost_has_optional_posted_at_defaulting_none():
    post = RawPost("github", "u1", "text", "Q1")
    assert post.posted_at is None
    dated = RawPost("nowcoder", "u2", "text", "Q2", posted_at="2025-09-01")
    assert RawPost.from_dict(dated.to_dict()) == dated
    assert dated.posted_at == "2025-09-01"


def test_question_has_optional_latest_posted_at_defaulting_none():
    q = Question("Q1", ["u1"])
    assert q.latest_posted_at is None
    dated = Question("Q2", ["u2"], latest_posted_at="2025-09-01")
    assert Question.from_dict(dated.to_dict()) == dated

• [ ] Step 2: Run to verify they fail

Run: cd interview-intelligence && .venv/bin/pytest tests/test_models.py -v Expected: FAIL — TypeError: __init__() got an unexpected keyword argument 'posted_at' (and same for latest_posted_at).

• [ ] Step 3: Add the fields in scripts/models.py

In RawPost, add posted_at AFTER raw_text and BEFORE the field(default_factory=list) lines (a non-default field cannot follow defaulted ones, but posted_at has a default so it must sit among the defaulted fields — place it as the first defaulted field):

@dataclass
class RawPost:
    source: str
    url: str
    post_type: str  # text | image | mixed
    raw_text: str
    posted_at: str | None = None  # ISO YYYY-MM-DD, or None if source has no date
    asset_paths: list[str] = field(default_factory=list)
    comments: list[str] = field(default_factory=list)

    def to_dict(self) -> dict:
        return asdict(self)

    @classmethod
    def from_dict(cls, d: dict) -> "RawPost":
        return cls(**d)

In Question, add latest_posted_at after freq:

@dataclass
class Question:
    text: str
    source_refs: list[str] = field(default_factory=list)
    freq: int = 1
    latest_posted_at: str | None = None  # most recent posted_at among merged duplicates
    role_tags: list[str] = field(default_factory=list)
    topic: str = ""
    modality_origin: str = "text"  # text | ocr | vision

    def to_dict(self) -> dict:
        return asdict(self)

    @classmethod
    def from_dict(cls, d: dict) -> "Question":
        return cls(**d)

Leave FollowUpChain unchanged.

• [ ] Step 4: Run the full suite

Run: cd interview-intelligence && .venv/bin/pytest -q Expected: all pass (existing 18 + 2 new = 20). The existing test_dedupe_rank.py still passes because Question field additions have defaults and the merge in dedupe_rank (Task 4) is not yet changed — but note dedupe_and_rank currently reconstructs Question field-by-field WITHOUT latest_posted_at; that still works (it defaults to None). Confirm green before moving on.

• [ ] Step 5: Commit

git add scripts/models.py tests/test_models.py
git commit -m "feat: add posted_at to RawPost and latest_posted_at to Question"

---

Task 3: Recency filter

Files:

• Create: interview-intelligence/scripts/corpus/recency.py
• Test: interview-intelligence/tests/test_recency.py

Filters RawPosts to those within window_days (default 730 = ~2 years) of a reference date. Policy: posts with posted_at=None are KEPT (undated supplementary sources like GitHub must not be silently dropped); only posts with a parseable date OLDER than the window are dropped. Unparseable date strings are treated as None (kept).

• [ ] Step 1: Write the failing test tests/test_recency.py

from datetime import date

from scripts.models import RawPost
from scripts.corpus.recency import filter_recent


def _post(posted_at):
    return RawPost("nowcoder", "u", "text", "Q", posted_at=posted_at)


def test_keeps_recent_drops_old():
    ref = date(2026, 5, 28)
    posts = [_post("2025-09-01"), _post("2023-01-01")]  # within 2y, older than 2y
    kept = filter_recent(posts, window_days=730, today=ref)
    assert [p.posted_at for p in kept] == ["2025-09-01"]


def test_none_dates_are_kept():
    ref = date(2026, 5, 28)
    posts = [_post(None), _post("2010-01-01")]
    kept = filter_recent(posts, window_days=730, today=ref)
    assert [p.posted_at for p in kept] == [None]


def test_unparseable_date_is_kept():
    ref = date(2026, 5, 28)
    posts = [_post("not-a-date")]
    kept = filter_recent(posts, window_days=730, today=ref)
    assert len(kept) == 1


def test_boundary_exactly_window_is_kept():
    ref = date(2026, 5, 28)
    posts = [_post("2024-05-29")]  # 729 days before ref → kept
    kept = filter_recent(posts, window_days=730, today=ref)
    assert len(kept) == 1

• [ ] Step 2: Run to verify it fails

Run: cd interview-intelligence && .venv/bin/pytest tests/test_recency.py -v Expected: FAIL — "No module named 'scripts.corpus.recency'".

• [ ] Step 3: Implement scripts/corpus/recency.py

from datetime import date, datetime

from scripts.models import RawPost


def _parse(posted_at: str | None) -> date | None:
    if not posted_at:
        return None
    try:
        return datetime.strptime(posted_at, "%Y-%m-%d").date()
    except ValueError:
        return None


def filter_recent(
    posts: list[RawPost], window_days: int = 730, today: date | None = None
) -> list[RawPost]:
    ref = today or date.today()
    kept: list[RawPost] = []
    for p in posts:
        d = _parse(p.posted_at)
        if d is None:
            kept.append(p)  # undated/unparseable → keep
            continue
        if (ref - d).days <= window_days:
            kept.append(p)
    return kept

• [ ] Step 4: Run to verify it passes

Run: cd interview-intelligence && .venv/bin/pytest tests/test_recency.py -v Expected: 4 passed.

• [ ] Step 5: Commit

git add scripts/corpus/recency.py tests/test_recency.py
git commit -m "feat: add recency filter (keep posts within ~2 years; keep undated)"

---

Task 4: Upgrade dedupe/rank to carry and weight recency

Files:

• Modify: interview-intelligence/scripts/corpus/dedupe_rank.py
• Test: interview-intelligence/tests/test_dedupe_rank.py (add cases; keep existing)

On merge, the surviving Question keeps the MAX (most recent) latest_posted_at among duplicates (preferring any real date over None). Ranking score = freq * recency_weight(latest_posted_at, today), where weight is: ≤365 days → 1.0; ≤730 days → 0.6; older → 0.3; None → 0.5 (undated content ranks below recently-dated content but above stale-dated content). Sort by score desc; ties broken by first-seen order (stable sort).

• [ ] Step 1: Add failing tests (append to tests/test_dedupe_rank.py)

from datetime import date


def test_merge_keeps_most_recent_date():
    qs = [
        Question("What is MCP?", ["u1"], latest_posted_at="2024-01-01"),
        Question("what is  mcp", ["u2"], latest_posted_at="2025-06-01"),
    ]
    out = dedupe_and_rank(qs, today=date(2026, 5, 28))
    assert len(out) == 1
    assert out[0].latest_posted_at == "2025-06-01"
    assert out[0].freq == 2


def test_recency_weight_can_outrank_lower_freq_when_close():
    # old item freq=2 → score 2*0.3=0.6 ; fresh item freq=1 → score 1*1.0=1.0
    qs = [
        Question("old hot", ["a"], freq=2, latest_posted_at="2023-01-01"),
        Question("fresh", ["b"], freq=1, latest_posted_at="2026-04-01"),
    ]
    out = dedupe_and_rank(qs, today=date(2026, 5, 28))
    assert out[0].text == "fresh"


def test_none_date_weight_between_fresh_and_stale():
    # freq all 1: fresh(1.0) > undated(0.5) > stale(0.3)
    qs = [
        Question("stale", ["a"], latest_posted_at="2022-01-01"),
        Question("undated", ["b"], latest_posted_at=None),
        Question("fresh", ["c"], latest_posted_at="2026-05-01"),
    ]
    out = dedupe_and_rank(qs, today=date(2026, 5, 28))
    assert [q.text for q in out] == ["fresh", "undated", "stale"]

NOTE: the existing tests in this file call dedupe_and_rank(qs) with no today. Keep them working by giving today a default of None (meaning date.today()). The existing test_dedupe_merges_and_sums_freq and test_rank_sorts_by_freq_desc use no dates (all None → equal weight 0.5), so freq alone decides order and they still pass.

• [ ] Step 2: Run to verify new tests fail

Run: cd interview-intelligence && .venv/bin/pytest tests/test_dedupe_rank.py -v Expected: the 3 new tests FAIL (dedupe_and_rank() got an unexpected keyword argument 'today'); existing 3 still pass.

• [ ] Step 3: Rewrite scripts/corpus/dedupe_rank.py

import re
from datetime import date, datetime

from scripts.models import Question

_PUNCT = re.compile(r"[^\w一-鿿]+")


def normalize(text: str) -> str:
    t = text.strip().lower()
    t = _PUNCT.sub(" ", t)
    return " ".join(t.split())


def _union(into: list[str], extra: list[str]) -> None:
    for item in extra:
        if item not in into:
            into.append(item)


def _max_date(a: str | None, b: str | None) -> str | None:
    # Returns the more recent ISO date string; any real date beats None.
    candidates = [d for d in (a, b) if d]
    return max(candidates) if candidates else None


def _recency_weight(posted_at: str | None, today: date) -> float:
    if not posted_at:
        return 0.5
    try:
        d = datetime.strptime(posted_at, "%Y-%m-%d").date()
    except ValueError:
        return 0.5
    days = (today - d).days
    if days <= 365:
        return 1.0
    if days <= 730:
        return 0.6
    return 0.3


def dedupe_and_rank(questions: list[Question], today: date | None = None) -> list[Question]:
    """Merge questions with the same normalized text and rank by frequency and recency.

    Contract: callers pass one Question per occurrence with freq=1; this sums
    incoming freq, so passing pre-aggregated freqs will skew ranking. Score is
    freq * recency_weight(latest_posted_at); ties keep first-seen order.
    """
    ref = today or date.today()
    merged: dict[str, Question] = {}
    order: list[str] = []
    for q in questions:
        key = normalize(q.text)
        if key not in merged:
            merged[key] = Question(
                text=q.text,
                source_refs=list(q.source_refs),
                freq=q.freq,
                latest_posted_at=q.latest_posted_at,
                role_tags=list(q.role_tags),
                topic=q.topic,
                modality_origin=q.modality_origin,
            )
            order.append(key)
        else:
            m = merged[key]
            m.freq += q.freq
            m.latest_posted_at = _max_date(m.latest_posted_at, q.latest_posted_at)
            _union(m.source_refs, q.source_refs)
            _union(m.role_tags, q.role_tags)

    def score(k: str) -> float:
        q = merged[k]
        return q.freq * _recency_weight(q.latest_posted_at, ref)

    ranked = sorted(order, key=lambda k: -score(k))
    return [merged[k] for k in ranked]

• [ ] Step 4: Run the file then full suite

Run: cd interview-intelligence && .venv/bin/pytest tests/test_dedupe_rank.py -v — expected 6 passed. Run: cd interview-intelligence && .venv/bin/pytest -q — expected all green.

• [ ] Step 5: Commit

git add scripts/corpus/dedupe_rank.py tests/test_dedupe_rank.py
git commit -m "feat: weight ranking by recency and carry latest_posted_at through merge"

---

Task 5: 牛客 connector — HTML parsing + date extraction

Files:

• Create: interview-intelligence/scripts/connectors/nowcoder.py
• Test: interview-intelligence/tests/test_nowcoder_connector.py
• Test fixture: interview-intelligence/tests/fixtures/nowcoder_sample.html

Parses a 牛客 discussion/面经 post HTML page into one RawPost: title + body text become raw_text, the post date becomes posted_at (ISO). The fixture mimics 牛客's structure: a title element, a date element, and body paragraphs. Live fetching is an injectable fetcher (defaults to a requests-based one); when no fetcher succeeds the connector degrades. NowCoder needs login for many pages, so the default fetcher will often fail — that is expected and handled by degradation; real cookie-based login is OUT of scope for this task.

• [ ] Step 1: Create the HTML fixture tests/fixtures/nowcoder_sample.html

<!DOCTYPE html>
<html>
<head><title>字节 AI 应用开发 一面面经</title></head>
<body>
  <div class="post-title">字节 AI 应用开发 一面面经</div>
  <span class="post-date">2025-09-15</span>
  <div class="post-content">
    <p>问了 MCP 和 Skill 的区别。</p>
    <p>介绍一下你做过的 agent 项目架构。</p>
    <p>RAG 的检索召回怎么优化？</p>
  </div>
</body>
</html>

• [ ] Step 2: Write the failing test tests/test_nowcoder_connector.py

from pathlib import Path

from scripts.connectors.base import SearchResult
from scripts.connectors.nowcoder import parse_nowcoder_post, NowCoderConnector

FIXTURE = Path(__file__).parent / "fixtures" / "nowcoder_sample.html"
SAMPLE_HTML = FIXTURE.read_text(encoding="utf-8")


def test_parse_extracts_text_and_date():
    post = parse_nowcoder_post(SAMPLE_HTML, "https://nowcoder.com/p/1")
    assert post.source == "nowcoder"
    assert post.url == "https://nowcoder.com/p/1"
    assert post.posted_at == "2025-09-15"
    assert "MCP 和 Skill 的区别" in post.raw_text
    assert "agent 项目架构" in post.raw_text
    assert "字节 AI 应用开发 一面面经" in post.raw_text


def test_connector_search_uses_injected_fetcher():
    conn = NowCoderConnector(
        post_urls=["https://nowcoder.com/p/1"],
        fetcher=lambda url: SAMPLE_HTML,
    )
    result = conn.search(["agent"])
    assert result.status == "ok"
    assert len(result.posts) == 1
    assert result.posts[0].posted_at == "2025-09-15"


def test_connector_degrades_on_fetch_error():
    def boom(url):
        raise RuntimeError("login wall")

    conn = NowCoderConnector(post_urls=["https://nowcoder.com/p/1"], fetcher=boom)
    result = conn.search(["agent"])
    assert result.status == "degraded"
    assert result.posts == []
    assert "cookie" in result.message.lower()


def test_parse_missing_date_yields_none():
    html = "<div class='post-title'>T</div><div class='post-content'><p>body text here</p></div>"
    post = parse_nowcoder_post(html, "https://nowcoder.com/p/2")
    assert post.posted_at is None
    assert "body text here" in post.raw_text

• [ ] Step 3: Run to verify it fails

Run: cd interview-intelligence && .venv/bin/pytest tests/test_nowcoder_connector.py -v Expected: FAIL — "No module named 'scripts.connectors.nowcoder'".

• [ ] Step 4: Implement scripts/connectors/nowcoder.py

import re
from collections.abc import Callable

import requests
from bs4 import BeautifulSoup

from scripts.connectors.base import Connector, SearchResult
from scripts.models import RawPost

_ISO_DATE = re.compile(r"\d{4}-\d{2}-\d{2}")


def parse_nowcoder_post(html: str, url: str) -> RawPost:
    soup = BeautifulSoup(html, "html.parser")

    title_el = soup.select_one(".post-title")
    title = title_el.get_text(strip=True) if title_el else ""

    date_el = soup.select_one(".post-date")
    posted_at = None
    if date_el:
        m = _ISO_DATE.search(date_el.get_text(strip=True))
        if m:
            posted_at = m.group(0)

    content_el = soup.select_one(".post-content")
    if content_el:
        paras = [p.get_text(strip=True) for p in content_el.find_all("p")]
        body = "\n".join(p for p in paras if p)
    else:
        body = ""

    raw_text = (title + "\n" + body).strip() if title else body.strip()
    return RawPost(
        source="nowcoder",
        url=url,
        post_type="text",
        raw_text=raw_text,
        posted_at=posted_at,
    )


def _default_fetcher(url: str) -> str:
    resp = requests.get(url, timeout=30)
    resp.raise_for_status()
    return resp.text


class NowCoderConnector(Connector):
    name = "nowcoder"

    def __init__(self, post_urls: list[str], fetcher: Callable[[str], str] | None = None):
        self.post_urls = post_urls
        self.fetcher = fetcher or _default_fetcher

    def search(self, queries: list[str]) -> SearchResult:
        posts: list[RawPost] = []
        try:
            for url in self.post_urls:
                posts.append(parse_nowcoder_post(self.fetcher(url), url))
        except Exception as exc:  # noqa: BLE001 - degrade, never crash the pipeline
            return SearchResult.degraded(
                self.name,
                f"fetch failed ({exc}); 牛客需要登录，请提供 cookie 或手动粘贴帖子链接/内容",
            )
        return SearchResult(posts=posts, status="ok", message=f"{len(posts)} posts")

Note: queries is accepted for interface symmetry; the connector fetches the given post_urls (the agent supplies URLs it discovered). Do NOT implement search-by-keyword crawling or login here.

• [ ] Step 5: Run to verify it passes

Run: cd interview-intelligence && .venv/bin/pytest tests/test_nowcoder_connector.py -v Expected: 4 passed.

• [ ] Step 6: Commit

git add scripts/connectors/nowcoder.py tests/test_nowcoder_connector.py tests/fixtures/nowcoder_sample.html
git commit -m "feat: add 牛客 connector parsing post HTML to dated RawPost with degrade"

---

Task 6: Update SKILL.md for recency + 牛客

Files:

• Modify: interview-intelligence/SKILL.md

• [ ] Step 1: Update the tools list

In the ## 工具 section, add these two lines (after the github.py line and after the dedupe_rank.py line respectively):

- `scripts/connectors/nowcoder.py` → `NowCoderConnector(post_urls).search(queries) -> SearchResult`
- `scripts/corpus/recency.py` → `filter_recent(posts, window_days=730, today=None) -> list[RawPost]`

• [ ] Step 2: Replace workflow step 3 source guidance

Find step 3. **迭代检索(V1 用 GitHub)。** and replace its first sentence so it reads (keep the rest of the step — harvest loop, degrade, HITL — intact):

3. **迭代检索。** 源的优先级:**牛客(主力,带时间戳)> GitHub(补充,常过时)**。把发现的牛客帖子 URL 传给 `NowCoderConnector(post_urls).search(...)`,把 GitHub 仓库 raw URL 传给 `GithubConnector(repo_raw_urls).search(...)`。两者结果都用 `save_raw_posts` 落盘。读取结果，**收割真实出现的岗位名 / 标签 / 高频术语**，再用收割到的词跑下一轮，直到不再冒出新词。若某 connector 返回 `status="degraded"`（例如牛客需要 cookie），把它需要的东西告诉用户；牛客降级会显著影响时效性，必须明确提示用户，不要默默用 GitHub 凑数。

• [ ] Step 3: Insert a recency-filter step

Immediately AFTER step 5. (题目抽取) and BEFORE step 6. (去重 & 排序), insert:

5b. **时效过滤。** 在抽取出题目之前，先用 `filter_recent(raw_posts)` 把超过约两年的帖子丢掉（默认窗口 730 天；`posted_at` 为 None 的无日期帖子保留）。时效性是硬需求——过时的面经没有价值。

• [ ] Step 4: Update step 6 ranking wording and the constraints

Change step 6. text to mention recency:

6. **去重 & 排序。** 跑 `dedupe_and_rank(load_questions(...))` 并保存排序结果。排序同时考虑**频次和时效**（近期题加权更高），这就是高频题集。

In the ## 约束 section, replace the line - V1 只有 GitHub 一个源(牛客/小红书 + OCR 在后续 plan)。 with:

- 当前源:牛客(主力)+ GitHub(补充)。小红书 + OCR 在后续 plan。
- 时效性是硬需求:只保留近两年的面经,排序向近期加权。

• [ ] Step 5: Validate front-matter and commit

Run: cd interview-intelligence && .venv/bin/python -c "t=open('SKILL.md').read(); assert t.startswith('---'); fm=t.split('---')[1]; assert 'name:' in fm and 'description:' in fm; print('OK')" Expected: prints OK.

git add SKILL.md
git commit -m "docs: wire 牛客 connector and recency filtering into SKILL.md workflow"

---

Task 7: Full suite + schema doc update

Files:

• Modify: interview-intelligence/assets/schema.md

• [ ] Step 1: Run the whole suite green

Run: cd interview-intelligence && .venv/bin/pytest -q Expected: all tests pass (V1's 18 + Task2's 2 + Task3's 4 + Task4's 3 + Task5's 4 = 31).

• [ ] Step 2: Update `assets/schema.md`

Replace the RawPost and Question bullet lines with:

- **RawPost** `{ source, url, post_type(text|image|mixed), raw_text, posted_at(ISO date|null), asset_paths[], comments[] }`
  One scraped unit. `posted_at` is the source post date (ISO `YYYY-MM-DD`) or null for undated
  sources. Produced by connectors. Filtered to the recency window by `corpus/recency.py`.
- **Question** `{ text, source_refs[], freq, latest_posted_at(ISO date|null), role_tags[], topic, modality_origin(text|ocr|vision) }`
  A normalized interview question. `latest_posted_at` is the most recent date among merged
  duplicates. Produced by the agent's extraction step, then merged/ranked (by frequency AND
  recency) by `corpus/dedupe_rank.py`.

• [ ] Step 3: Commit

git add assets/schema.md
git commit -m "docs: document posted_at/latest_posted_at and recency in schema"

---

Self-Review

Spec coverage:

• Recency as first-class (filter + weighted ranking) → Tasks 3, 4 + SKILL steps 5b/6. ✓
• posted_at on RawPost / latest_posted_at on Question → Task 2. ✓
• 牛客 primary connector with date extraction + degrade → Task 5. ✓
• GitHub demoted to supplementary; degrade surfaced not silently tolerated → SKILL step 3 (Task 6). ✓
• Rejected stale offline corpus → not built (correct); spec §6 records the rejection. ✓
• Schema doc updated → Task 7. ✓
• Deferred (correctly out of scope): real 牛客 cookie/login scraping; 小红书 + hybrid OCR (Plan 3);

ASR sources; interactive mock.

Placeholder scan: No TBD/TODO; every code step has complete code; commands have expected output.

Type consistency:

• RawPost(... posted_at=...) and Question(... latest_posted_at=...) field names identical across

Tasks 2, 3, 4, 5 and the schema doc.

• filter_recent(posts, window_days=730, today=None) signature identical in Task 3 and SKILL/Task 6.
• dedupe_and_rank(questions, today=None) signature identical in Task 4, its tests, and SKILL step 6.
• NowCoderConnector(post_urls, fetcher=None).search(queries) -> SearchResult and

parse_nowcoder_post(html, url) -> RawPost consistent in Task 5 and SKILL tools list (Task 6).

• SearchResult / Connector reused from V1 unchanged.

Interview Intelligence — 小红书 Connector + Hybrid OCR Implementation Plan (Plan 3)

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Add the 小红书 image source — a connector that turns a MediaCrawler JSON export into dated image-RawPosts, plus a hybrid OCR module that coarse-reads images and falls back to the agent's vision when confidence is low.

Architecture: Builds on V1 + Plan 2 at interview-intelligence/. Adds connectors/xiaohongshu.py (parses a MediaCrawler export — offline, no live login) producing RawPost(post_type="image", asset_paths=[...], posted_at=...), and ocr/extract.py with an injectable OCR engine and a vision fallback (mirrors the established degrade/vision-fallback pattern in resume_extract.py). Reasoning stays in SKILL.md.

Tech Stack: Python 3.11, pytest. No new pip dependency. (A real OCR engine such as PaddleOCR/Tesseract is injectable but intentionally NOT wired here.)

Scope note (intentional deferrals, same discipline as Plan 2's 牛客):

• No live MediaCrawler login/scraping is implemented. The connector ingests an already-produced MediaCrawler JSON export; when the export is missing/unreadable it degrades cleanly.
• No OCR engine binary is bundled. extract_text_from_image takes an injectable engine; with no engine, or when the engine's confidence is below threshold, it flags needs_vision=True so the agent reads the image with its own vision. Wiring PaddleOCR/Tesseract is a documented future config step.
• Image downloading (turning remote asset_paths URLs into local files) is out of scope; asset_paths hold the export's image references as-is.

Prerequisite: Plan 2 merged to main. Existing: scripts/models.py (RawPost with posted_at, post_type, asset_paths), scripts/connectors/base.py (Connector, SearchResult.degraded), scripts/connectors/{github,nowcoder}.py (established connector pattern), scripts/resume_extract.py (established vision-fallback pattern). Work on a branch off main.

---

Task 1: Hybrid OCR module

Files:

• Create: interview-intelligence/scripts/ocr/__init__.py (empty)
• Create: interview-intelligence/scripts/ocr/extract.py
• Test: interview-intelligence/tests/test_ocr_extract.py

extract_text_from_image(path, engine=None, min_confidence=0.6) returns OcrResult{text, confidence, needs_vision}. An engine is any callable (path: str) -> tuple[str, float] returning (text, confidence). Policy: no engine → needs_vision=True, text="", confidence=0.0. Engine present → call it; if confidence >= min_confidence AND text is non-empty → needs_vision=False; otherwise needs_vision=True (keep whatever text the engine returned as a hint). This is the hybrid: coarse OCR when confident, agent-vision fallback otherwise.

• [ ] Step 1: Create the package marker

Create interview-intelligence/scripts/ocr/__init__.py as an empty file.

• [ ] Step 2: Write the failing test tests/test_ocr_extract.py

from scripts.ocr.extract import extract_text_from_image, OcrResult


def test_no_engine_flags_vision():
    result = extract_text_from_image("img.png")
    assert isinstance(result, OcrResult)
    assert result.needs_vision is True
    assert result.text == ""
    assert result.confidence == 0.0


def test_confident_engine_returns_text():
    engine = lambda path: ("什么是 RAG？", 0.95)
    result = extract_text_from_image("img.png", engine=engine)
    assert result.needs_vision is False
    assert result.text == "什么是 RAG？"
    assert result.confidence == 0.95


def test_low_confidence_flags_vision_but_keeps_hint():
    engine = lambda path: ("blurry guess", 0.30)
    result = extract_text_from_image("img.png", engine=engine, min_confidence=0.6)
    assert result.needs_vision is True
    assert result.text == "blurry guess"


def test_empty_text_flags_vision_even_if_confident():
    engine = lambda path: ("", 0.99)
    result = extract_text_from_image("img.png", engine=engine)
    assert result.needs_vision is True

• [ ] Step 3: Run to verify it fails

Run: cd interview-intelligence && .venv/bin/pytest tests/test_ocr_extract.py -v Expected: FAIL — "No module named 'scripts.ocr.extract'".

• [ ] Step 4: Implement scripts/ocr/extract.py

from collections.abc import Callable
from dataclasses import dataclass

OcrEngine = Callable[[str], tuple[str, float]]


@dataclass
class OcrResult:
    text: str
    confidence: float
    needs_vision: bool


def extract_text_from_image(
    path: str, engine: OcrEngine | None = None, min_confidence: float = 0.6
) -> OcrResult:
    if engine is None:
        return OcrResult(text="", confidence=0.0, needs_vision=True)
    text, confidence = engine(path)
    needs_vision = confidence < min_confidence or not text.strip()
    return OcrResult(text=text, confidence=confidence, needs_vision=needs_vision)

• [ ] Step 5: Run to verify it passes

Run: cd interview-intelligence && .venv/bin/pytest tests/test_ocr_extract.py -v Expected: 4 passed.

• [ ] Step 6: Commit

git add scripts/ocr/__init__.py scripts/ocr/extract.py tests/test_ocr_extract.py
git commit -m "feat: add hybrid OCR module (injectable engine + vision fallback)"

---

Task 2: 小红书 connector — MediaCrawler export parsing

Files:

• Create: interview-intelligence/scripts/connectors/xiaohongshu.py
• Test: interview-intelligence/tests/test_xiaohongshu_connector.py
• Test fixture: interview-intelligence/tests/fixtures/xhs_mediacrawler_export.json

Parses a MediaCrawler 小红书 JSON export (a list of note objects) into RawPosts. Field mapping per note: note_url→url; title+desc→raw_text (joined, stripped); image_list (list of image URLs)→asset_paths; time (epoch milliseconds)→posted_at (ISO YYYY-MM-DD, UTC); post_type="image"; source="xiaohongshu". Missing/zero time→posted_at=None. The connector reads the export via an injectable loader (defaults to reading the file); it degrades when the loader fails (export missing → MediaCrawler hasn't been run / needs login).

• [ ] Step 1: Create the fixture tests/fixtures/xhs_mediacrawler_export.json

[
  {
    "note_id": "n1",
    "note_url": "https://www.xiaohongshu.com/explore/n1",
    "title": "字节 AI 应用开发 面经",
    "desc": "一面问了 MCP 和 Skill 的区别，还有 agent 项目细节。",
    "time": 1758326400000,
    "image_list": [
      "https://sns-img.xhs.cn/n1_a.jpg",
      "https://sns-img.xhs.cn/n1_b.jpg"
    ]
  },
  {
    "note_id": "n2",
    "note_url": "https://www.xiaohongshu.com/explore/n2",
    "title": "无日期帖",
    "desc": "RAG 检索优化怎么答。",
    "time": 0,
    "image_list": ["https://sns-img.xhs.cn/n2_a.jpg"]
  }
]

Note: time 1758326400000 ms = 2025-09-20T00:00:00Z → posted_at "2025-09-20".

• [ ] Step 2: Write the failing test tests/test_xiaohongshu_connector.py

from pathlib import Path

from scripts.connectors.base import SearchResult
from scripts.connectors.xiaohongshu import (
    parse_mediacrawler_export,
    XiaohongshuConnector,
)

FIXTURE = Path(__file__).parent / "fixtures" / "xhs_mediacrawler_export.json"
SAMPLE_JSON = FIXTURE.read_text(encoding="utf-8")


def test_parse_maps_notes_to_image_rawposts():
    posts = parse_mediacrawler_export(SAMPLE_JSON)
    assert len(posts) == 2
    first = posts[0]
    assert first.source == "xiaohongshu"
    assert first.url == "https://www.xiaohongshu.com/explore/n1"
    assert first.post_type == "image"
    assert first.posted_at == "2025-09-20"
    assert first.asset_paths == [
        "https://sns-img.xhs.cn/n1_a.jpg",
        "https://sns-img.xhs.cn/n1_b.jpg",
    ]
    assert "MCP 和 Skill 的区别" in first.raw_text
    assert "字节 AI 应用开发 面经" in first.raw_text


def test_parse_zero_time_yields_none_date():
    posts = parse_mediacrawler_export(SAMPLE_JSON)
    assert posts[1].posted_at is None


def test_connector_search_uses_injected_loader():
    conn = XiaohongshuConnector(export_path="whatever.json", loader=lambda p: SAMPLE_JSON)
    result = conn.search(["agent"])
    assert result.status == "ok"
    assert len(result.posts) == 2
    assert result.posts[0].posted_at == "2025-09-20"


def test_connector_degrades_when_loader_fails():
    def boom(path):
        raise FileNotFoundError("no export")

    conn = XiaohongshuConnector(export_path="missing.json", loader=boom)
    result = conn.search(["agent"])
    assert result.status == "degraded"
    assert result.posts == []
    assert "mediacrawler" in result.message.lower()

• [ ] Step 3: Run to verify it fails

Run: cd interview-intelligence && .venv/bin/pytest tests/test_xiaohongshu_connector.py -v Expected: FAIL — "No module named 'scripts.connectors.xiaohongshu'".

• [ ] Step 4: Implement scripts/connectors/xiaohongshu.py

import json
from collections.abc import Callable
from datetime import datetime, timezone
from pathlib import Path

from scripts.connectors.base import Connector, SearchResult
from scripts.models import RawPost


def _epoch_ms_to_iso(ms) -> str | None:
    if not ms:
        return None
    try:
        dt = datetime.fromtimestamp(int(ms) / 1000, tz=timezone.utc)
    except (ValueError, TypeError, OSError):
        return None
    return dt.date().isoformat()


def parse_mediacrawler_export(json_text: str) -> list[RawPost]:
    notes = json.loads(json_text)
    posts: list[RawPost] = []
    for note in notes:
        title = (note.get("title") or "").strip()
        desc = (note.get("desc") or "").strip()
        raw_text = "\n".join(part for part in (title, desc) if part)
        posts.append(
            RawPost(
                source="xiaohongshu",
                url=note.get("note_url", ""),
                post_type="image",
                raw_text=raw_text,
                posted_at=_epoch_ms_to_iso(note.get("time")),
                asset_paths=list(note.get("image_list") or []),
            )
        )
    return posts


def _default_loader(path: str) -> str:
    return Path(path).read_text(encoding="utf-8")


class XiaohongshuConnector(Connector):
    name = "xiaohongshu"

    def __init__(self, export_path: str, loader: Callable[[str], str] | None = None):
        self.export_path = export_path
        self.loader = loader or _default_loader

    def search(self, queries: list[str]) -> SearchResult:
        try:
            posts = parse_mediacrawler_export(self.loader(self.export_path))
        except Exception as exc:  # noqa: BLE001 - degrade, never crash the pipeline
            return SearchResult.degraded(
                self.name,
                f"无法读取 MediaCrawler 导出 ({exc});请先用 MediaCrawler 登录并采集小红书笔记，导出 JSON 后再试",
            )
        return SearchResult(posts=posts, status="ok", message=f"{len(posts)} posts")

Note: queries is accepted for interface symmetry; the connector ingests the given export file (the user runs MediaCrawler with their own keywords). Do NOT implement live scraping or login here.

• [ ] Step 5: Run to verify it passes

Run: cd interview-intelligence && .venv/bin/pytest tests/test_xiaohongshu_connector.py -v Expected: 4 passed.

• [ ] Step 6: Commit

git add scripts/connectors/xiaohongshu.py tests/test_xiaohongshu_connector.py tests/fixtures/xhs_mediacrawler_export.json
git commit -m "feat: add 小红书 connector parsing MediaCrawler export to dated image RawPosts"

---

Task 3: Update SKILL.md for 小红书 + OCR

Files:

• Modify: interview-intelligence/SKILL.md

• [ ] Step 1: Add tools — in the ## 工具 section, add these lines (after the nowcoder.py line and after the dedupe_rank.py line respectively):

- `scripts/connectors/xiaohongshu.py` → `XiaohongshuConnector(export_path).search(queries) -> SearchResult`
- `scripts/ocr/extract.py` → `extract_text_from_image(path, engine=None, min_confidence=0.6) -> OcrResult{text, confidence, needs_vision}`

• [ ] Step 2: Update the source-priority sentence in step 3. Find the line that begins 3. **迭代检索。** 源的优先级: and replace just that first sentence so it reads:

3. **迭代检索。** 源的优先级:**牛客 + 小红书(主力,带时间戳)> GitHub(补充,常过时)**。把发现的牛客帖子 URL 传给 `NowCoderConnector(post_urls).search(...)`;小红书先用 MediaCrawler 采集导出 JSON,再传给 `XiaohongshuConnector(export_path).search(...)`;把 GitHub 仓库 raw URL 传给 `GithubConnector(repo_raw_urls).search(...)`。三者结果都用 `save_raw_posts` 落盘。读取结果,**收割真实出现的岗位名 / 标签 / 高频术语**,再用收割到的词跑下一轮,直到不再冒出新词。若某 connector 返回 `status="degraded"`(例如牛客需要 cookie、小红书需要先跑 MediaCrawler),把它需要的东西告诉用户;主力源降级会显著影响时效性,必须明确提示用户,不要默默用 GitHub 凑数。

• [ ] Step 3: Update step 5 (题目抽取) to cover image posts via OCR. Replace the step-5 line with:

5. **题目抽取。** 文本类 RawPost 直接读。**图片类 RawPost(小红书,`post_type="image"`)**:对每个 `asset_paths` 里的图片调用 `extract_text_from_image(path, engine)`;若返回 `needs_vision=True`(没接 OCR 引擎或置信度低),就用你自己的视觉能力直接读这张图。把读到的题目转成标准化的 `Question`,图片来源的设 `modality_origin="ocr"` 或 `"vision"`。用 `save_questions` 落盘。

• [ ] Step 4: Update the constraints. Replace the line - 当前源:牛客(主力)+ GitHub(补充)。小红书 + OCR 在后续 plan。 with:

- 当前源:牛客 + 小红书(主力,带时间戳)+ GitHub(补充)。
- 小红书走 MediaCrawler 采集导出,OCR 采用混合策略(粗 OCR + 视觉回退);MediaCrawler 仅供个人、非商业用途。

• [ ] Step 5: Validate front-matter and commit

Run: cd interview-intelligence && .venv/bin/python -c "t=open('SKILL.md').read(); assert t.startswith('---'); fm=t.split('---')[1]; assert 'name:' in fm and 'description:' in fm; print('OK')" Expected: prints OK.

git add SKILL.md
git commit -m "docs: wire 小红书 connector and hybrid OCR into SKILL.md workflow"

---

Task 4: Full suite + schema doc update

Files:

• Modify: interview-intelligence/assets/schema.md

• [ ] Step 1: Run the whole suite green

Run: cd interview-intelligence && .venv/bin/pytest -q Expected: all pass (Plan 2's 31 + Task1's 4 + Task2's 4 = 39).

• [ ] Step 2: Add an image-post note to `assets/schema.md`. Append this paragraph to the end of the file:

## Image posts & OCR

Image-based posts (小红书) use `post_type="image"`, carry image references in `asset_paths`, and
usually have empty/short `raw_text` (caption only). Their questions are extracted by
`ocr/extract.py` `extract_text_from_image(path, engine=None)`: a coarse OCR engine when one is
wired and confident, otherwise `needs_vision=True` and the agent reads the image directly. The
resulting `Question.modality_origin` is `"ocr"` or `"vision"` accordingly.

• [ ] Step 3: Commit

git add assets/schema.md
git commit -m "docs: document image posts and hybrid OCR in schema"

---

Self-Review

Spec coverage:

• 小红书 source (image-based, MediaCrawler adapter) → Task 2 + SKILL step 3 (Task 3). ✓
• Hybrid OCR (coarse OCR + vision fallback) → Task 1 + SKILL step 5 (Task 3). ✓
• posted_at carried from 小红书 (epoch ms → ISO) → Task 2; feeds the existing recency filter/ranking from Plan 2. ✓
• MediaCrawler non-commercial constraint recorded → SKILL constraints (Task 3). ✓
• Connector degrade pattern preserved (export missing) → Task 2. ✓
• Schema doc updated for image posts/OCR → Task 4. ✓
• Deferred (correctly out of scope): live MediaCrawler login/scraping; real OCR engine wiring;

image downloading; ASR sources (抖音/B站); interactive mock — all noted in the scope section.

Placeholder scan: No TBD/TODO; every code step has complete code; commands have expected output.

Type consistency:

• RawPost(source, url, post_type, raw_text, posted_at=..., asset_paths=...) field names match the

V1/Plan-2 model used in Task 2.

• OcrResult{text, confidence, needs_vision} and `extract_text_from_image(path, engine=None,

min_confidence=0.6)` identical across Task 1, its tests, SKILL tools list, and schema doc.

• parse_mediacrawler_export(json_text) -> list[RawPost] and

XiaohongshuConnector(export_path, loader=None).search(queries) -> SearchResult consistent in Task 2 and SKILL tools list (Task 3).

• SearchResult / Connector / SearchResult.degraded reused unchanged from V1.

Interview Intelligence — MediaCrawler Adapter + Setup Doc (Plan 4)

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Ship a file-level adapter that turns MediaCrawler's native 小红书 notes JSON into the schema Plan 3's XiaohongshuConnector already eats, plus a user-facing setup doc and a tiny SKILL.md tie-in.

Architecture: Pure file IO. No network. No subprocess. No MediaCrawler dependency. New module scripts/scrape/normalize_xhs.py (function + __main__ CLI). Tested with hand-written fixtures, and an end-to-end fixture test that feeds adapter output into Plan 3's parse_mediacrawler_export. Setup doc lives at docs/setup_mediacrawler.md.

Tech Stack: Python 3.11, pytest. No new pip dependency.

Scope discipline (carried from Plans 2/3): Do not vendor MediaCrawler. Do not implement login/cookies/scheduling. Do not download images. Do not change 牛客/GitHub/OCR modules.

Prerequisite: Plans 1/2/3 merged to main. Existing: interview-intelligence/scripts/connectors/xiaohongshu.py with parse_mediacrawler_export(json_text); RawPost model. Work on a branch off main.

---

Task 1: Normalize adapter — pure function

Files:

• Create: interview-intelligence/scripts/scrape/__init__.py (empty)
• Create: interview-intelligence/scripts/scrape/normalize_xhs.py
• Test: interview-intelligence/tests/test_normalize_xhs.py
• Test fixture: interview-intelligence/tests/fixtures/mc_xhs_raw.json

Pure function normalize(notes: list[dict]) -> list[dict]. Field map per the spec §4:

• note_url missing → synthesize https://www.xiaohongshu.com/explore/<note_id>
• both note_id and note_url missing → SKIP that note (do not raise)
• image_list: string → split(",") strip empties; list → keep as-is; missing → []
• time: int → keep; missing / non-int → 0
• title/desc missing → ""
• Drop all other MediaCrawler keys (liked_count, comments, tag_list, type, …)
• Preserve input order

Output dict shape (exactly the keys XiaohongshuConnector.parse_mediacrawler_export reads):

{
    "note_id": "...",  # carried through (Plan 3 doesn't require it but harmless)
    "note_url": "...",
    "title": "...",
    "desc": "...",
    "time": 1758326400000,
    "image_list": ["url1", "url2"],
}

• [ ] Step 1: Create the package marker

Create interview-intelligence/scripts/scrape/__init__.py as an empty file.

• [ ] Step 2: Create the fixture interview-intelligence/tests/fixtures/mc_xhs_raw.json

[
  {
    "note_id": "n1",
    "type": "normal",
    "title": "字节 AI 应用开发 面经",
    "desc": "一面问了 MCP 和 Skill 的区别。",
    "time": 1758326400000,
    "note_url": "https://www.xiaohongshu.com/explore/n1",
    "image_list": ["https://sns-img.xhs.cn/n1_a.jpg", "https://sns-img.xhs.cn/n1_b.jpg"],
    "tag_list": "面经,实习",
    "liked_count": "1.2万",
    "comments": [{"text": "mark"}]
  },
  {
    "note_id": "n2",
    "title": "RAG 八股",
    "desc": "讲讲检索优化。",
    "time": 0,
    "image_list": "https://sns-img.xhs.cn/n2_a.jpg,https://sns-img.xhs.cn/n2_b.jpg,"
  },
  {
    "note_id": "n3",
    "title": "无 URL 但有 id",
    "desc": "",
    "time": 1700000000000,
    "image_list": []
  },
  {
    "title": "无 id 无 url,应被跳过",
    "desc": "garbage",
    "time": 1700000000000,
    "image_list": []
  },
  {
    "note_id": "n5",
    "time": "not-a-number",
    "image_list": null
  }
]

• [ ] Step 3: Write the failing test interview-intelligence/tests/test_normalize_xhs.py

import json
from pathlib import Path

from scripts.scrape.normalize_xhs import normalize

FIXTURE = Path(__file__).parent / "fixtures" / "mc_xhs_raw.json"


def _load():
    return json.loads(FIXTURE.read_text(encoding="utf-8"))


def test_skips_notes_with_no_id_and_no_url():
    out = normalize(_load())
    # input has 5 notes, one of which has neither note_id nor note_url
    assert len(out) == 4
    assert "garbage" not in {n["desc"] for n in out}


def test_passes_through_normal_note():
    out = normalize(_load())
    first = out[0]
    assert first["note_id"] == "n1"
    assert first["note_url"] == "https://www.xiaohongshu.com/explore/n1"
    assert first["title"] == "字节 AI 应用开发 面经"
    assert first["desc"].startswith("一面")
    assert first["time"] == 1758326400000
    assert first["image_list"] == [
        "https://sns-img.xhs.cn/n1_a.jpg",
        "https://sns-img.xhs.cn/n1_b.jpg",
    ]


def test_splits_comma_image_list_and_strips_empties():
    out = normalize(_load())
    n2 = next(n for n in out if n["note_id"] == "n2")
    assert n2["image_list"] == [
        "https://sns-img.xhs.cn/n2_a.jpg",
        "https://sns-img.xhs.cn/n2_b.jpg",
    ]


def test_synthesizes_url_from_note_id_when_url_missing():
    out = normalize(_load())
    n3 = next(n for n in out if n["note_id"] == "n3")
    assert n3["note_url"] == "https://www.xiaohongshu.com/explore/n3"


def test_drops_unknown_keys():
    out = normalize(_load())
    first = out[0]
    assert "liked_count" not in first
    assert "comments" not in first
    assert "tag_list" not in first
    assert "type" not in first


def test_invalid_time_becomes_zero_and_null_image_list_becomes_empty():
    out = normalize(_load())
    n5 = next(n for n in out if n["note_id"] == "n5")
    assert n5["time"] == 0
    assert n5["image_list"] == []
    assert n5["title"] == ""
    assert n5["desc"] == ""


def test_preserves_input_order():
    out = normalize(_load())
    assert [n["note_id"] for n in out] == ["n1", "n2", "n3", "n5"]

• [ ] Step 4: Run to verify it fails

Run: cd interview-intelligence && .venv/bin/pytest tests/test_normalize_xhs.py -v Expected: FAIL — "No module named 'scripts.scrape.normalize_xhs'".

• [ ] Step 5: Implement interview-intelligence/scripts/scrape/normalize_xhs.py

"""Normalize MediaCrawler's native 小红书 notes export into the schema that
`scripts/connectors/xiaohongshu.py:parse_mediacrawler_export` consumes.

Field assumptions about MediaCrawler's output (based on the public repo,
NanmiCoder/MediaCrawler, 小红书 module). If MediaCrawler changes their schema,
only this file needs to be touched.
"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

_NOTE_URL_TEMPLATE = "https://www.xiaohongshu.com/explore/{note_id}"


def _coerce_image_list(value) -> list[str]:
    if value is None:
        return []
    if isinstance(value, list):
        return [str(v) for v in value if v]
    if isinstance(value, str):
        return [part.strip() for part in value.split(",") if part.strip()]
    return []


def _coerce_time(value) -> int:
    if isinstance(value, int):
        return value
    if isinstance(value, str):
        try:
            return int(value)
        except ValueError:
            return 0
    return 0


def normalize(notes: list[dict]) -> list[dict]:
    out: list[dict] = []
    for note in notes:
        note_id = note.get("note_id")
        note_url = note.get("note_url")
        if not note_id and not note_url:
            continue
        if not note_url:
            note_url = _NOTE_URL_TEMPLATE.format(note_id=note_id)
        out.append(
            {
                "note_id": note_id or "",
                "note_url": note_url,
                "title": note.get("title") or "",
                "desc": note.get("desc") or "",
                "time": _coerce_time(note.get("time")),
                "image_list": _coerce_image_list(note.get("image_list")),
            }
        )
    return out


def _main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Normalize MediaCrawler 小红书 notes JSON into XiaohongshuConnector input."
    )
    parser.add_argument("input", help="Path to MediaCrawler notes JSON.")
    parser.add_argument("-o", "--output", required=True, help="Path to write normalized JSON.")
    args = parser.parse_args(argv)

    raw = json.loads(Path(args.input).read_text(encoding="utf-8"))
    normalized = normalize(raw)
    Path(args.output).write_text(
        json.dumps(normalized, ensure_ascii=False, indent=2), encoding="utf-8"
    )
    print(f"wrote {len(normalized)} notes to {args.output}")
    return 0


if __name__ == "__main__":
    sys.exit(_main())

• [ ] Step 6: Run to verify it passes

Run: cd interview-intelligence && .venv/bin/pytest tests/test_normalize_xhs.py -v Expected: 7 passed.

• [ ] Step 7: Commit

cd interview-intelligence && git add scripts/scrape/__init__.py scripts/scrape/normalize_xhs.py tests/test_normalize_xhs.py tests/fixtures/mc_xhs_raw.json && git commit -m "feat: add MediaCrawler->小红书 connector normalize adapter"

---

Task 2: CLI smoke test + end-to-end with Plan 3 connector

Files:

• Test: interview-intelligence/tests/test_normalize_xhs_cli.py

Two integration-flavored tests:

1. CLI: invoke scripts.scrape.normalize_xhs._main([input, "-o", output]) in-process, assert file is written, JSON shape matches normalized output, and the printed line is correct. 2. End-to-end: feed the normalize output JSON text into Plan 3's parse_mediacrawler_export and assert the resulting RawPost for n1 has source="xiaohongshu", post_type="image", posted_at="2025-09-20", asset_paths with both URLs.

• [ ] Step 1: Write the failing test interview-intelligence/tests/test_normalize_xhs_cli.py

import json
from pathlib import Path

from scripts.scrape.normalize_xhs import _main
from scripts.connectors.xiaohongshu import parse_mediacrawler_export

FIXTURE = Path(__file__).parent / "fixtures" / "mc_xhs_raw.json"


def test_cli_writes_normalized_file(tmp_path, capsys):
    out_path = tmp_path / "xhs_export.json"
    rc = _main([str(FIXTURE), "-o", str(out_path)])
    assert rc == 0
    assert out_path.exists()
    written = json.loads(out_path.read_text(encoding="utf-8"))
    assert isinstance(written, list)
    assert len(written) == 4
    assert written[0]["note_url"] == "https://www.xiaohongshu.com/explore/n1"
    captured = capsys.readouterr()
    assert "wrote 4 notes" in captured.out


def test_end_to_end_with_plan3_connector(tmp_path):
    out_path = tmp_path / "xhs_export.json"
    _main([str(FIXTURE), "-o", str(out_path)])
    posts = parse_mediacrawler_export(out_path.read_text(encoding="utf-8"))
    assert len(posts) == 4
    n1 = posts[0]
    assert n1.source == "xiaohongshu"
    assert n1.post_type == "image"
    assert n1.posted_at == "2025-09-20"
    assert n1.asset_paths == [
        "https://sns-img.xhs.cn/n1_a.jpg",
        "https://sns-img.xhs.cn/n1_b.jpg",
    ]
    # n2 has time=0 in the source, so posted_at should be None
    n2 = next(p for p in posts if "RAG" in p.raw_text)
    assert n2.posted_at is None

• [ ] Step 2: Run the test

Run: cd interview-intelligence && .venv/bin/pytest tests/test_normalize_xhs_cli.py -v Expected: 2 passed.

Note: this is an integration test for the contract between Task 1's adapter and Plan 3's connector. It locks in the end-to-end shape; no separate failing-then-passing TDD cycle needed because both endpoints already exist.

• [ ] Step 3: Commit

cd interview-intelligence && git add tests/test_normalize_xhs_cli.py && git commit -m "test: CLI + end-to-end adapter→XiaohongshuConnector contract"

---

Task 3: Setup doc

Files:

• Create: docs/setup_mediacrawler.md (at repo root, not inside interview-intelligence/)

• [ ] Step 1: Write docs/setup_mediacrawler.md verbatim:

````markdown

小红书面经采集 — MediaCrawler 设置指引

启用 interview-intelligence skill 的小红书源是可选步骤。skill 本身不抓数据; 真正的采集由开源工具 MediaCrawler 完成, 本文档教你怎么把它产出的 JSON 喂给本 skill。

仅供个人、非商业用途。请遵守目标平台的服务条款。

1. 前提

• Python 3.11+、Git、一个可登录小红书的微信/手机号
• 一个普通浏览器(扫码登录用)

2. 装 MediaCrawler(放在本仓库外面)

cd ~/Code   # 或任意你存放第三方仓库的位置
git clone https://github.com/NanmiCoder/MediaCrawler.git
cd MediaCrawler
# 按它 README 的步骤装依赖(它通常用 uv 或 pip + requirements.txt)

不要把 MediaCrawler clone 到本仓库里。它有自己的依赖、许可证、更新节奏,

解耦更好维护。

3. 登录小红书

按 MediaCrawler README 的「登录方式」一节操作(QR 扫码 / Cookie 注入二选一)。 登录态保存在它自己的目录里。

4. 用关键词搜面经

在 MediaCrawler 仓库里运行它的搜索命令,目标平台选 xhs,关键词建议:

• 你的目标岗位别名(例如「AI 应用开发 面经」「Agent 工程师 面试」「大模型应用 实习」)
• 公司 + 岗位组合(例如「字节 AI 实习 面经」)

具体命令格式以 MediaCrawler 当前版本 README 为准。跑完后输出文件通常在 MediaCrawler/data/xhs/json/ 之类的位置,文件名形如 search_contents_2026-xx-xx.json。

5. 归一化

回到本仓库:

cd /path/to/InterviewPrepare/interview-intelligence
.venv/bin/python -m scripts.scrape.normalize_xhs \
    /path/to/MediaCrawler/data/xhs/json/search_contents_*.json \
    -o corpus_cache/xhs_export.json

成功会打印 wrote N notes to corpus_cache/xhs_export.json。

如果适配器报错,多半是 MediaCrawler 升级了输出 schema。 检查 scripts/scrape/normalize_xhs.py 顶部的字段假设注释,对照真实 JSON 修字段名, 跑测试 pytest tests/test_normalize_xhs.py 验证,再用真实文件重跑。

6. 喂给 skill

在 skill 的工作流里,给 XiaohongshuConnector 传刚才的输出路径:

XiaohongshuConnector(export_path="corpus_cache/xhs_export.json")

剩下的(时效过滤、OCR、去重、项目锚定)skill 会自己处理。

7. 复跑

数据陈旧时直接重跑步骤 4–5。Plan 2 的时效过滤会把超过两年的笔记从结果里剔掉, 所以你不用手动清理旧数据。 ````

• [ ] Step 2: Commit

git add docs/setup_mediacrawler.md && git commit -m "docs: add MediaCrawler setup guide for 小红书 source"

(Note: this commit runs from the repo root, NOT from interview-intelligence/.)

---

Task 4: SKILL.md tie-in

Files:

• Modify: interview-intelligence/SKILL.md

Add a single "0. 准备" step before step 1, and tweak the constraints to point at the setup doc.

• [ ] Step 1: Insert step 0 — find the line ## 工作流 and insert a new step block immediately before the existing 1. **简历理解。** line:

0. **准备(仅当启用小红书源)。** 让用户先按 `docs/setup_mediacrawler.md` 跑一遍 MediaCrawler 采集 + 适配器归一化,产出 `corpus_cache/xhs_export.json`(适配器在 `scripts/scrape/normalize_xhs.py`,把 MediaCrawler 的原生输出转成 `XiaohongshuConnector` 能吃的格式)。文本/牛客/GitHub 源不需要这一步。

• [ ] Step 2: Add adapter to the tools list — under ## 工具, add this line right after the existing scripts/connectors/xiaohongshu.py line:

- `scripts/scrape/normalize_xhs.py` → `normalize(notes) -> list[dict]`(CLI:`python -m scripts.scrape.normalize_xhs <in.json> -o <out.json>`),把 MediaCrawler 原生输出归一化为 `XiaohongshuConnector` 的输入。

• [ ] Step 3: Update constraint — find the existing line

- 小红书走 MediaCrawler 采集导出,OCR 采用混合策略(粗 OCR + 视觉回退);MediaCrawler 仅供个人、非商业用途。

and replace it with:

- 小红书走 MediaCrawler 采集导出(用户预先离线跑一次,流程见 `docs/setup_mediacrawler.md`),OCR 采用混合策略(粗 OCR + 视觉回退);MediaCrawler 仅供个人、非商业用途。

• [ ] Step 4: Validate front-matter

Run: cd interview-intelligence && .venv/bin/python -c "t=open('SKILL.md').read(); assert t.startswith('---'); fm=t.split('---')[1]; assert 'name:' in fm and 'description:' in fm; print('OK')" Expected: prints OK.

• [ ] Step 5: Commit

cd interview-intelligence && git add SKILL.md && git commit -m "docs: wire MediaCrawler adapter pre-step into SKILL.md"

---

Task 5: Full suite green

• [ ] Step 1: Run the whole suite

Run: cd interview-intelligence && .venv/bin/pytest -q Expected: 39 (Plan 3) + 7 (Task 1) + 2 (Task 2) = 48 passed.

• [ ] Step 2: No commit if no changes. If the suite is green and nothing is staged, skip. Otherwise investigate the regression.

---

Self-Review

Spec coverage:

• Adapter at scripts/scrape/normalize_xhs.py with the field rules from spec §4 → Task 1. ✓
• CLI python -m scripts.scrape.normalize_xhs <in> -o <out> → Task 1 (_main) + Task 2 (CLI smoke). ✓
• End-to-end contract adapter → Plan 3 connector → Task 2. ✓
• Setup doc docs/setup_mediacrawler.md covering install/login/run/normalize/feed → Task 3. ✓
• SKILL.md step 0 + tools list + constraint update → Task 4. ✓
• Fixture covers all branches from spec §7 (normal note, comma-string image list, missing URL synth, missing id+url skipped, time=0, invalid time, null image list, extra keys dropped, order preserved) → Task 1 fixture + tests. ✓
• Deferred correctly: no vendoring, no auto-login, no image download, no validate-export command.

Placeholder scan: No TBD/TODO. Every code step has full code. All shell commands have expected output.

Type consistency:

• normalize(notes: list[dict]) -> list[dict] signature identical across Task 1 implementation, Task 1 tests, Task 2 tests, SKILL.md tools list.
• Output keys (note_id, note_url, title, desc, time, image_list) exactly match what Plan 3's parse_mediacrawler_export reads — verified by the end-to-end test (Task 2 step 3).
• _main(argv) -> int callable from tests (Task 2) and CLI (Task 1 if __name__).
• XiaohongshuConnector(export_path=...) referenced unchanged in Task 3 doc and Task 4 SKILL.md (matches Plan 3).

GitHub Hints + Undated Demote 实施计划

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 给 GithubConnector 加 relevance_hints 过滤减噪音,把 posted_at=None 在排序里降权到 0.2(低于已知 >730 天),并在 SKILL.md 里记录 hints 用法。

Architecture: 两个独立的小改:(A) extract_posts_from_markdown 和 GithubConnector 多一个 relevance_hints 可选参数,默认 None 表示不过滤,保证向后兼容;(B) _recency_weight 把 None/解析失败两条分支的返回值从 0.5 改成 0.2。两改通过独立任务、独立 commit 落地。

Tech Stack: Python 3.11、pytest、.venv/bin/python(包内 venv)

仓库根:/Users/kun/Desktop/Projects/InterviewPrepare 子包根:interview-intelligence/ pytest 工作目录:interview-intelligence/(pytest tests/...) Spec:docs/superpowers/specs/2026-06-01-github-relevance-and-undated-demote-design.md

---

File Structure

---

Task 1: GithubConnector relevance_hints 过滤

Files:

• Modify: interview-intelligence/scripts/connectors/github.py
• Modify: interview-intelligence/tests/test_github_connector.py

---

• [ ] Step 1: 写失败测试 — hints 过滤

把下面的测试追加到 interview-intelligence/tests/test_github_connector.py 末尾:

def test_extract_with_hints_keeps_only_matching():
    posts = extract_posts_from_markdown(
        SAMPLE_MD,
        "https://example.com/repo",
        relevance_hints=["RAG"],
    )
    texts = [p.raw_text for p in posts]
    assert "什么是 RAG？" in texts
    assert "说明 MCP 和 Skill 的区别" not in texts
    assert "介绍一下你的 agent 项目架构" not in texts


def test_extract_with_hints_case_insensitive():
    posts = extract_posts_from_markdown(
        SAMPLE_MD,
        "https://example.com/repo",
        relevance_hints=["mcp"],  # lowercase, candidate has uppercase MCP
    )
    texts = [p.raw_text for p in posts]
    assert any("MCP" in t for t in texts)
    assert all("RAG" not in t for t in texts)


def test_extract_with_empty_hints_does_not_filter():
    posts_none = extract_posts_from_markdown(SAMPLE_MD, "https://example.com/repo", relevance_hints=None)
    posts_empty = extract_posts_from_markdown(SAMPLE_MD, "https://example.com/repo", relevance_hints=[])
    assert {p.raw_text for p in posts_none} == {p.raw_text for p in posts_empty}
    assert len(posts_empty) >= 3  # all question-like lines


def test_connector_passes_hints_through():
    conn = GithubConnector(
        repo_raw_urls=["https://example.com/repo"],
        fetcher=lambda url: SAMPLE_MD,
        relevance_hints=["agent"],
    )
    result = conn.search([])
    assert result.status == "ok"
    texts = [p.raw_text for p in result.posts]
    assert any("agent" in t.lower() for t in texts)
    assert all("RAG" not in t for t in texts)

• [ ] Step 2: 跑测试验证失败

cd /Users/kun/Desktop/Projects/InterviewPrepare/interview-intelligence
.venv/bin/python -m pytest tests/test_github_connector.py -v

预期:4 个新测试 FAIL,失败原因 extract_posts_from_markdown() got an unexpected keyword argument 'relevance_hints' 和 GithubConnector.__init__() got an unexpected keyword argument 'relevance_hints'。已有 4 个测试仍 PASS。

• [ ] Step 3: 改 github.py — 加 hints 过滤

把 interview-intelligence/scripts/connectors/github.py 完整替换为:

import re
from collections.abc import Callable, Iterable

import requests

from scripts.connectors.base import Connector, SearchResult
from scripts.models import RawPost

_KEYWORDS = ("介绍", "说明", "区别", "原理", "什么是", "如何", "为什么", "解释")
_HEADING = re.compile(r"^#{1,6}\s+(.*)$")
_BULLET = re.compile(r"^(?:[-*]|\d+\.)\s+(.*)$")


def _is_question_like(text: str) -> bool:
    t = text.strip()
    if not t:
        return False
    if t.endswith("?") or t.endswith("?"):
        return True
    return any(k in t for k in _KEYWORDS)


def _matches_any_hint(text: str, hints: Iterable[str]) -> bool:
    t = text.lower()
    return any(h.lower() in t for h in hints)


def extract_posts_from_markdown(
    md_text: str,
    url: str,
    relevance_hints: list[str] | None = None,
) -> list[RawPost]:
    posts: list[RawPost] = []
    use_hints = bool(relevance_hints)
    for line in md_text.splitlines():
        m = _HEADING.match(line) or _BULLET.match(line)
        candidate = m.group(1).strip() if m else line.strip()
        if not _is_question_like(candidate):
            continue
        if use_hints and not _matches_any_hint(candidate, relevance_hints):
            continue
        posts.append(RawPost(source="github", url=url, post_type="text", raw_text=candidate))
    return posts


def _default_fetcher(url: str) -> str:
    resp = requests.get(url, timeout=30)
    resp.raise_for_status()
    return resp.text


class GithubConnector(Connector):
    name = "github"

    def __init__(
        self,
        repo_raw_urls: list[str],
        fetcher: Callable[[str], str] | None = None,
        relevance_hints: list[str] | None = None,
    ):
        self.repo_raw_urls = repo_raw_urls
        self.fetcher = fetcher or _default_fetcher
        self.relevance_hints = relevance_hints

    def search(self, queries: list[str]) -> SearchResult:
        posts: list[RawPost] = []
        try:
            for url in self.repo_raw_urls:
                posts.extend(
                    extract_posts_from_markdown(
                        self.fetcher(url),
                        url,
                        relevance_hints=self.relevance_hints,
                    )
                )
        except Exception as exc:  # noqa: BLE001 - degrade, never crash the pipeline
            return SearchResult.degraded(self.name, f"fetch failed: {exc}")
        return SearchResult(posts=posts, status="ok", message=f"{len(posts)} posts")

注意:_is_question_like 里的两个问号原文件用的就是英文 ? + 中文 ?,这里保持原样。

• [ ] Step 4: 跑测试验证通过

cd /Users/kun/Desktop/Projects/InterviewPrepare/interview-intelligence
.venv/bin/python -m pytest tests/test_github_connector.py -v

预期:8 个测试全部 PASS。

• [ ] Step 5: 提交

cd /Users/kun/Desktop/Projects/InterviewPrepare
git add interview-intelligence/scripts/connectors/github.py interview-intelligence/tests/test_github_connector.py
git commit -m "feat(github): add relevance_hints to filter noise

When relevance_hints is provided, only keep question-like lines that
contain at least one hint (case-insensitive substring). Default None
preserves prior behavior."

---

Task 2: posted_at=None 降权到 0.2

Files:

• Modify: interview-intelligence/scripts/corpus/dedupe_rank.py
• Modify: interview-intelligence/tests/test_dedupe_rank.py

---

• [ ] Step 1: 改测试 — 更新 None 顺序 + 新增解析失败测试

打开 interview-intelligence/tests/test_dedupe_rank.py。

替换 test_none_date_weight_between_fresh_and_stale 整段(原断言 fresh > undated > stale)为:

def test_undated_ranks_below_known_stale():
    # New policy: undated (0.2) ranks BELOW known-stale (0.3).
    # freq all 1: fresh(1.0) > stale(0.3) > undated(0.2)
    qs = [
        Question("stale", ["a"], latest_posted_at="2022-01-01"),
        Question("undated", ["b"], latest_posted_at=None),
        Question("fresh", ["c"], latest_posted_at="2026-05-01"),
    ]
    out = dedupe_and_rank(qs, today=date(2026, 5, 28))
    assert [q.text for q in out] == ["fresh", "stale", "undated"]

追加到文件末尾:

def test_malformed_date_treated_as_undated():
    # Malformed posted_at should weight the same as None (0.2), i.e. rank below known-stale.
    qs = [
        Question("stale", ["a"], latest_posted_at="2022-01-01"),
        Question("garbled", ["b"], latest_posted_at="not-a-date"),
    ]
    out = dedupe_and_rank(qs, today=date(2026, 5, 28))
    assert [q.text for q in out] == ["stale", "garbled"]

• [ ] Step 2: 跑测试验证失败

cd /Users/kun/Desktop/Projects/InterviewPrepare/interview-intelligence
.venv/bin/python -m pytest tests/test_dedupe_rank.py -v

预期:test_undated_ranks_below_known_stale FAIL(当前权重 0.5 让 undated 排在 stale 前面),test_malformed_date_treated_as_undated FAIL(原因同上)。其余 PASS。

• [ ] Step 3: 改 dedupe_rank.py — None / ValueError 分支返回 0.2

修改 interview-intelligence/scripts/corpus/dedupe_rank.py 的 _recency_weight 函数:

把这段:

def _recency_weight(posted_at: str | None, today: date) -> float:
    if not posted_at:
        return 0.5
    try:
        d = datetime.strptime(posted_at, "%Y-%m-%d").date()
    except ValueError:
        return 0.5
    days = (today - d).days
    if days <= 365:
        return 1.0
    if days <= 730:
        return 0.6
    return 0.3

换成:

def _recency_weight(posted_at: str | None, today: date) -> float:
    if not posted_at:
        return 0.2
    try:
        d = datetime.strptime(posted_at, "%Y-%m-%d").date()
    except ValueError:
        return 0.2
    days = (today - d).days
    if days <= 365:
        return 1.0
    if days <= 730:
        return 0.6
    return 0.3

• [ ] Step 4: 跑测试验证通过

cd /Users/kun/Desktop/Projects/InterviewPrepare/interview-intelligence
.venv/bin/python -m pytest tests/test_dedupe_rank.py -v

预期:全部 PASS(原 5 个 + 改名后的 1 个 + 新增 1 个 = 7 个)。

• [ ] Step 5: 提交

cd /Users/kun/Desktop/Projects/InterviewPrepare
git add interview-intelligence/scripts/corpus/dedupe_rank.py interview-intelligence/tests/test_dedupe_rank.py
git commit -m "feat(dedupe_rank): demote undated/malformed posts to 0.2

Timeliness is a hard requirement; undated posts are empirically often
older than known-stale. New ordering: fresh(<=365d, 1.0) > 730d(0.6)
> stale(>730d, 0.3) > undated/malformed(0.2)."

---

Task 3: SKILL.md 文档同步

Files:

• Modify: interview-intelligence/SKILL.md(步骤 3b 段)

---

• [ ] Step 1: 编辑 SKILL.md

打开 interview-intelligence/SKILL.md,找到这一段(在第 44 行附近,步骤 3b 的开头):

   **3b. 调 connectors + 收割。** 把分好桶的 URL 喂给对应 connector,结果用 `save_raw_posts` 落盘。读取结果,**收割真实出现的岗位名 / 标签 / 高频术语**,用收割到的词跑下一轮 3a,直到不再冒出新词。若某 connector 返回 `status="degraded"`(例如牛客需要 cookie、小红书需要先跑 MediaCrawler、或消息含 `selector` 表示 HTML 漂移),把它需要的东西告诉用户;主力源降级会显著影响时效性,必须明确提示用户,不要默默用 GitHub 凑数。

在这一段之后追加一段(空一行,然后):

   **GitHub 调用必带 `relevance_hints`。** GitHub 仓库里夹带大量算法/八股,如果不带提示词过滤会污染语料。`GithubConnector(repo_raw_urls=..., relevance_hints=<当前一轮的术语/岗位别名>)`,第一轮没收割到东西时用步骤 2 的种子查询当 hints。命中规则:子串、大小写不敏感,只要正文里出现任一 hint 就保留。

• [ ] Step 2: 验证 SKILL.md 仍可解析

cd /Users/kun/Desktop/Projects/InterviewPrepare
.venv/bin/python - <<'EOF'
from pathlib import Path
text = Path("interview-intelligence/SKILL.md").read_text(encoding="utf-8")
# Minimal sanity check: front-matter, key steps present
assert text.startswith("---\nname: interview-intelligence")
assert "**3a. URL 发现" in text
assert "**3b. 调 connectors" in text
assert "relevance_hints" in text
print("ok")
EOF

预期输出 ok。如果 .venv 路径不在仓库根,改成 interview-intelligence/.venv/bin/python。

• [ ] Step 3: 跑全套测试回归

cd /Users/kun/Desktop/Projects/InterviewPrepare/interview-intelligence
.venv/bin/python -m pytest tests/ -v

预期:所有测试 PASS,无新失败。

• [ ] Step 4: 提交

cd /Users/kun/Desktop/Projects/InterviewPrepare
git add interview-intelligence/SKILL.md
git commit -m "docs: document relevance_hints usage for GithubConnector in step 3b"

---

完成后

回到 main 提交合并(由控制方在 finishing-a-development-branch 阶段执行):

git checkout main
git merge --no-ff fix/github-hints-and-undated

测试再跑一次确认无回归后,删分支。

MediaCrawler 适配器 + 设置文档 — Design Spec

Date: 2026-05-31 Status: Approved (brainstorm) Predecessors: 2026-05-28-interview-intelligence-skill-design.md; Plans 1/2/3 merged.

1. Goal

让用户能用开源 MediaCrawler 真实抓取小红书面经,产出能被 Plan 3 的 XiaohongshuConnector 直接消费的 JSON。

2. Scope

In scope:

• 一个文件级适配器 scripts/scrape/normalize_xhs.py,把 MediaCrawler 原生 小红书 输出归一化为 connector 期望的 schema。CLI 可独立运行。
• 一份 docs/setup_mediacrawler.md 设置文档:用户怎么 clone MediaCrawler、登录、跑搜索、把输出交给适配器。
• SKILL.md 增补一条「可选 0. 准备」步骤,指向设置文档,并说明适配器输出即 connector 输入。
• Fixture-based 单元测试。无网络、无 subprocess。

Out of scope (deliberately):

• Vendor MediaCrawler(不作为 submodule / 不写自动 clone)。
• 自动登录、cookie 管理、调度。
• 图片下载。
• 牛客 / GitHub / OCR 模块改动。
• 真实端到端拉取(这就是 offline 路线的初衷)。

3. Architecture

┌─────────────────┐    ┌──────────────────────────┐    ┌─────────────────────┐
│ MediaCrawler    │    │ scripts/scrape/          │    │ scripts/connectors/ │
│ (用户自己装并跑)  │ →  │ normalize_xhs.py         │ →  │ xiaohongshu.py      │
│ xhs_notes_*.json│    │ (适配器,纯文件IO)         │    │ (Plan 3, 不动)       │
└─────────────────┘    └──────────────────────────┘    └─────────────────────┘

文件级解耦:适配器和 connector 之间只通过磁盘上的 JSON 文件交流。MediaCrawler 改了输出格式只动适配器,connector 一行不用改。

4. Adapter contract

输入: MediaCrawler 小红书 notes JSON 文件路径(假设字段——会在实现阶段对照真实样本调整):

[
  {
    "note_id": "...",
    "type": "normal" | "video",
    "title": "...",
    "desc": "...",
    "time": 1758326400000,           // epoch ms
    "image_list": "url1,url2"        // MediaCrawler often joins as comma string
       | ["url1", "url2"],           // or list — adapter must accept both
    "tag_list": "面经,实习",          // optional
    "note_url": "...",               // sometimes missing; adapter synthesizes from note_id
    "liked_count": "1.2万",          // out of scope, dropped
    "comments": [ ... ]              // out of scope, dropped
  }
]

输出: Plan 3 parse_mediacrawler_export 已经接受的 schema:

[
  {
    "note_id": "...",
    "note_url": "https://www.xiaohongshu.com/explore/<note_id>",
    "title": "...",
    "desc": "...",
    "time": 1758326400000,
    "image_list": ["url1", "url2"]
  }
]

映射规则:

• note_url 缺失 → 用 note_id 合成 https://www.xiaohongshu.com/explore/<note_id>;两者都缺 → 跳过该条。
• image_list 是字符串 → split , 并去空;是列表 → 原样;缺失 → []。
• time 解析失败或 0 → 输出仍写 0(下游已经把 0 视作 posted_at=None)。
• title / desc 缺失 → 空串。
• 其它 MediaCrawler 字段一律丢弃。
• 顺序保持。

CLI:

python -m scripts.scrape.normalize_xhs <input.json> -o <output.json>

写完打印 wrote N notes to <output.json>(N 是输出条数,可能少于输入,因为跳过了无 id 的)。

5. Setup doc structure (docs/setup_mediacrawler.md)

简短、命令为主、不复述 MediaCrawler 文档:

1. 前提(Python 3.11+、git、浏览器) 2. git clone https://github.com/NanmiCoder/MediaCrawler.git 到 skill 仓库外面(强调不放进来) 3. 跟 MediaCrawler README 装依赖 4. 登录步骤(指 MediaCrawler 的 QR / cookie 文档,不复述) 5. 跑搜索命令模板(以小红书 + 关键词为例) 6. 找输出文件位置 7. 用适配器归一化:python -m scripts.scrape.normalize_xhs <mc输出> -o corpus_cache/xhs_export.json 8. 把 corpus_cache/xhs_export.json 喂给 skill / XiaohongshuConnector(export_path=...) 9. 提示:仅供个人非商业用途;关键词建议(岗位别名)

6. SKILL.md change

新增「0. 准备(仅当启用小红书)」段落,放在工作流第 1 步之前,指向 docs/setup_mediacrawler.md,一句话:「适配器输出 = XiaohongshuConnector(export_path=...) 的输入」。其他步骤保持不变。

7. Testing strategy

完全 fixture-based,不连网:

• Fixture A:一份手写的伪 MediaCrawler 输出 JSON,覆盖以下情况:
• 正常 note(有 note_url、image_list 列表)
• image_list 是逗号字符串
• 缺 note_url、但有 note_id(应合成 URL)
• 缺 note_id 和 note_url(应跳过)
• time=0、缺 title、缺 desc(应保留并空串/0)
• 含 MediaCrawler 多余字段(liked_count、comments)(应丢弃)
• 单元测试:
• normalize(mc_notes) -> list[dict] 直接断言映射规则
• CLI 走文件 IO,写出后读回,字段集严格等于预期
• 端到端串通测试:适配器输出文件喂给 Plan 3 的 parse_mediacrawler_export,断言 RawPost 字段(尤其 posted_at、asset_paths)正确。

8. Risks / open questions

• 真实 MediaCrawler 字段名是基于公开仓库的推测。 实现 PR 里实施者必须在 commit 信息或代码注释里列出适配器对每个字段的具体假设;如果用户跑过后发现某字段不对,后续小 PR 改 fixture + 映射即可。这是适配器解耦的全部价值——风险被关在一个文件里。
• MediaCrawler 自己的输出 schema 在跨版本演化时可能再变;我们不做 schema 版本协商,出错就让 normalize 报错,文档里写「如果适配器报错,可能是 MediaCrawler 升级了 schema,提 issue」。

9. Deferred for later

• 自动 install / 自动登录 / 抓视频(抖音、B 站走 ASR)
• 图片下载
• validate_export 命令(用户提议过,先不做;normalize 已经会在数据怪时报错)
• 牛客也走类似适配器路线(目前牛客直接从 HTML 解析,够用)