用WorkBuddy做AI工具站06:写第一个采集器

大家好，我是宇哥，专注AI编程、智能体，解决小白AI编程问题。

前 5 篇我们已经把 AI 工具站做到了一个很关键的位置：有项目文件、有工具字段、有第一版页面、有搜索筛选，也有了数据源清单。

这一篇开始进入第二阶段：让工具站的数据可以增长。

但我先强调一句：我们不是一上来就写一个到处乱爬的脚本，更不是把抓到的内容直接覆盖线上数据。

今天只做一个小而稳的版本：

读取上一篇的数据源清单 → 访问少量公开页面 → 提取 title / meta description → 写入待审核 JSON → 人工确认后再决定是否更新正式数据。

这篇你能做出什么

跟着做完，你会得到一个真实可运行的小采集器：

tutorials/ai-tool-directory/demo/lesson-06/├── data/│   ├── tool-source-map.json│   ├── source-policy.json│   └── collected-tools-review.json├── scripts/│   ├── collect-first-batch.py│   └── check-collected-review.py└── prompts/    └── workbuddy-first-collector-prompt.md

最后你能运行：

python3 scripts/collect-first-batch.py && python3 scripts/check-collected-review.py

看到类似这次真实跑出来的结果：

✅ 第一版采集器运行完成输入工具：5 个成功访问：5 个待复核/失败：0 个输出文件：data/collected-tools-review.json发布闸门：全部记录保持 needs_manual_review，没有覆盖正式 tools.json✅ 采集结果检查通过待人工复核记录：5 条成功访问公开页面：5 个审核状态：全部 needs_manual_review正式数据保护：未创建/覆盖 data/tools.json

这就说明：你的第一版采集器已经跑通，而且没有误伤正式工具数据。

适合谁，不适合谁

适合你，如果你现在想做的是：

• 给 AI 工具目录站补充更多工具信息；
• 先做一个可控的小采集流程；
• 让 AI Agent 帮你写脚本，但你还想保留人工审核；
• 不想一上来就碰复杂数据库、队列、后台系统。

不适合你，如果你想要的是：

• 大规模自动采集；
• 绕过登录、限制或付费墙；
• 自动发布抓取结果；
• 不做人工审核直接改线上数据。

这篇的原则很清楚：宁可慢一点，也不要把脏数据自动写进站点。

准备环境

你需要准备：

1. WorkBuddy：用来让 AI Agent 帮你生成脚本、解释代码和修报错。
2. 本地 Python 3：本篇脚本只用 Python 标准库，不需要安装第三方包。
3. 前一篇产物：tool-source-map.json 和 source-policy.json。
4. 一个本地项目目录：建议沿用系列目录。

本篇 demo 的路径是：

/root/workspace/wechat-articles/tutorials/ai-tool-directory/demo/lesson-06/

你自己的电脑可以用类似路径：

ai-tool-directory/demo/lesson-06/

先看最终效果

这一篇不是做一个页面，而是做一个“数据进入正式工具库之前的缓冲区”。

流程长这样：

采集器会读取上一篇准备好的数据源清单：

{"slug":"playwright","name":"Playwright","source":{"source_type":"docs","source_url":"https://playwright.dev","fields_to_check":["website","summary","tags","open_source_status"]},"collection_status":"ready_for_manual_review"}

然后输出一份只用于审核的文件：

data/collected-tools-review.json

注意，它不应该创建或覆盖：

data/tools.json

因为正式数据要等第 7 篇做审核后台之后，再进入发布流程。

第一步｜新建 lesson-06 目录

打开你的项目目录，新建本篇目录：

mkdir -p demo/lesson-06/data demo/lesson-06/scripts demo/lesson-06/promptscd demo/lesson-06

如果你跟着我的路径做，就是：

mkdir -p /root/workspace/wechat-articles/tutorials/ai-tool-directory/demo/lesson-06/data \  /root/workspace/wechat-articles/tutorials/ai-tool-directory/demo/lesson-06/scripts \  /root/workspace/wechat-articles/tutorials/ai-tool-directory/demo/lesson-06/promptscd /root/workspace/wechat-articles/tutorials/ai-tool-directory/demo/lesson-06

验收点：看到下面这三个目录就算成功。

data/scripts/prompts/

