content-harvester/PRD-minio-upload.md

PRD —— 采集服务下载视频并上传 MinIO,资源库存 MinIO 地址

状态:已实现 · 范围:本工程(content-harvester) + 主站 agent-system-backend 少量改动 · 实现说明见 CLAUDE.md

1. 背景

现状:主站「多内容数据采集 / 来源=内容网站」任务 → 走 lobster 执行器调本采集服务 → 服务用无头 Chrome(xvfb)抓 xchina 详情页,回传每条视频的 {title, source_url(详情页), cover_image(图床URL), video_url(m3u8 签名URL)} → 主站 _maybe_ingest_to_resource_pool 写入 ResourcePool。

问题:video_url 现在存的是 xchina CDN 上带 expires + md5 的限时签名 m3u8 地址,大约 1 小时后 xchina 自己返回 403。资源库里的视频地址会失效,下游(转码 / 搬运 / 发布到 smnqae)无法稳定使用。

目标:采集时就把 m3u8 拉成 mp4、连同封面图一起传到主站现有的 MinIO(<bucket> 桶),资源库 video_url / cover_image 改存 MinIO 上的稳定地址。MinIO 配置不在采集服务里硬编码,而是由主站通过 lobster payload 下发(复用主站已有的 minio_* Skill 变量),采集服务保持"不持有密钥"。

2. 范围

In scope

• 采集服务:m3u8 → mp4 下载(yt-dlp 并发分片 + ffmpeg 合并)、封面图下载、上传 MinIO、video_url/cover_image 改为 MinIO 地址。
• 采集服务:从 lobster payload 读取 MinIO 配置(minio_endpoint/minio_access_key/minio_secret_key/minio_bucket);缺失时降级(见 §9)。
• 主站:_dispatch_lobster 把解析后的 Skill 变量并入发给采集服务的 payload(目前只有 script 执行器这么做);把 crawl 类 lobster 任务的轮询超时调到足够长(下载转码耗时)。
• 配置:给 xchina 那个 Skill 绑定 4 个 minio_* Skill 变量(运维操作,非代码)。

Out of scope

• 不改主站 ResourcePool 表结构、不改 ResourceIngestItem 字段(仍是 video_url/cover_image 两个字符串,只是值变了)。
• 不在主站后端加 ffmpeg/yt-dlp/minio(媒体重活全在采集服务)。
• 不做断点续传、不做 MinIO 去重(同 source_url 重复采集会覆盖同名对象,可接受)。
• 不动 OCR / 人工复核 / 发布流程。

3. 参与系统

系统	角色
主站后端 agent-system-backend	创建任务、派发 lobster 调用、轮询、把回传 resources 入库;持有 MinIO 凭据(作为 Skill 变量)
本采集服务 content-harvester	抓 xchina 元数据 + 下载 m3u8→mp4 + 下载封面 + 上传 MinIO + 回传 resources
xchina.co	数据源(列表页开放,详情页 Cloudflare JS 质询,媒体 m3u8 限时签名)
对象存储(MinIO/S3)	视频/封面对象存储,产出稳定可访问 URL;endpoint/bucket/凭据由主站作为 Skill 变量提供

4. 端到端流程

用户在主站创建任务(选 xchina Skill,来源大类=内容网站,填系列页/详情页 URL)
        │  task.input_params = { source_kind:"website", content_url:"...", platform_name?, start_page?, end_page?, _crawl_config:{...} }
        ▼
主站 create() → 后台 _run_lobster_task_bg → _dispatch_lobster
        │  ① 解析该 Skill 绑定的 Skill 变量(含 minio_endpoint/access_key/secret_key/bucket)→ var_env
        │  ② payload = { ...task.input_params, ...var_env }      (Skill 变量同名覆盖 input_params,与 script 执行器一致)
        │  ③ POST {lobster_api_url}/agents/{agent_id}/invoke   body=payload   Authorization: Bearer {lobster_api_key}
        ▼
