content-harvester
内容采集服务。给一个来源页 URL,用真实有头浏览器逐个打开内容详情页,抓出每条内容的元数据(标题 / 详情页 URL / 封面图 / 媒体地址);若调用方下发了对象存储(MinIO/S3)配置,顺带把媒体(目前是 m3u8 视频)拉成 mp4、连同封面一起上传,把媒体地址换成对象存储上的稳定 URL(很多站点的媒体地址带 expires 签名、一小时左右就失效)。结果以 resources 列表回传。
主要给「启源 AI 平台」主站调用:主站「多内容数据采集」任务里来源大类选「内容网站」时,把作业 POST 给本服务,本服务返回 resources,主站写进资源库。
当前内置一个 xchina.co 的 scraper(core/scraper.py);接别的来源就再加一个 scraper 模块。
前身是个 Tkinter GUI 视频下载器,已废弃改造成本服务。详细设计见
PRD-minio-upload.md,实现速查见CLAUDE.md。
1. 功能
采集流程(一次作业)
- 收到
POST /agents/{id}/invoke(载荷来自主站),校验 token、source_kind、content_urlhost 白名单,从载荷取 MinIO 配置,生成job_id,立即返回{"task_id": job_id, "status":"running"},作业转后台异步跑(进程内单作业串行,防风控)。 - 用有头 Chromium(容器里靠 Xvfb 提供虚拟显示)打开来源页:
content_url是系列页 → 取页内.item.video卡片里的详情链接,按start_page..end_page翻页(默认只爬给定页);content_url是详情页 → 直接采这一条。
- 逐个打开详情页:处理广告中转页;遇 Cloudflare JS 质询页等它自动放行(最多 ~24s);抓 标题、简介/正文(meta/og/jsonld 描述 + 详情面板文本 + 标签;实在没有就用标题——主站资源库对空正文会拒收)、封面图 URL(截屏 tab 图 → og:image/JSON-LD thumbnail/
<video poster>→ 页面最大图,排除 logo/占位)、视频地址(扫页面 HTML 里的.m3u8/.mp4+ 嗅探网络响应)。 - 若调用方给了完整 MinIO 配置:逐条用 yt-dlp(并发分片下载)+ ffmpeg(合并/转封装,不重编码、无损、很快)把 m3u8 拉成
<videoid>.mp4→ 上传xchina/<videoid>.mp4到桶;封面图也下载上传xchina/<videoid>.<ext>;把该条的video_url/cover_image改成对象存储 URL,原始地址留进extra_data(original_m3u8/original_cover),并记minio_object_video/minio_object_cover/video_bytes。临时文件用完即删,不长期落盘。 - 作业结束:
status=succeeded(只要至少抓到一条)/failed(0 条且有抓取错误)。主站轮询GET /tasks/{job_id}拿结果。
降级(任一环节出问题不让整作业白跑)
| 情况 | 行为 |
|---|---|
| 未给完整 MinIO 配置 | 不上传,video_url 保留原始 m3u8,extra_data.upload_skipped_reason="minio config absent",作业仍 succeeded |
| 某详情页被 Cloudflare 卡住等不出来 | 该条进 item_errors,不计入 resources,不影响其它条 |
m3u8 下载/合并失败、视频超 MAX_VIDEO_MB、MinIO 上传失败 |
该条降级:video_url 保留原始地址,extra_data.upload_skipped_reason 记原因,仍作为一条 resource 回传 |
| 封面下载/上传失败 | cover_image 保留原图床地址,不致命 |
| 整作业 0 条 resources 且有 item_errors | status=failed,error 汇总前几条原因 |
2. 技术栈与结构
- FastAPI + uvicorn —— HTTP 服务。
- Playwright(Chromium,有头模式)+ Xvfb —— 抓页面。有头是必须的:Cloudflare 的 JS 质询对真实有头浏览器才会自动放行,无头 chromium-headless-shell 会被一眼识破。容器里用
docker-entrypoint.sh起一个Xvfb :99提供虚拟显示。x86_64 上还可改用真实 Google Chrome 通道(BROWSER_CHANNEL=chrome)更稳。 - yt-dlp + ffmpeg —— m3u8 分片下载 + 合并成 mp4。ffmpeg 是
apt install进镜像的,不碰宿主机。 - minio SDK —— 上传对象存储(
fput_object),兼容 S3 协议。 - 作业模型 —— 进程内内存级 job 注册表 +
asyncio.create_task后台跑;asyncio.Lock串行化(同一时刻只跑一个浏览器作业,防站点风控);作业状态只在内存,服务重启即丢——主站靠轮询拿结果,够用。重计算(yt-dlp/minio 都是同步库)用asyncio.to_thread跑,不阻塞事件循环。 - 不持有任何密钥 —— MinIO 凭据、token 都由部署方/调用方提供(env 或 invoke 载荷),仓库里不含。日志里不打印
secret_key等敏感值。
core/
browser.py # Playwright 持久化 BrowserContext(资料目录可挂卷),反自动化参数 / zh-CN / Asia-Shanghai
scraper.py # xchina.co 专属 scraper:系列页翻页、详情页解析(标题/简介/封面/视频)、广告中转页处理、Cloudflare 质询等待
media.py # MinioConfig(从载荷/env 解析)、download_video(yt-dlp+ffmpeg)、download_cover、upload_to_minio、extract_video_id
models.py # ScrapedItem(字段对齐主站 ResourceIngestItem)、ScrapeResult
server/
app.py # FastAPI:/agents/{id}/invoke、/tasks/{id}、/tasks/{id}/cancel、/healthz;Bearer 鉴权;SSRF host 白名单
jobs.py # 内存 job 表 + 后台执行 + 单并发串行 + 下载/上传/降级 + 临时目录清理
Dockerfile # base mcr.microsoft.com/playwright/python;apt 装 xvfb + ffmpeg;ENTRYPOINT 起 Xvfb → uvicorn
docker-entrypoint.sh
requirements.txt # playwright / fastapi / uvicorn / requests / yt-dlp / minio
PRD-minio-upload.md # 设计文档
CLAUDE.md # 给 AI / 维护者的实现速查
加新来源:新建 core/scraper_<site>.py(实现 run(url, start, end) -> ScrapeResult),在 server/jobs.py 里按 content_url host 或载荷里的某字段选用对应 scraper。core/media.py、server/ 不用动。
3. HTTP 接口
实现的是主站 lobster 执行器的调用契约(主站 agent-system-backend/app/services/lobster_client.py::call_lobster_agent:POST /agents/{id}/invoke 取 task_id → 轮询 /tasks/{id} 直到 status∈{succeeded,failed} → 用 output/error)。
POST /agents/{agent_id}/invoke
请求体(= 主站发来的合并载荷):
| 字段 | 必填 | 说明 |
|---|---|---|
content_url |
是 | 来源页 / 详情页 URL;host 必须在 ALLOWED_HOSTS 白名单(默认 xchina.co) |
source_kind |
否 | 非 website 直接 400 |
platform_name |
否 | 作为 source_name,默认 xchina |
start_page / end_page |
否 | 默认 1/start_page(即只爬给定页/这一条);区间超 MAX_PAGES 会被截断 |
minio_endpoint / minio_access_key / minio_secret_key / minio_bucket |
否 | 四者齐全才启用上传;缺则降级。也可改由本服务 env 提供(MINIO_*),载荷优先 |
_crawl_config 等其它字段 |
— | 本服务忽略,主站自己用 |
响应:{"task_id": "<job_id>", "status": "running"},立即返回。
GET /tasks/{task_id}
{
"status": "running" | "succeeded" | "failed",
"output": { // succeeded 时
"resources": [
{
"source_type": "video",
"source_name": "xchina",
"source_url": "https://xchina.co/video/id-XXXX.html", // 主站资源库按它去重
"content_type": "video",
"title": "...",
"summary": "...", // 简介(可空)
"content": "...", // 正文(保证非空,实在没有就是标题)——主站对空正文拒收
"cover_image": "<对象存储URL 或 原图床URL>",
"video_url": "<对象存储URL 或 原始m3u8URL>",
"extra_data": {
"referer": "...", "page_no": 1, "collector": "xchina", "video_id": "XXXX",
// 上传成功还有:
"original_m3u8": "...", "original_cover": "...",
"minio_object_video": "xchina/XXXX.mp4", "minio_object_cover": "xchina/XXXX.jpg",
"video_bytes": 78167166
// 没上传则有:"upload_skipped_reason": "minio config absent | download failed: ... | video too large: ... | minio upload failed: ..."
}
}
// ...
],
"pages_scanned": 1,
"item_errors": [ {"source_url": "...", "reason": "..."} ]
},
"error": null | "...",
"log_tail": ["HH:MM:SS ...", ...] // 最近 80 行作业日志
}
每条 resources[i] 的顶层字段就是主站 ResourceIngestItem 的字段集,主站可直接 model_validate 入库。
其它
POST /tasks/{task_id}/cancel—— 标记取消(仅 running 时有效)。GET /healthz——{"ok": true, "headless": false, "allowed_hosts": [...]}。- 鉴权:设了
HARVESTER_TOKEN则所有/agents/*/tasks/*要求Authorization: Bearer <token>(即主站对应 Skill 的lobster_api_key);没设则不鉴权(仅限内网/本地)。 - 并发:进程内单作业串行;已有作业在跑时新作业排队。
4. 配置(环境变量)
| 变量 | 默认 | 说明 |
|---|---|---|
HARVESTER_TOKEN |
空 | Bearer token;空=不鉴权 |
HEADLESS |
0 |
0=有头(容器内靠 Xvfb,过 Cloudflare 必须);1=无头(本地调试,过不了详情页) |
BROWSER_CHANNEL |
空 | x86_64 可设 chrome(需 Dockerfile 里 playwright install chromium chrome);arm64 无 Chrome 包 |
BROWSER_PROFILE_DIR |
/data/browser_profile |
持久浏览器资料目录(挂卷;可预置过了 Cloudflare 的 cookie 作为反爬退路) |
ALLOWED_HOSTS |
xchina.co |
content_url host 白名单(SSRF 防护),逗号分隔 |
MAX_PAGES |
50 |
单作业最多翻页数 |
CONCURRENT_FRAGMENTS |
8 |
yt-dlp 并发分片数 |
MAX_VIDEO_MB |
0 |
单视频大小上限(0=不限),超过则跳过上传、降级保留原媒体地址 |
TMP_DIR |
/tmp |
临时下载目录(<TMP_DIR>/content-harvester/<jobid>/,用完即删) |
MINIO_ENDPOINT / MINIO_ACCESS_KEY / MINIO_SECRET_KEY / MINIO_BUCKET |
空 | 对象存储配置;invoke 载荷里没传时的兜底 |
5. 部署
Docker(推荐)
git clone -b dev https://git.aidfxn.com/admin/content-harvester.git
cd content-harvester
docker build -t content-harvester . # base 镜像 ~1GB+,首次几分钟;之后改代码增量很快
docker run -d --name content-harvester \
--restart unless-stopped \
--network <主站 docker compose 的网络名> \ # 跟主站 backend 同一个 docker 网络,让主站能用 http://content-harvester:8000 访问
-p 8019:8000 \ # host 端口随意,主要给本机直接调试用
-e HARVESTER_TOKEN=<和主站 Skill 的 lobster_api_key 一致> \
-e MINIO_ENDPOINT=<...> -e MINIO_BUCKET=<...> -e MINIO_ACCESS_KEY=<...> -e MINIO_SECRET_KEY=<...> \
-v content-harvester-profile:/data/browser_profile \
--memory=2g --cpus=2 \
content-harvester
curl -s localhost:8019/healthz # {"ok":true,"headless":false,...}
镜像里已含 xvfb(有头浏览器)+ ffmpeg(yt-dlp 合并 m3u8)。资源占用主要是 Chromium 内存(给 2g 够用)+ 下载视频时的临时磁盘(每条用完即删,同一时刻最多一个视频在磁盘上;超大视频用 MAX_VIDEO_MB 限);ffmpeg 合并很轻(无损 remux,几秒)。
MinIO 凭据也可不用 env,改由主站把绑定的
minio_*Skill 变量并入 invoke 载荷下发(见下「与主站交互」)——那样本服务 env 里不放任何密钥。
更新
本仓库没有 gitea 自动部署 workflow,要手动:
cd <部署目录> && git pull origin dev
docker build -t content-harvester .
docker rm -f content-harvester
docker run -d ... content-harvester # 同上面那串参数
各环境的具体服务器、目录、容器、token / 凭据见
DEPLOY.md与本地运维记录(不入库)。
6. 与主站的交互
调用链
用户在主站「任务创建」选采集 Skill(lobster 执行器),来源大类=内容网站,地址填来源页 URL
│ task.input_params = { source_kind:"website", content_url:"...", start_page?, end_page?, _crawl_config:{dedupe_strategy,default_status,visibility} }
▼
主站 TaskService.create() → 不阻塞,asyncio.create_task(_run_lobster_task_bg)
│ _dispatch_lobster:① 解析该 Skill 绑定的 Skill 变量(含 minio_* 等)→ 并入 payload(Skill 变量同名覆盖 input_params)
│ ② POST {lobster_api_url}/agents/{agent_id}/invoke body=payload Authorization: Bearer {lobster_api_key}
│ ③ 轮询 GET {lobster_api_url}/tasks/{task_id}(crawl 类超时 CRAWL_LOBSTER_TIMEOUT,默认 7200s)
▼
content-harvester:抓页面 → 下 m3u8 转 mp4 → 传 MinIO → 回传 output.resources
▼
主站 _dispatch_lobster 把结果写回 task.result(={success,output,error,duration_ms}),task.status=SUCCEEDED
│ _run_lobster_task_bg:成功后调 _maybe_ingest_to_resource_pool(task)
│ 从 task.result.output.resources 逐条 ResourceIngestItem.model_validate
│ 组 ResourceIngestRequest(task_id/skill_id/created_by/product_id + _crawl_config 里的 dedupe_strategy/default_status/visibility)
│ ResourcePoolService.ingest_resources():内容审核(违规 block)→ 写 ResourcePool → 派 AI 改标题/润色正文任务
▼
资源库列表出现这些内容,video_url/cover_image 指向 MinIO,长期有效;走的是和 TG 采集完全相同的入库/审核/润色流程。
主站需要的配套(已在 agent-system-backend 做好)
app/services/task_service.py 5 处改动:① 非批次 lobster 任务改 asyncio.create_task(_run_lobster_task_bg) 后台执行,不内联阻塞创建请求;② _run_lobster_task_bg 成功后调 _maybe_ingest_to_resource_pool(原来只有 script 执行器走入库);③ _maybe_ingest_to_resource_pool 兼容从 task.result["output"]["resources"] 取(lobster 嵌套结构);④ _dispatch_lobster 把 _resolve_skill_variables 解出的 Skill 变量并入发给远端的 payload;⑤ crawl 类 lobster 任务轮询超时改 CRAWL_LOBSTER_TIMEOUT(env,默认 7200s)。前端 agent-system-frontend 不用改(CrawlerTaskBlock 已有 source_kind=website)。
主站侧配置(运维一次性)
「Skill 管理」新建/编辑一个采集 Skill:
executor_type=lobster、skill_category=crawllobster_api_url=http://content-harvester:8000(同 docker compose 网络)、lobster_api_key=<HARVESTER_TOKEN>、lobster_agent_id=任意(如xchina)input_schema含source_kind字段(前端才渲染采集表单)+ 声明content_url(必填),可选start_page/end_page- (可选,若不想在 content-harvester 容器里配 MinIO env)变量关联里勾上
minio_endpoint/minio_access_key/minio_secret_key/minio_bucket这几个 Skill 变量,值由运维在主站填——_dispatch_lobster会把它们并入 payload 下发
注意:这是个独立的 Skill,跟原有「多类型数据采集」(script 执行器,跑 clawler 脚本,主打 Telegram)并存——执行器类型互斥(RB-13),不能合成一个。用户建任务时选哪个 Skill 就走哪条采集路径。
已知行为 / 注意点
- 主站资源库对空
content直接拒收(reason:"正文为空,不予入库",且_maybe_ingest_to_resource_pool只记 log 不抛错 → 任务仍 SUCCEEDED 但资源没进库)。所以本服务保证每条content非空(抓不到描述就用标题兜底)。 - 入库成功后主站会自动派 AI 改标题任务,资源库里的
title会被润色掉(如「0144 小雪」→「小雪:怀旧风格丝袜摄影大赏」),属预期。文本内容审核、AI 正文润色也一并跑(和 TG 采集同一套);视频封面图目前没走图片 OCR 审核(那套只对images数组里的 URL 触发,而视频资源只有单个cover_image)。 - m3u8 媒体地址带
expires签名约 1h 失效——所以才在采集时就转 mp4 上传对象存储;不上传的话资源库里那个 URL 一会儿就 403。 - 性能:每条
15s 抓页 + 几十秒几分钟下载 + 几秒~几十秒上传;一页 11 条实测 ~4 分钟(服务器带宽好的话)。CRAWL_LOBSTER_TIMEOUT默认 7200s 够用。多页/大量视频会很慢,属预期。
7. 反爬 / 限制
- 详情页常被 Cloudflare 挡:只有真实有头浏览器(
HEADLESS=0+ Xvfb,或真实 Chrome)才能自动过 JS 质询;_assert_not_blocked会等质询自动放行(最多 ~24s),仍卡住则该条进item_errors。 - 站点策略变化(改成交互式验证码、改 DOM 结构)可能导致采集失效。退路:在桌面环境用同一套 Chromium 持久目录登录 / 过一次验证,把
browser_profile目录拷进容器挂载的卷。站点结构变了就改对应的core/scraper*.py(选择器/正则/JS 片段)。 - 单并发是有意的;并发跑多个浏览器作业容易触发站点风控。