第二步｜把上一篇的数据源文件复制进来

从第 5 篇复制两个文件：

cp ../lesson-05/data/tool-source-map.json data/tool-source-map.jsoncp ../lesson-05/data/source-policy.json data/source-policy.json

如果你的目录不一样，就把路径换成自己的。

复制后检查：

python3 - <<'PY'from pathlib import Pathfor path in ['data/tool-source-map.json', 'data/source-policy.json']:    p = Path(path)print(path, '存在'if p.exists() else'缺失', p.stat().st_size if p.exists() else 0)PY

你应该看到两个文件都存在，而且大小不是 0。

为什么要复制，而不是直接改上一篇？

因为教程项目要留痕。每一篇都有自己的 demo，后面出问题时，你能回到任何一篇复盘。

第三步｜把这段提示词复制到 WorkBuddy

新建文件：

prompts/workbuddy-first-collector-prompt.md

复制下面这段：

你是我的 AI 工具目录站开发助手。请只在当前项目目录里工作，目标是写一个“第一版安全采集器”，不要直接修改正式 tools.json。已存在输入文件：- data/tool-source-map.json：每个工具对应的数据源、字段、风险等级- data/source-policy.json：采集边界和发布闸门请生成：1. scripts/collect-first-batch.py   - 使用 Python 标准库即可，不引入第三方依赖。   - 读取 data/tool-source-map.json 和 data/source-policy.json。   - 每次只采集少量公开页面，不登录、不绕过限制、不采集个人数据。   - 先检查 robots.txt 是否能访问，记录 robots 状态。   - 访问 source_url，提取 HTTP 状态码、页面 title、meta description 或 GitHub 仓库 description。   - 输出 data/collected-tools-review.json。   - 每条记录必须带 review_status: needs_manual_review。2. scripts/check-collected-review.py   - 检查输出文件存在。   - 检查每条记录包含 slug/name/source_url/fetched_at/http_status/review_status/suggested_update/evidence。   - 检查 review_status 不能是 published。   - 检查采集结果没有覆盖 data/tools.json。请写清楚命令：python3 scripts/collect-first-batch.pypython3 scripts/check-collected-review.py如果某个页面访问失败，只记录 error 字段，不要让整个脚本崩溃；最终输出成功采集/失败数量。

在 WorkBuddy 里让它执行这个任务。

这里的关键不是“让 AI 一口气帮你做完全部采集系统”，而是让它先生成一个能被你检查的小脚本。

第四步｜创建采集脚本

WorkBuddy 生成后，你应该得到：

scripts/collect-first-batch.py

如果你想直接跟着本篇做，也可以新建这个文件，把下面代码复制进去：