采集服务 /agents/{id}/invoke
        │  校验 token、source_kind=website、content_url host 白名单
        │  从 payload 取 minio 配置(4 个 key) + start_page/end_page/platform_name
        │  生成 job_id,后台 asyncio 任务开跑,立即返回 {"task_id": job_id, "status":"running"}
        ▼
后台采集作业(单并发串行)
   for 每页(start..end):
     打开列表页(headed Chrome/xvfb,过 Cloudflare)→ 取 .item.video 卡片里的详情链接
     for 每个详情链接:
        打开详情页(过 Cloudflare、处理广告中转页)
        抓 title / cover_image(图床URL) / video_url(m3u8 签名URL,带 referer)
        ┌─ 若拿到 minio 配置且拿到 m3u8:
        │    yt-dlp(并发分片, 带 Referer 头)下载 → ffmpeg 合并 → /tmp/<jobid>/<videoid>.mp4
        │    minio.fput_object(bucket, "xchina/<videoid>.mp4", mp4, content_type="video/mp4")
        │    requests 下载封面图 → minio.fput_object(bucket, "xchina/<videoid>.jpg", img, content_type=...)
        │    video_url = "{endpoint}/{bucket}/xchina/<videoid>.mp4"
        │    cover_image = "{endpoint}/{bucket}/xchina/<videoid>.jpg"
        │    extra_data += { original_m3u8: <原签名URL>, original_cover: <原图床URL>, minio_object_video, minio_object_cover, video_bytes }
        │    删 /tmp/<jobid>/<videoid>.mp4
        └─ 否则(无 minio 配置 / 下载失败 / 超大):降级 —— 保留原 m3u8 / 原图床 URL,extra_data += { upload_skipped_reason: "..." }
        累积一条 ScrapedItem
   作业结束:job.status = succeeded(只要至少抓到 1 条)/ failed(0 条且有错误)
        ▼
主站轮询 GET {lobster_api_url}/tasks/{job_id}  直到 status ∈ {succeeded, failed}
        │  succeeded → body.output = { resources:[...], pages_scanned, item_errors }
        ▼
主站 _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() → 写 ResourcePool 表(每条 video_url/cover_image 已是 MinIO 地址)
        ▼
完成。资源库列表里出现这些视频,video_url 指向 MinIO,长期有效。

5. 数据流转(字段级)

5.1 主站 → 采集服务:POST /agents/{agent_id}/invoke 请求体(= 合并后的 payload)

字段	来源	说明
source_kind	任务表单(CrawlerTaskBlock)	必须 "website",否则 400
content_url	任务表单「地址」	系列页或单个详情页 URL;host 必须在采集服务白名单(默认 xchina.co)
platform_name	任务表单	可选 → 作为 source_name,默认 "xchina"
start_page / end_page	任务表单(可选,Skill input_schema 里声明)	默认 1 / start_page —— 即只爬给定 URL 这一页/这一条
_crawl_config	主站任务创建页注入	{dedupe_strategy, default_status, visibility},采集服务不用,原样回传给主站也行;实际由主站 ingest 用
minio_endpoint	Skill 变量(主站 _dispatch_lobster 注入)	形如 <minio-endpoint>
minio_access_key / minio_secret_key	Skill 变量	MinIO 鉴权
minio_bucket	Skill 变量	形如 <bucket>

注:Skill 变量同名覆盖 input_params(与 script 执行器 _resolve_skill_variables 后的合并优先级一致)。minio_secret_key 是敏感值,采集服务日志里不打印。

5.2 采集服务内部:ScrapedItem(每条视频)

字段	上传前	上传后
source_url	xchina 详情页 URL(不变)	同
title	抓到的标题	同
source_name	platform_name 或 "xchina"	同
cover_image	xchina 图床 URL(img.xchina.io/...)	{endpoint}/{bucket}/xchina/<videoid>.jpg(上传成功时)
video_url	xchina m3u8 签名 URL	{endpoint}/{bucket}/xchina/<videoid>.mp4(上传成功时)
video_id	从 source_url 的 id-(\w+)\.html 提取	同;作 MinIO object key 前缀
page_no	列表页页码	同

