huyue
497fae99c6
FastAPI + Playwright(headed Chrome/xvfb 过 Cloudflare)抓内容详情页元数据; 可选把 m3u8 视频转 mp4 + 封面上传对象存储(配置由调用方下发,本服务不持密钥); 走主站 lobster 执行器契约(invoke/poll),回传 ResourceIngestItem 形式的 resources。 当前内置 xchina.co 的 scraper(core/scraper.py)。
63 lines
2.2 KiB
Python
63 lines
2.2 KiB
Python
"""采集结果的数据结构。
|
|
|
|
ScrapedItem 的字段刻意与主站 `ResourceIngestItem`(app/schemas/resource_pool.py)
|
|
对齐,这样主站 `_maybe_ingest_to_resource_pool` 能直接 `model_validate` 每一条。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
|
|
@dataclass
|
|
class ScrapedItem:
|
|
# 详情页 URL —— 主站资源库按它去重
|
|
source_url: str
|
|
title: str
|
|
# 来源站点名,默认 "xchina"(可被 invoke 载荷里的 platform_name 覆盖)
|
|
source_name: str = "xchina"
|
|
# 封面图 URL(上传 MinIO 后改为 MinIO 地址,否则是 xchina 图床地址)
|
|
cover_image: str | None = None
|
|
# 视频 URL(上传 MinIO 后改为 MinIO 地址,否则是 xchina 限时签名 m3u8)
|
|
video_url: str | None = None
|
|
# 列表页页码(单详情页直采时为 1)
|
|
page_no: int = 1
|
|
# xchina 详情页里的 id 段;作 MinIO object key 前缀
|
|
video_id: str = ""
|
|
# 附加进 extra_data 的字段(original_m3u8 / minio_object_video / video_bytes / upload_skipped_reason ...)
|
|
extra: dict[str, Any] = field(default_factory=dict)
|
|
|
|
def to_ingest_dict(self) -> dict[str, Any]:
|
|
extra_data: dict[str, Any] = {
|
|
"referer": self.source_url,
|
|
"page_no": self.page_no,
|
|
"collector": "xchina",
|
|
}
|
|
if self.video_id:
|
|
extra_data["video_id"] = self.video_id
|
|
extra_data.update(self.extra or {})
|
|
return {
|
|
"source_type": "video",
|
|
"source_name": self.source_name,
|
|
"source_url": self.source_url,
|
|
"content_type": "video",
|
|
"title": self.title,
|
|
"cover_image": self.cover_image,
|
|
"video_url": self.video_url,
|
|
"extra_data": extra_data,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class ScrapeResult:
|
|
resources: list[ScrapedItem] = field(default_factory=list)
|
|
pages_scanned: int = 0
|
|
item_errors: list[dict[str, str]] = field(default_factory=list)
|
|
|
|
def to_output(self) -> dict[str, Any]:
|
|
return {
|
|
"resources": [it.to_ingest_dict() for it in self.resources],
|
|
"pages_scanned": self.pages_scanned,
|
|
"item_errors": self.item_errors,
|
|
}
|