#!/usr/bin/env python3from __future__ import annotationsimport jsonimport reimport sslfrom datetime import datetime, timezonefrom html import unescapefrom pathlib import Pathfrom urllib.parse import urlparsefrom urllib.request import Request, urlopenfrom urllib.error import URLError, HTTPErrorROOT = Path(__file__).resolve().parents[1]DATA = ROOT / "data"SOURCE_MAP = DATA / "tool-source-map.json"SOURCE_POLICY = DATA / "source-policy.json"OUTPUT = DATA / "collected-tools-review.json"TIMEOUT = 15USER_AGENT = "YugeAIToolDirectoryTutorial/0.1 (+manual-review-demo)"defload_json(path: Path):return json.loads(path.read_text(encoding="utf-8"))deffetch_url(url: str, *, timeout: int = TIMEOUT):    req = Request(url, headers={"User-Agent": USER_AGENT})try:with urlopen(req, timeout=timeout, context=ssl.create_default_context()) as resp:            status = getattr(resp, "status", None)            content_type = resp.headers.get("content-type", "")            raw = resp.read(350_000)            charset = resp.headers.get_content_charset() or"utf-8"            text = raw.decode(charset, errors="replace")return status, text, content_typeexcept HTTPError as exc:        body = exc.read(80_000).decode("utf-8", errors="replace") if exc.fp else""return exc.code, body, exc.headers.get("content-type", "") if exc.headers elseNoneexcept (URLError, TimeoutError, ssl.SSLError) as exc:returnNone, "", f"ERROR: {exc}"defrobots_status(source_url: str):    parsed = urlparse(source_url)    robots_url = f"{parsed.scheme}://{parsed.netloc}/robots.txt"    status, text, content_type = fetch_url(robots_url, timeout=8)    result = {"url": robots_url, "http_status": status, "content_type": content_type}if text:        result["sample"] = " ".join(text.split())[:160]    result["status"] = "available"if status and200 <= status < 400else"unavailable_or_restricted"return resultdefclean(value: str | None, limit: int = 220) -> str:ifnot value:return""    value = unescape(re.sub(r"\s+", " ", value)).strip()return value[:limit]defextract_title(html: str) -> str:match = re.search(r"<title[^>]*>(.*?)</title>", html, re.I | re.S)return clean(match.group(1) ifmatchelse"", 140)defextract_meta_description(html: str) -> str:    patterns = [r'<meta[^>]+name=["\']description["\'][^>]+content=["\'](.*?)["\']',r'<meta[^>]+content=["\'](.*?)["\'][^>]+name=["\']description["\']',r'<meta[^>]+property=["\']og:description["\'][^>]+content=["\'](.*?)["\']',r'<meta[^>]+content=["\'](.*?)["\'][^>]+property=["\']og:description["\']',    ]for pattern in patterns:match = re.search(pattern, html, re.I | re.S)ifmatch:return clean(match.group(1), 260)return""defsuggested_update_from_page(item: dict, page_text: str, status: int | None, content_type: str | None) -> dict:    title = extract_title(page_text)    description = extract_meta_description(page_text)    suggestion = {"website": item["source"]["source_url"],"summary_candidate": description or title,"title_candidate": title,"pricing_candidate": item.get("current_pricing", "manual_review"),"tags_candidate": item["source"].get("fields_to_check", []),    }if item["source"].get("source_type") == "github":        suggestion["open_source_status_candidate"] = "需要人工确认仓库许可证、活跃度和社区版边界"if status isNoneor (status and status >= 400):        suggestion["review_warning"] = "页面未正常返回，保留旧数据并进入人工复核"if content_type and"text/html"notin content_type and"json"notin content_type:        suggestion["review_warning"] = "返回内容类型不是常规 HTML/JSON，需要人工复核"return suggestiondefcollect_one(item: dict, policy: dict) -> dict:    source = item["source"]    now = datetime.now(timezone.utc).isoformat(timespec="seconds")    robot = robots_status(source["source_url"]) if policy["collection_rules"].get("respect_robots_txt") else {"status": "skipped_by_policy"}    status, text, content_type = fetch_url(source["source_url"])    record = {"slug": item["slug"],"name": item["name"],"source_type": source["source_type"],"source_url": source["source_url"],"fetched_at": now,"http_status": status,"robots": robot,"review_status": "needs_manual_review","suggested_update": suggested_update_from_page(item, text, status, content_type),"evidence": {"page_title": extract_title(text),"meta_description": extract_meta_description(text),"content_type": content_type,"fields_checked": source.get("fields_to_check", []),"risk_level": source.get("risk_level", "medium"),"review_notes": source.get("review_notes", ""),        },    }if status isNone:        record["error"] = content_type or"fetch_failed"return recorddefmain() -> int:    source_map = load_json(SOURCE_MAP)    policy = load_json(SOURCE_POLICY)ifnot policy["collection_rules"].get("manual_review_before_publish"):raise SystemExit("source-policy.json must require manual_review_before_publish=true")    results = [collect_one(item, policy) for item in source_map]    OUTPUT.write_text(json.dumps(results, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")    ok = sum(1for r in results ifisinstance(r.get("http_status"), int) and200 <= r["http_status"] < 400)    failed = len(results) - okprint("✅ 第一版采集器运行完成")print(f"输入工具：{len(source_map)} 个")print(f"成功访问：{ok} 个")print(f"待复核/失败：{failed} 个")print(f"输出文件：{OUTPUT.relative_to(ROOT)}")print("发布闸门：全部记录保持 needs_manual_review，没有覆盖正式 tools.json")return0if __name__ == "__main__":raise SystemExit(main())

这段脚本做了几件事：

• 读取 data/tool-source-map.json；
• 根据 source_url 访问公开页面；
• 尝试提取网页标题和描述；
• 把候选更新写入 data/collected-tools-review.json；
• 每条记录都标记成 needs_manual_review。