5.3 采集服务 → 主站:GET /tasks/{job_id} 响应里的 output

{
  "resources": [
    {
      "source_type": "video",
      "source_name": "xchina",
      "source_url": "https://xchina.co/video/id-XXXXXXXXXXXXX.html",
      "content_type": "video",
      "title": "0145 依依",
      "cover_image": "<minio-endpoint>/<bucket>/xchina/XXXXXXXXXXXXX.jpg",
      "video_url":   "<minio-endpoint>/<bucket>/xchina/XXXXXXXXXXXXX.mp4",
      "extra_data": {
        "referer": "https://xchina.co/video/id-XXXXXXXXXXXXX.html",
        "page_no": 1,
        "collector": "xchina",
        "original_m3u8": "https://video.xchina.download/m3u8/XXXXXXXXXXXXX/720.m3u8?expires=...&md5=...",
        "original_cover": "https://img.xchina.io/photos/.../00060.jpg",
        "minio_object_video": "xchina/XXXXXXXXXXXXX.mp4",
        "minio_object_cover": "xchina/XXXXXXXXXXXXX.jpg",
        "video_bytes": 41234567
      }
    }
    // ...
  ],
  "pages_scanned": 1,
  "item_errors": [ { "source_url": "...", "reason": "..." } ]
}

每条 resources[i] 的顶层字段就是主站 ResourceIngestItem 的字段集,主站可直接 model_validate。

5.4 主站入库:ResourcePool 行(节选)

source_type=video, content_type=video, source_url(去重键), title, cover_image(MinIO URL), video_url(MinIO URL), extra_data(含 original_m3u8 等), task_id/skill_id/created_by/product_id(主站补), status= _crawl_config.default_status(ready/pending), visibility= _crawl_config.visibility(普通用户强制 private)。

6. 各步骤逻辑

6.1 主站 —— Skill 配置(运维一次性)

• 「Skill 管理」里那个采集 Skill(executor_type=LOBSTER、skill_category=CRAWL、lobster_api_url 指向采集服务):在变量关联里勾上 minio_endpoint、minio_access_key、minio_secret_key、minio_bucket 这 4 个 Skill 变量(它们的实际值由运维在主站配置,本服务不持有)。input_schema 仍需含 source_kind 字段(前端才渲染采集表单),并声明 content_url(必填),可选 start_page/end_page。

6.2 主站 —— _dispatch_lobster 注入 Skill 变量(代码,~5–8 行)

• 在 call_lobster_agent 之前:var_env, var_files_dir = await self._resolve_skill_variables(skill, task);payload = {**(task.input_params or {}), **var_env};把 payload 传给 call_lobster_agent(原来传 task.input_params)。var_files_dir 若产生则 finally 里 shutil.rmtree(本场景 minio 变量是 TEXT,通常不产生 FILE 目录)。
• crawl 类任务轮询超时:timeout_sec = CRAWL_LOBSTER_TIMEOUT(默认 7200) if skill.skill_category == "crawl" else 30(下载转码 + 上传可能几十分钟到小时级;call_lobster_agent 在 asyncio.create_task 后台跑,长时持有连接可接受)。

6.3 采集服务 —— /agents/{id}/invoke

• 鉴权(Bearer == HARVESTER_TOKEN,未配则不校验)。
• 校验 source_kind(若有)== website;content_url 非空且 host 在 ALLOWED_HOSTS。
• 从 payload 取 MinIO 配置:minio_endpoint/minio_access_key/minio_secret_key/minio_bucket 四者齐全才启用上传;任一缺失 → 本次作业不上传(降级)。也支持从采集服务自身 env 兜底(可选,默认不配)。
• 取 start_page/end_page(默认 1..1,上限 MAX_PAGES)、source_name=platform_name or "xchina"。
• jobs.submit(...) → 后台跑;立即返回 {"task_id": job_id}。

