content-harvester/core/models.py

"""采集结果的数据结构。

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"
    # 资源类型: "video"(视频) | "photo"(图集)
    media_kind: str = "video"
    # 封面图 URL(上传 MinIO 后改为 MinIO 地址,否则是 xchina 图床地址)
    cover_image: str | None = None
    # 视频 URL(media_kind=video 时;上传 MinIO 后改为 MinIO 地址,否则是 xchina 限时签名 m3u8)
    video_url: str | None = None
    # 图集图片 URL 列表(media_kind=photo 时;上传 MinIO 后逐个改为 MinIO 地址)
    images: list[str] = field(default_factory=list)
    # 图集原始总张数(images 可能因 max_images 被截断)
    image_total: int = 0
    image_capped: bool = False
    # 简介 / 正文(主站资源库要求 content 非空才入库;为空时回退用 title)
    summary: str | None = None
    content: 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
        if self.media_kind == "photo":
            extra_data["image_total"] = self.image_total or len(self.images)
            extra_data["image_count"] = len(self.images)
            extra_data["image_capped"] = bool(self.image_capped)
        extra_data.update(self.extra or {})
        # 主站 ingest 对空 content 直接拒收 —— 兜底用 title,确保非空
        content = (self.content or "").strip() or (self.title or "").strip() or self.source_url
        if self.media_kind == "photo":
            return {
                # 主站 SourceTypeEnum: image;ContentTypeEnum 只有 article/video,图集归 article
                "source_type": "image",
                "source_name": self.source_name,
                "source_url": self.source_url,
                "content_type": "article",
                "title": self.title,
                "cover_image": self.cover_image or (self.images[0] if self.images else None),
                "summary": (self.summary or None),
                "content": content,
                "images": list(self.images),
                "video_url": None,
                "extra_data": extra_data,
            }
        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,
            "summary": (self.summary or None),
            "content": content,
            "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,
        }