162 lines
5.6 KiB
Python
162 lines
5.6 KiB
Python
"""新闻抓取模块的数据模型与配置模型。
|
|
|
|
所有核心对象统一使用 Pydantic 定义(CLAUDE.md 第十条要求)。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from enum import StrEnum
|
|
|
|
from pydantic import BaseModel, Field, HttpUrl, field_validator
|
|
|
|
|
|
class CrawlStage(StrEnum):
|
|
"""抓取阶段枚举,区分列表页和文章页。"""
|
|
|
|
LIST = "list" # 列表/首页(用于发现文章链接)
|
|
ARTICLE = "article" # 文章详情页
|
|
|
|
|
|
class SourceConfig(BaseModel):
|
|
"""单个新闻源的配置。
|
|
|
|
映射 configs/sources.yaml 中 sources 列表的每一项。
|
|
"""
|
|
|
|
id: str = Field(..., description="新闻源短 ID,用作目录名,仅小写字母与下划线")
|
|
name: str = Field(..., description="中文名,用于日志展示")
|
|
enabled: bool = Field(default=True, description="是否启用")
|
|
homepage: HttpUrl = Field(..., description="新闻列表页/首页 URL")
|
|
extra_homepages: list[HttpUrl] = Field(
|
|
default_factory=list,
|
|
description="额外的列表页 URL(同站多频道时使用),共用同一个 article_url_pattern",
|
|
)
|
|
|
|
def all_homepages(self) -> list[str]:
|
|
"""返回所有入口 URL(主入口 + 额外入口)。"""
|
|
urls = [str(self.homepage)]
|
|
urls.extend(str(u) for u in self.extra_homepages)
|
|
return urls
|
|
|
|
# 链接发现规则
|
|
article_url_pattern: str = Field(
|
|
...,
|
|
description="正则表达式,从首页所有链接中筛选出文章 URL",
|
|
)
|
|
article_link_selector: str | None = Field(
|
|
default=None,
|
|
description="可选 CSS 选择器,优先在选择器内查找文章链接",
|
|
)
|
|
|
|
# JS 渲染相关
|
|
js_render: bool = Field(
|
|
default=True,
|
|
description="是否需要 JS 渲染(财联社/东方财富等动态站点必须为 true)",
|
|
)
|
|
wait_for: str | None = Field(
|
|
default=None,
|
|
description="可选 CSS,等待该元素出现后再抓取(JS 渲染时使用)",
|
|
)
|
|
page_timeout_ms: int = Field(default=30_000, ge=1_000, description="页面加载超时(毫秒)")
|
|
|
|
# GNE 提取提示(可选)
|
|
body_xpath: str | None = Field(
|
|
default=None,
|
|
description="GNE body_xpath 参数,强制指定正文容器 XPath",
|
|
)
|
|
|
|
# 抓取数量限制
|
|
max_articles_per_run: int = Field(
|
|
default=20,
|
|
ge=1,
|
|
le=200,
|
|
description="单次运行最多抓取的文章数",
|
|
)
|
|
|
|
@field_validator("id")
|
|
@classmethod
|
|
def _validate_id(cls, v: str) -> str:
|
|
if not v.replace("_", "").isalnum() or not v.islower():
|
|
raise ValueError("source.id 必须为小写字母/数字/下划线")
|
|
return v
|
|
|
|
|
|
class CrawlerSettings(BaseModel):
|
|
"""全局抓取设置。"""
|
|
|
|
concurrency: int = Field(default=3, ge=1, le=20, description="并发抓取上限")
|
|
retry_max_attempts: int = Field(default=3, ge=1, le=10, description="单 URL 最大重试次数")
|
|
retry_min_wait_sec: float = Field(default=1.0, ge=0.0, description="重试最小等待秒")
|
|
retry_max_wait_sec: float = Field(default=10.0, ge=0.0, description="重试最大等待秒")
|
|
user_agent: str = Field(
|
|
default=(
|
|
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
|
|
"(KHTML, like Gecko) Chrome/124.0 Safari/537.36"
|
|
),
|
|
description="浏览器 User-Agent",
|
|
)
|
|
headless: bool = Field(default=True, description="浏览器是否无头")
|
|
output_root: str = Field(default="data/raw", description="抓取结果根目录")
|
|
|
|
|
|
class CrawlerConfig(BaseModel):
|
|
"""整个抓取系统的配置(对应 sources.yaml 顶层)。"""
|
|
|
|
settings: CrawlerSettings = Field(default_factory=CrawlerSettings)
|
|
sources: list[SourceConfig]
|
|
|
|
def enabled_sources(self) -> list[SourceConfig]:
|
|
"""返回启用的源,按 id 排序。"""
|
|
return sorted([s for s in self.sources if s.enabled], key=lambda s: s.id)
|
|
|
|
|
|
class CninfoItem(BaseModel):
|
|
"""cninfo 公告/调研/互动易数据条目。
|
|
|
|
一条记录对应一条公告、一次调研活动或一组互动问答。
|
|
"""
|
|
|
|
stock_code: str = Field(..., description="股票代码,6 位数字")
|
|
stock_name: str = Field(..., description="公司简称")
|
|
title: str = Field(..., description="标题")
|
|
content: str = Field(default="", description="正文/摘要内容")
|
|
publish_time: str = Field(default="", description="发布时间,格式 YYYY-MM-DD")
|
|
item_type: str = Field(
|
|
default="announcement",
|
|
description="数据类型: announcement(公告)|research(调研)|irm(互动易)",
|
|
)
|
|
url: str = Field(default="", description="详情页 URL 或 PDF 链接")
|
|
ann_id: str = Field(default="", description="公告唯一 ID,用于增量去重")
|
|
extra: dict = Field(default_factory=dict, description="附加字段(公告类型等)")
|
|
|
|
|
|
class ArticleLink(BaseModel):
|
|
"""从列表页发现的文章链接。"""
|
|
|
|
source_id: str
|
|
url: str
|
|
anchor_text: str | None = None
|
|
|
|
|
|
class CrawlResult(BaseModel):
|
|
"""单次抓取结果(列表页或文章页通用)。"""
|
|
|
|
source_id: str
|
|
stage: CrawlStage
|
|
url: str
|
|
success: bool
|
|
status_code: int | None = None
|
|
title: str | None = None
|
|
html: str = ""
|
|
markdown: str = ""
|
|
error: str | None = None
|
|
fetched_at: datetime = Field(default_factory=datetime.now)
|
|
attempts: int = 1
|
|
|
|
def short_summary(self) -> str:
|
|
"""生成单行摘要,便于日志输出。"""
|
|
flag = "OK" if self.success else "FAIL"
|
|
size = len(self.html)
|
|
return f"[{flag}] {self.source_id} {self.stage.value} {self.url} html={size}B"
|