6.4 采集服务 —— 抓取(不变,已实现)

• headed Chromium + Xvfb 过 Cloudflare(质询页等其自动放行,最多 ~24s);列表页取 .item.video 详情链接;详情页处理广告中转页;抓 title / cover(截屏tab图 → og:image/JSON-LD thumbnail/<video poster> → 页面最大图,排除 logo/占位)/ video(扫 page.content() 的 .m3u8/.mp4 + 网络响应嗅探)。

6.5 采集服务 —— 下载 + 上传(新增)

• video_id = re.search(r"/video/id-([A-Za-z0-9_-]+)\.html", source_url);取不到则用 url hash 兜底。
• 下载视频:m3u8 → 用 yt-dlp Python API(format 默认、concurrent_fragments=N(env CONCURRENT_FRAGMENTS,默认 8)、http_headers={"Referer": source_url, "User-Agent": ...}、outtmpl=/tmp/<jobid>/<videoid>.%(ext)s)→ 落 mp4(yt-dlp 内部调 ffmpeg 合并)。直链 mp4 则直接流式下载。
  • 大小保护:env MAX_VIDEO_MB(默认 0=不限);超过则跳过上传、降级保留 m3u8,记 extra_data.upload_skipped_reason="video too large: Xmb"。
• 下载封面:requests.get(cover_image, headers={Referer:..., UA:...}, stream=True) → 落 /tmp/<jobid>/<videoid>.<ext>(从 URL 后缀猜,默认 .jpg)。失败不致命:cover_image 保留原图床 URL。
• 上传 MinIO:Minio(host, access_key, secret_key, secure=endpoint以https开头).fput_object(bucket, "xchina/<videoid>.mp4", path, content_type="video/mp4");封面同理 content_type 按扩展名。公开 URL = endpoint.rstrip('/') + '/' + bucket + '/' + object_key(与脚本侧 upload_file_to_minio 的 public_base/relative_path 拼法一致)。
• 回写:video_url/cover_image 改 MinIO URL;extra_data 加 original_m3u8 / original_cover / minio_object_video / minio_object_cover / video_bytes。
• 清理:无论成败,作业结束(或每条处理完)删 /tmp/<jobid>/。采集服务容器不长期落盘。

6.6 采集服务 —— 组装回传 & 主站入库(不变,已实现)

• output = {resources:[ScrapedItem.to_ingest_dict()...], pages_scanned, item_errors};job.status succeeded/failed。
• 主站 _run_lobster_task_bg 成功后 _maybe_ingest_to_resource_pool(已兼容 result.output.resources)→ ResourcePoolService.ingest_resources 写库。

7. 接口契约(汇总)

