59 lines
2.1 KiB
Python
59 lines
2.1 KiB
Python
"""正文提取模块的数据模型。
|
|
|
|
核心: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
|