这就是工具站采集的第一道安全阀。

第五步｜创建检查脚本

再新建：

scripts/check-collected-review.py

复制下面代码：

#!/usr/bin/env python3from __future__ import annotationsimport jsonfrom pathlib import PathROOT = Path(__file__).resolve().parents[1]DATA = ROOT / "data"OUTPUT = DATA / "collected-tools-review.json"TOOLS = DATA / "tools.json"REQUIRED_TOP_LEVEL = {"slug", "name", "source_url", "fetched_at", "http_status","review_status", "suggested_update", "evidence",}REQUIRED_SUGGESTED = {"website", "summary_candidate", "title_candidate", "pricing_candidate", "tags_candidate"}defmain() -> int:ifnot OUTPUT.exists():raise SystemExit("❌ 缺少 data/collected-tools-review.json，请先运行 collect-first-batch.py")    records = json.loads(OUTPUT.read_text(encoding="utf-8"))ifnotisinstance(records, list) ornot records:raise SystemExit("❌ collected-tools-review.json 必须是非空数组")    errors: list[str] = []    visited_ok = 0for index, record inenumerate(records, start=1):        missing = REQUIRED_TOP_LEVEL - set(record)if missing:            errors.append(f"第 {index} 条缺少字段：{sorted(missing)}")if record.get("review_status") != "needs_manual_review":            errors.append(f"{record.get('slug', index)} review_status 必须是 needs_manual_review")        suggested = record.get("suggested_update") or {}        missing_suggested = REQUIRED_SUGGESTED - set(suggested)if missing_suggested:            errors.append(f"{record.get('slug', index)} suggested_update 缺少：{sorted(missing_suggested)}")ifisinstance(record.get("http_status"), int) and200 <= record["http_status"] < 400:            visited_ok += 1        evidence = record.get("evidence") or {}ifnot evidence.get("fields_checked"):            errors.append(f"{record.get('slug', index)} evidence.fields_checked 不能为空")if TOOLS.exists():        errors.append("data/tools.json 不应在第 6 篇采集器里被创建或覆盖，采集结果只能进入 collected-tools-review.json")if visited_ok < 3:        errors.append(f"成功访问页面少于 3 个，当前为 {visited_ok} 个；请检查网络或数据源")if errors:print("❌ 采集结果检查失败")for error in errors:print("- " + error)return1print("✅ 采集结果检查通过")print(f"待人工复核记录：{len(records)} 条")print(f"成功访问公开页面：{visited_ok} 个")print("审核状态：全部 needs_manual_review")print("正式数据保护：未创建/覆盖 data/tools.json")return0if __name__ == "__main__":raise SystemExit(main())

检查脚本的作用是防呆：

• 输出文件必须存在；
• 每条记录必须有关键字段；
• 审核状态必须是 needs_manual_review；
• 不能提前生成或覆盖正式 tools.json；
• 至少 3 个公开页面访问成功。

第六步｜运行采集器

在 lesson-06 目录下执行：

python3 scripts/collect-first-batch.py

我这次真实运行结果是：

✅ 第一版采集器运行完成输入工具：5 个成功访问：5 个待复核/失败：0 个输出文件：data/collected-tools-review.json发布闸门：全部记录保持 needs_manual_review，没有覆盖正式 tools.json

运行后打开：

data/collected-tools-review.json

你会看到类似结构：

{"slug":"playwright","name":"Playwright","source_type":"docs","source_url":"https://playwright.dev","http_status":200,"review_status":"needs_manual_review","suggested_update":{"website":"https://playwright.dev","summary_candidate":"...","title_candidate":"...","pricing_candidate":"free","tags_candidate":["website","summary","tags","open_source_status"]},"evidence":{"page_title":"...","meta_description":"...","fields_checked":["website","summary","tags","open_source_status"]}}

第七步｜运行检查脚本

继续执行：

python3 scripts/check-collected-review.py

我这次真实输出是：

✅ 采集结果检查通过待人工复核记录：5 条成功访问公开页面：5 个审核状态：全部 needs_manual_review正式数据保护：未创建/覆盖 data/tools.json

也可以把两条命令合在一起：

python3 scripts/collect-first-batch.py && python3 scripts/check-collected-review.py