接口	方向	关键约定
POST /agents/{agent_id}/invoke	主站→采集	body=合并后 payload(§5.1);返回 {"task_id": str, "status":"running"};鉴权 Authorization: Bearer <token>
GET /tasks/{task_id}	主站→采集	`{"status":"running"
POST /tasks/{task_id}/cancel	主站→采集	标记取消(仅 running)
GET /healthz	任意	健康检查

与主站 app/services/lobster_client.py::call_lobster_agent 完全对齐(POST /agents/{id}/invoke 取 task_id → 轮询 /tasks/{id} 直到 status∈{succeeded,failed} → 用 output/error)。

8. MinIO 对象命名约定

• 桶:由 payload minio_bucket 指定(默认 <bucket>)。
• object key:xchina/<videoid>.mp4、xchina/<videoid>.jpg(或封面原扩展名)。<videoid> = xchina 详情页 URL 里的 id 段,稳定且唯一。
• 同一视频重复采集 → 覆盖同名对象(幂等;不另起版本)。
• 公开 URL:<minio_endpoint>/<bucket>/<key>(假定该桶对象可匿名 GET —— 与现有 clawler 脚本写入的对象同桶同策略)。

9. 异常与降级

情况	行为
payload 里 MinIO 4 个 key 不齐	本次作业整体不上传;video_url 保留原 m3u8,extra_data.upload_skipped_reason="minio config absent";作业仍可 succeeded
某条详情页被 Cloudflare 卡住(等不出来)	该条进 item_errors,不计入 resources;不影响其它条
m3u8 下载/合并失败	该条降级:video_url 保留原 m3u8,extra_data.upload_skipped_reason="download failed: ...";仍作为一条 resource 回传
视频超过 MAX_VIDEO_MB	同上,reason="video too large"
封面下载失败	cover_image 保留原图床 URL;不致命
MinIO 上传失败(网络/鉴权/桶不存在)	该条降级保留原 URL,reason="minio upload failed: ..."；连续多条失败可考虑整作业 failed(可选,先不做)
整作业:0 条 resources 且有 item_errors	job.status="failed",error 汇总前几条原因
主站轮询超时(超过 CRAWL_LOBSTER_TIMEOUT)	主站 LobsterResult(success=False, error="polling timeout") → task FAILED;采集服务那边作业可能还在跑(无害,结果丢弃);可调大超时

降级策略的意义:即使 MinIO 这条链路出问题,也不要让整个采集白跑 —— 至少把元数据 + 原始(限时)URL 入库,人工/后续可补救。

10. 配置项

采集服务(env)

变量	默认	说明
HARVESTER_TOKEN	空	Bearer token(= 主站 Skill 的 lobster_api_key)
HEADLESS	0	0=有头(容器内靠 Xvfb);过 Cloudflare 必须
BROWSER_CHANNEL	空	x86_64 服务器可设 chrome 用真实 Chrome(arm64 无 Chrome 包)
BROWSER_PROFILE_DIR	/data/browser_profile	持久浏览器资料目录(挂卷)
ALLOWED_HOSTS	xchina.co	content_url host 白名单(SSRF)
MAX_PAGES	50	单作业最多翻页数
CONCURRENT_FRAGMENTS	8	yt-dlp 并发分片数
MAX_VIDEO_MB	0	单视频大小上限(0=不限),超过则跳过上传降级
TMP_DIR	/tmp	临时下载目录(用完即删)
MINIO_ENDPOINT / MINIO_ACCESS_KEY / MINIO_SECRET_KEY / MINIO_BUCKET	空	仅作 payload 缺失时的兜底;生产默认不配,由主站下发

主站(已有 Skill 变量,运维勾选关联即可)

minio_endpoint、minio_bucket、minio_access_key、minio_secret_key(实际值由运维在主站配置,本仓库不含)。

主站(新增 env / 常量,可选)

CRAWL_LOBSTER_TIMEOUT(默认 7200s):crawl 类 lobster 任务的轮询超时。

11. 性能 / 时延

• 每条:抓元数据 15s + 下载 m3u8→mp4(随视频大小,几十秒几分钟)+ 上传 MinIO(几秒~几十秒)。一页 11 条估计 30 分钟 ~ 1 小时+。
• 单并发(进程内串行),避免 xchina 风控;多页/大量视频会很慢,属预期。
• 主站侧靠 asyncio.create_task 后台跑 + 轮询,不阻塞任何用户请求。

12. 安全

• minio_secret_key 等敏感值:采集服务不写日志(log_tail / logger 都过滤);只在内存里用。
• SSRF:content_url host 白名单;封面图下载也只允许 xchina/img.xchina.io 域(或宽松些但限制 scheme=https)。
• token 鉴权:/agents/* /tasks/* 都校验 Bearer。
• 采集服务容器不长期落盘媒体文件(临时目录用完即删);浏览器 profile 卷里可能含 Cloudflare cookie,按敏感卷管理。

13. 改动清单(文件级)

采集服务(本工程)

• requirements.txt:加 yt-dlp、minio、requests。
• Dockerfile:apt-get install -y ffmpeg(与 xvfb 一起);其余不变。
• core/media.py(新):download_video(m3u8_or_mp4_url, referer, dest_dir, video_id, max_mb, concurrent_fragments) -> Path、download_cover(url, referer, dest_dir, video_id) -> Path | None、upload_to_minio(local_path, object_key, content_type, cfg) -> public_url、build_public_url(endpoint, bucket, key)、MinioConfig dataclass + from_payload(payload) / is_complete。
• core/models.py:ScrapedItem 加 video_id 字段;to_ingest_dict 的 extra_data 支持附加 original_m3u8/original_cover/minio_object_*/video_bytes/upload_skipped_reason。
• server/jobs.py:Job 带 minio_cfg;_run 里每条 scrape_detail 后,若 minio_cfg.is_complete 则 download+upload+回写,失败降级;作业结束清临时目录;敏感值不入 log_lines。
• server/app.py:invoke 从 payload 解析 MinioConfig.from_payload(payload),传给 jobs.submit。
• README.md / CLAUDE.md:更新(新增依赖、env、行为、对接说明)。

主站 agent-system-backend

• app/services/task_service.py::_dispatch_lobster:① var_env = await self._resolve_skill_variables(skill, task) 后 payload = {**(task.input_params or {}), **var_env} 传给 call_lobster_agent;② crawl 类用 CRAWL_LOBSTER_TIMEOUT(默认 7200)。(_run_lobster_task_bg 成功后调 _maybe_ingest_to_resource_pool 与 _maybe_ingest_to_resource_pool 读 output.resources 已在前面做过。)
• app/config.py(可选):加 CRAWL_LOBSTER_TIMEOUT。

配置(非代码)

• 主站「Skill 管理」给 xchina Skill 绑定 4 个 minio_* Skill 变量。
• 采集服务部署到 x86_64 时:Dockerfile 改 playwright install chromium chrome + 设 BROWSER_CHANNEL=chrome(更稳过 Cloudflare)。

14. 验收

采集服务本地(docker):

1. docker build + docker run(HEADLESS=0),/healthz 返回 {"ok":true,"headless":false}。
2. curl -X POST /agents/xchina/invoke -d '{"source_kind":"website","content_url":"<xchina系列页URL>","minio_endpoint":"<minio-endpoint>","minio_access_key":"...","minio_secret_key":"...","minio_bucket":"<bucket>"}' → 拿 task_id。
3. 轮询 /tasks/{id} 到 succeeded:output.resources 每条 video_url 形如 <minio-endpoint>/<bucket>/xchina/<id>.mp4,cover_image 形如 .../xchina/<id>.jpg,extra_data 含 original_m3u8、minio_object_video、video_bytes。
4. curl -I <那个 mp4 URL> 返回 200(对象确实上传成功且可访问);随便取一个验证能播放。
5. 不带 minio 字段再跑一次 → video_url 仍是 m3u8,extra_data.upload_skipped_reason="minio config absent",作业仍 succeeded。
6. 负向:错误 token → 401;source_kind:"platform" → 400;非白名单 host → 400。

主站联调(dev 环境): 7. 给 xchina Skill 绑定 4 个 minio 变量;创建任务(选该 Skill、来源大类=内容网站、地址填 xchina 系列页、入库后状态=ready)。 8. 等任务 SUCCEEDED;GET /resource-pool 出现这些视频,详情里 video_url 是 MinIO 地址、cover_image 是 MinIO 地址、extra_data 含 original_m3u8、task_id/skill_id/created_by 正确、status=ready。 9. curl -I 资源里的 video_url → 200,且过 1 小时后再 curl -I 仍 200(证明不再依赖 xchina 限时签名)。 10. 同 URL 再跑一次,dedupe_strategy=skip 时资源库不重复入库;MinIO 上同名对象被覆盖。 11. UI/e2e 步骤丢给子 agent 跑,主线程只收结论摘要。