content-harvester/CLAUDE.md

CLAUDE.md — content-harvester

内容采集服务(FastAPI + Playwright headed-Chrome/xvfb)。给一个来源页 URL,逐个打开内容详情页抓元数据;若调用方下发了对象存储(MinIO/S3)配置,把媒体 + 封面一起上传,媒体地址改成对象存储稳定地址(很多站点的媒体地址带 expires 签名、约 1h 失效)。以 resources 列表回传。给「启源 AI 平台」主站「多内容数据采集 / 来源=内容网站」任务(lobster 执行器)调用,主站把结果写进资源库。两类内容:视频(m3u8→yt-dlp+ffmpeg 转 mp4,source_type=video)和图集(套图详情页几百~上千张图逐张下载上传,source_type=image/content_type=article/回传 images[],主站对 images 自动跑图片 OCR 审核)——按 content_url 路径(/videos/·/video/ ↔ video,/photos/·/photo/ ↔ photo)或载荷 media_kind 自动选 scraper。当前内置 xchina.co 的视频 scraper(core/scraper.py)+ 图集 scraper(core/scraper_photos.py);加新来源就再加一个 scraper 模块。

前身是 Tkinter GUI 下载器,已删除。无单元测试。详细设计见 PRD-minio-upload.md。

结构

• core/browser.py —— browser_context(headless) async ctxmgr,launch_persistent_context(资料目录 BROWSER_PROFILE_DIR)。HEADLESS=0 + BROWSER_CHANNEL(x86_64 可设 chrome)用真实有头浏览器过 Cloudflare;容器里靠 docker-entrypoint.sh 起 Xvfb。
• core/scraper.py —— xchina 视频 scraper。XChinaScraper(context, is_cancelled, log),run(series_url,start,end) -> ScrapeResult;is_detail_url、_page_url_candidates、_collect_video_links(.item.video + 正则兜底)、scrape_detail(_extract_title / _find_cover_url(截屏tab→og:image/JSON-LD/poster→最大图)/ _find_video_url(扫 page.content() 的 .m3u8 + 网络嗅探))、_handle_interstitial_ad(广告中转页)、_assert_not_blocked(Cloudflare 质询页等其自动放行 ~24s 否则抛)。ScrapeCancelled 取消信号。导航/反爬 helper(_visit/_assert_not_blocked/_looks_blocked/_page_url_candidates/_check_cancel)被 scraper_photos.py 继承复用。
• core/scraper_photos.py —— xchina 图集 scraper。XChinaPhotoScraper(context, is_cancelled, log, max_images=100) 继承 XChinaScraper;run 同签名,is_photo_series_url/is_photo_detail_url/_canon_detail_url(/photo/id-X/3.html→/photo/id-X.html) 模块函数。流程:系列页 .item.photo 收 /photo/id- 链接(带 title hint)→ scrape_photo_detail:① _photo_bg_images(读 div.img[role=img] 的 background-image URL)② _parse_img_template(推 .../{detailId}/{NNNN}{_WxH}.{ext} 的 .format(n=) 模板,实测 xchina 是 https://img.xchina.io/photos2/{id}/{n:04d}_600x0.webp,旧套图前缀可能是 photos/ 不带 2 → 不硬编码)③ 总张数 = (\d+)P 文本 ∪ .pager 最大页码×14 ∪ 第 1 页渲染数 ④ images = [模板.format(n=1..min(total,max_images))],head_ok 校验末张;不通过 → _scrape_all_pages_images 逐页加载 /photo/id-{id}/{N}.html 收 bg-image ⑤ _extract_photo_meta(模特/系列/日期/标签,系列名优先取面包屑/.subs 里的避免误命中侧边栏)→ _compose_content(标题 + 模特/系列/日期/共N张/标签;保证非空)。返回 ScrapedItem(media_kind="photo", images=[...], image_total, image_capped, ...)。
• core/media.py —— MinioConfig(from_payload 从 invoke 载荷读 minio_endpoint/access_key/secret_key/bucket,任一缺则 is_complete=False;载荷找不到从 MINIO_* env 兜底;safe_repr 不露 secret)、download_video(yt-dlp 并发分片 + ffmpeg 合并 → <id>.mp4,max_mb 超限抛)、download_cover/download_image(requests 流式;失败返回 None,不致命)、head_ok(HEAD 探活,405 退 GET)、upload_to_minio(minio.Minio.fput_object → 返回 endpoint/bucket/key)、extract_video_id(=extract_detail_id;从 /(video|videos|photo|photos)/id-(\w+)(/\d+)?\.html)。
• core/models.py —— ScrapedItem:media_kind(video|photo,默认 video)+ 视频字段(video_url)+ 图集字段(images: list[str]/image_total/image_capped)。to_ingest_dict() 按 media_kind 分流:video → source_type:"video"/content_type:"video"/video_url;photo → source_type:"image"/content_type:"article"/images:[...]/video_url:None(content_type 是 article 因为主站 ContentTypeEnum 只有 article/video)。extra_data 里 video 放 original_m3u8/original_cover/minio_object_video/minio_object_cover/video_bytes;photo 放 image_total/image_count/image_capped/base_prefix/detail_id/img_template/minio_objects_images/images_uploaded/images_failed/original_images_count/original_cover;两者都可能有 upload_skipped_reason。
• server/jobs.py —— 内存级 job 表 + asyncio.create_task 后台跑;_RUN_LOCK 串行化(单并发)。Job 加 media_kind/max_images。_run:① _resolve_media_kind(载荷给了用载荷,否则按 URL 判)+ _make_scraper(photo→XChinaPhotoScraper,否则 XChinaScraper)→ 浏览器开着 scraper.run() → 关浏览器;② 若 minio_cfg.is_complete 则 _process_uploads(按 item.media_kind 调 _process_video_item:yt-dlp 下视频+封面上传;或 _process_photo_item:逐张 download_image→upload_to_minio→ images[] 换成 MinIO URL,边下边传边删不囤盘,单张失败保留原图床地址,封面=已上传第一张)③ 临时目录 /tmp/content-harvester/<jobid>/ 用完即删。MAX_VIDEO_MB/CONCURRENT_FRAGMENTS env。不要把 secret_key 传进 _log。
• server/app.py —— FastAPI。POST /agents/{id}/invoke(body=主站调用载荷:content_url/platform_name/source_kind/start_page/end_page/media_kind?/max_images? + minio_*)解析 media_kind(载荷 → URL 路径 → agent_id 兜底)与 max_images(默认 DEFAULT_MAX_IMAGES=100,clamp 到 MAX_IMAGES_CAP=2000),MinioConfig.from_payload(payload),→ jobs.submit(...media_kind, max_images);返回 {"task_id","status"}。GET /tasks/{id}、POST /tasks/{id}/cancel、GET /healthz。Bearer 鉴权 HARVESTER_TOKEN(空则不校验)、SSRF 白名单 ALLOWED_HOSTS(默认 xchina.co,图集视频同域)。
• Dockerfile —— base mcr.microsoft.com/playwright/python;apt 装 xvfb(有头)+ffmpeg(yt-dlp 合并);HEADLESS=0;挂卷 /data/browser_profile;ENTRYPOINT=docker-entrypoint.sh(起 Xvfb)→ uvicorn。x86_64 上可改 playwright install chromium chrome + BROWSER_CHANNEL=chrome。
• requirements.txt —— playwright / fastapi / uvicorn / requests / yt-dlp / minio。

对接契约(主站 lobster 执行器)

主站 app/services/lobster_client.py::call_lobster_agent:POST {api_url}/agents/{agent_id}/invoke(body=payload)→ 取 task_id → 轮询 GET {api_url}/tasks/{task_id} 直到 status∈{succeeded,failed} → 用 body["output"]/body["error"]。本服务 output 成功时 = {"resources":[ResourceIngestItem字段...], "pages_scanned":n, "item_errors":[...]},resources[i] 是视频条目(source_type=video,video_url/cover_image 上传成功即对象存储地址、原始签名 URL 进 extra_data.original_m3u8)或图集条目(source_type=image/content_type=article,images[] 上传成功即对象存储 URL 数组、cover_image=已上传第一张);上传失败则相应地址保留原始来源 + extra_data.upload_skipped_reason。主站对 images[] 数组会自动逐张排图片 OCR 审核(命中 ad/minor/non_frontal 剔除;一条资源 content/images/video 全空 → soft delete)。

主站侧配套改动(agent-system-backend/app/services/task_service.py):① 非批次 lobster 改 asyncio.create_task(_run_lobster_task_bg(...)) 后台跑;② _run_lobster_task_bg 成功后 _maybe_ingest_to_resource_pool;③ _maybe_ingest_to_resource_pool 兼容从 result["output"]["resources"] 取;④ _dispatch_lobster 把 _resolve_skill_variables 解出的 Skill 变量并入 payload(让远端拿到 minio_*);⑤ crawl 类用 CRAWL_LOBSTER_TIMEOUT(env,默认 7200s)。配置侧:给那个采集 Skill 绑定 minio_endpoint/minio_access_key/minio_secret_key/minio_bucket 4 个 Skill 变量(实际值由运维在主站配置,本服务不持有任何密钥)。

跑

python3 -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt && python -m playwright install chromium
# 本地无 Xvfb 可设 HEADLESS=1(但那样过不了 Cloudflare 详情页);minio 配置走 payload 传
HEADLESS=1 uvicorn server.app:app --reload   # :8000

Docker:docker build -t content-harvester . && docker run -d -p 8000:8000 -v harvester_profile:/data/browser_profile -e HARVESTER_TOKEN=xxx --memory=2g --cpus=2 content-harvester

坑

• 详情页被 Cloudflare 挡:必须有头浏览器(HEADLESS=0 + Xvfb,或真实 Chrome)才能自动过 JS 质询;无头/headless-shell 过不了。站点策略变了可能又挂 → 退路是预置一份过了验证的 browser_profile 卷。
• 媒体地址(m3u8)常带 expires 签名(~1h 失效)→ 所以要在采集时就转 mp4 上传对象存储;不传的话资源库里那个 URL 一会儿就 403。
• 下载+上传很慢:每条 15s 抓 + 几十秒几分钟下载 + 上传;一页 11 条估 30min~1h+。主站对 crawl lobster 任务给了 7200s 轮询超时。
• 单并发是有意的;并发跑多个浏览器作业容易触发风控。
• 站点结构变了改对应的 core/scraper*.py 的选择器/正则/JS 片段;媒体下载/上传逻辑改 core/media.py。
• secret_key 等敏感值别打日志(_log / logger 都注意)。