看到这个结果，就说明本篇完成。

第八步｜用 WorkBuddy 读一遍采集结果

这一步很重要。

不要因为脚本通过，就直接相信采集内容可以上线。

把下面这段话复制到 WorkBuddy：

请读取 data/collected-tools-review.json，帮我做人工审核前检查：1. 哪些工具页面访问成功？2. 哪些字段可以作为候选更新？3. 哪些 summary_candidate 太像官网宣传，需要重写成中性描述？4. 哪些价格、开源、功能信息不能直接采用？5. 请输出一个审核清单，不要修改 data/tools.json。

你要让 WorkBuddy 做的是“帮你审稿”，不是“替你发布”。

这就是把 AI Agent 用在正确位置：

• 它可以帮你写脚本；
• 可以帮你总结候选变化；
• 可以提醒风险；
• 但正式进入工具库前，还是要经过你的审核。

常见报错和排查

1. ModuleNotFoundError

本篇脚本只使用 Python 标准库，正常不需要安装依赖。

如果你看到第三方库缺失，大概率是 WorkBuddy 生成了依赖版本。

处理方式：让 WorkBuddy 改成“只用 Python 标准库”。

2. data/tool-source-map.json 缺失

说明你没有从第 5 篇复制输入文件。

检查：

ls data

至少应该有：

tool-source-map.jsonsource-policy.json

3. 成功访问页面少于 3 个

可能原因：

• 当前网络访问目标站点失败；
• 某些站点临时不可用；
• 目标站点拒绝脚本访问；
• 数据源 URL 写错了。

处理方式：

1. 先打开 data/collected-tools-review.json；
2. 找到带 error 或非 2xx 状态码的记录；
3. 手动访问对应 source_url；
4. 如果页面确实不可访问，就把该工具标记为人工复核；
5. 不要为了通过检查而删除失败记录。

失败记录也是信息，它告诉你这个数据源不稳定。

4. 检查脚本提示生成了 data/tools.json

这说明采集器越权了。

本篇不能生成或覆盖正式工具库。

处理方式：

rm data/tools.json

然后让 WorkBuddy 修改采集器：

请修改 scripts/collect-first-batch.py：只能输出 data/collected-tools-review.json，不允许创建或覆盖 data/tools.json。

5. 页面描述太像广告

采集器提取的是页面里的 title 和 meta description，里面经常会有宣传语。

所以 summary_candidate 只能当参考，不要直接复制到正式工具站。

第 7 篇我们会做审核后台，把这类内容变成待处理项。

完整代码 / 配置汇总

本篇最终文件：

data/tool-source-map.json data/source-policy.json data/collected-tools-review.jsonscripts/collect-first-batch.pyscripts/check-collected-review.pyprompts/workbuddy-first-collector-prompt.md

最重要的命令：

python3 scripts/collect-first-batch.py && python3 scripts/check-collected-review.py

最重要的产物：

data/collected-tools-review.json

最重要的规则：

所有采集结果先进入 needs_manual_review，不直接覆盖正式 tools.json。

这一篇你学会了什么

这一篇你已经完成了工具站数据增长的第一步：

• 用 WorkBuddy 生成一个保守采集器；
• 从第 5 篇的数据源清单读取目标；
• 抓取公开页面的基础信息；
• 输出待审核 JSON；
• 用检查脚本确认没有误改正式数据。

这一步看起来不炫，但非常重要。

很多工具站做不下去，不是因为页面做不出来，而是数据越来越乱：来源不清、字段不稳、描述像广告、价格过期、工具失效。

所以我们从第一版采集器开始，就把“审核队列”这个概念加进去。

下一篇预告

下一篇：

用WorkBuddy做AI工具站07：做一个审核后台

我们会基于这一篇生成的：

data/collected-tools-review.json

做一个最小审核后台，让你能在页面里看见每条采集结果，并给它打上：

• 通过；
• 修改后通过；
• 拒绝；
• 稍后复查。

这样，采集器就不会直接碰正式数据，而是进入一套可控的内容闸门。

资料来源

• 本系列第 5 篇 demo：tool-source-map.json、source-policy.json
• Python 标准库：urllib.request、json、pathlib
• 公开网页基础信息：页面 title、meta description、HTTP 状态码