simon
65ead54b4f
- 删除 5 个过时/残留文档(project_plan/agent_prompt/optimization_plan/report_db_design/deploy/README) - 新建 docs/architecture.md(项目架构:11 包职责+数据模型+配置+产物) - 重写 docs/user-guide.md(CLI 全量+增量/断点续跑+MCP+FAQ) - 重写 README.md(精简入口+文档索引) - 更新 continuation.md(追加本次记录) - 更新 .gitignore(排除 data/* 运行产物)
91 lines
3.1 KiB
Python
91 lines
3.1 KiB
Python
"""三层去重模块的数据模型。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from enum import StrEnum
|
|
from typing import Literal, Self
|
|
|
|
from pydantic import BaseModel, Field, model_validator
|
|
|
|
|
|
class DedupLayer(StrEnum):
|
|
"""命中去重的层。"""
|
|
|
|
URL = "url" # L1: 完全相同 URL
|
|
CONTENT = "content" # L2: 标准化后 content 完全一致
|
|
SIMHASH = "simhash" # L3: SimHash 汉明距离 <= 阈值
|
|
|
|
|
|
class Fingerprint(BaseModel):
|
|
"""单篇文章的指纹记录,持久化到 SQLite。
|
|
|
|
source_ids: 同一内容组(去重后视为同一篇新闻)的全部来源列表,
|
|
第一位是主源(即本指纹的 source_id);重复文章命中时由
|
|
Deduper.ingest 自动合并,实现「一条唯一新闻记录多个源」。
|
|
"""
|
|
|
|
url_hash: str = Field(..., description="主键,与 Article.url_hash 一致")
|
|
content_hash: str = Field(..., description="标准化 content 的 SHA1[:16]")
|
|
simhash: int = Field(..., description="64 位 SimHash 整数(无符号)")
|
|
source_id: str
|
|
url: str
|
|
title: str
|
|
publish_date: str | None = Field(default=None, description="YYYY-MM-DD,用于时间窗口")
|
|
ingested_at: datetime = Field(default_factory=datetime.now)
|
|
source_ids: list[str] = Field(
|
|
default_factory=list,
|
|
description="同内容组全部来源(去重合并),始终包含 source_id 且其居首",
|
|
)
|
|
|
|
@model_validator(mode="after")
|
|
def _ensure_source_ids(self) -> Self:
|
|
"""保证 source_ids 非空、去重且以主源 source_id 开头。"""
|
|
seen: list[str] = []
|
|
for s in [self.source_id, *self.source_ids]:
|
|
if s and s not in seen:
|
|
seen.append(s)
|
|
self.source_ids = seen
|
|
return self
|
|
|
|
|
|
class DedupResult(BaseModel):
|
|
"""对单篇文章的判重结果。"""
|
|
|
|
url_hash: str
|
|
is_duplicate: bool
|
|
matched_layer: DedupLayer | None = None
|
|
matched_url_hash: str | None = None
|
|
matched_url: str | None = None
|
|
matched_title: str | None = None
|
|
matched_source_id: str | None = Field(
|
|
default=None, description="匹配指纹的主源 source_id"
|
|
)
|
|
all_source_ids: list[str] = Field(
|
|
default_factory=list,
|
|
description="该内容组(唯一新闻)的全部来源;含匹配指纹自身的来源",
|
|
)
|
|
hamming_distance: int | None = Field(
|
|
default=None, description="仅 SimHash 层有值"
|
|
)
|
|
|
|
def short_summary(self) -> str:
|
|
if not self.is_duplicate:
|
|
return f"[UNIQUE] {self.url_hash}"
|
|
layer = self.matched_layer.value if self.matched_layer else "?"
|
|
extra = f" hd={self.hamming_distance}" if self.hamming_distance is not None else ""
|
|
return f"[DUP/{layer}] {self.url_hash} ~ {self.matched_url_hash}{extra}"
|
|
|
|
|
|
class DedupStats(BaseModel):
|
|
"""指纹库统计。"""
|
|
|
|
total: int = 0
|
|
by_source: dict[str, int] = Field(default_factory=dict)
|
|
earliest: str | None = None
|
|
latest: str | None = None
|
|
|
|
|
|
# 类型别名,便于在批处理日志中归类
|
|
DedupVerdict = Literal["unique", "duplicate"]
|