"""正文提取模块的数据模型。 核心:Article 是 M2 与下游(去重 / LLM / Embedding)的唯一交换格式。 """ from __future__ import annotations from datetime import datetime from pydantic import BaseModel, Field class Article(BaseModel): """提取后的标准化文章。 本模型是 M2 输出 / M3+ 输入的唯一格式。 """ # ---- 标识 ---- source_id: str = Field(..., description="M1 sources.yaml 中的 source id") url: str = Field(..., description="文章 URL") url_hash: str = Field(..., description="URL 的 SHA1 前 16 位,用作主键") # ---- 正文字段 ---- title: str = Field(..., min_length=1, description="文章标题") content: str = Field(..., min_length=1, description="清理后的正文纯文本") author: str | None = Field(default=None, description="作者(可选)") source_name: str | None = Field(default=None, description="网站中文名") # ---- 时间 ---- publish_time: datetime | None = Field( default=None, description="标准化后的发布时间(本地时区或 naive)" ) publish_time_raw: str | None = Field( default=None, description="原始时间字符串,便于人工核对" ) extracted_at: datetime = Field(default_factory=datetime.now, description="提取时刻") # ---- 辅助 ---- images: list[str] = Field(default_factory=list, description="正文中的图片 URL") word_count: int = Field(default=0, ge=0, description="正文中文字符数") item_type: str | None = Field( default=None, description="cninfo 数据类型: announcement|research|irm, 新闻源为 None" ) def short_summary(self) -> str: """单行摘要,用于日志。""" t = self.publish_time.strftime("%Y-%m-%d") if self.publish_time else "??" return f"[{self.source_id}] {t} 《{self.title[:40]}》 {self.word_count}字" class ExtractError(Exception): """提取失败时抛出。""" def __init__(self, reason: str, url: str | None = None) -> None: super().__init__(reason) self.reason = reason self.url = url