simon/news
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 大模型使用场景化配置与去重多源记录

Browse Source 
- 新增 configs/llm_models.yaml: 4 个场景(event_extraction/daily_report/stock_report/embedding)
  可独立配置 provider/model/api_key_env/base_url_env/temperature 等,含用途与模型要求说明
- 新增 configs/loader.py: YAML 场景加载器(优先级: CLI 参数 > YAML > .env > 内置默认)
- llm/client.py: load_llm_config 支持 scene 参数,LLMConfig 增加 max_attempts
- embedding/factory+remote+local: provider/model/batch_limit 支持场景覆盖
- scheduler/reporter+stock_reporter: 日报/个股摘要接入场景配置
- dedup: Fingerprint.source_ids 多源记录 + 旧库自动迁移 + DedupResult 多源字段
- scripts/run_dedup: uniques JSON 的 sources 字段 + data/deduped/{day}/sources.json 汇总
- scripts/run_event_extraction: 接入 event_extraction 场景
- 补充测试: 场景优先级/零值、多源合并、旧库迁移、embedding 场景覆盖
This commit is contained in:
simon
2026-08-12 07:57:10 +08:00
parent 0c032196d2
commit 3c65701449
21 changed files with 886 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dedup/deduper.py
+33 -5
View File
@@ -35,7 +35,10 @@ def _publish_date(article: Article) -> str | None:
def article_to_fingerprint(article: Article) -> Fingerprint:
"""构造 Fingerprint(用于 ingest 写入或对外只读)。"""
"""构造 Fingerprint(用于 ingest 写入或对外只读)。
source_ids 初始为 [article.source_id],后续重复文章命中时由 ingest 合并。
"""
return Fingerprint(
url_hash=article.url_hash,
content_hash=content_hash(article.content),
@@ -45,6 +48,7 @@ def article_to_fingerprint(article: Article) -> Fingerprint:
title=article.title,
publish_date=_publish_date(article),
ingested_at=datetime.now(),
source_ids=[article.source_id],
)
@@ -80,7 +84,7 @@ class Deduper:
# ------------------------------------------------------------------ #
def check(self, article: Article) -> DedupResult:
"""三层判重(只读)。"""
"""三层判重(只读)。命中时附带匹配指纹的多源信息(all_source_ids)。"""
fp = article_to_fingerprint(article)
# L1: URL hash
@@ -93,6 +97,8 @@ class Deduper:
matched_url_hash=existing.url_hash,
matched_url=existing.url,
matched_title=existing.title,
matched_source_id=existing.source_id,
all_source_ids=existing.source_ids,
)
# L2: 内容 hash
@@ -105,6 +111,8 @@ class Deduper:
matched_url_hash=existing.url_hash,
matched_url=existing.url,
matched_title=existing.title,
matched_source_id=existing.source_id,
all_source_ids=existing.source_ids,
)
# L3: SimHash 模糊
@@ -129,20 +137,40 @@ class Deduper:
matched_url_hash=best_match.url_hash,
matched_url=best_match.url,
matched_title=best_match.title,
matched_source_id=best_match.source_id,
all_source_ids=best_match.source_ids,
hamming_distance=best_dist,
)
return DedupResult(url_hash=fp.url_hash, is_duplicate=False)
def ingest(self, article: Article) -> DedupResult:
"""判重 + 不重复则入库。"""
"""判重 + 不重复则入库。
命中重复时,把当前文章的 source_id 合并进匹配指纹的 source_ids
(记录同一内容组的全部来源),并更新 all_source_ids 后返回。
"""
result = self.check(article)
if not result.is_duplicate:
fp = article_to_fingerprint(article)
self.store.upsert(fp)
logger.debug("入库: {} {}", fp.url_hash, fp.title[:30])
else:
logger.debug("命中重复: {}", result.short_summary())
return result
# 重复:合并来源到匹配指纹(主源保持首位,Fingerprint validator 负责去重)
if result.matched_url_hash and article.source_id not in result.all_source_ids:
matched = self.store.get_by_url_hash(result.matched_url_hash)
if matched is not None:
merged = [*matched.source_ids, article.source_id]
self.store.upsert(matched.model_copy(update={"source_ids": merged}))
result = result.model_copy(
update={"all_source_ids": merged}
)
logger.debug(
"合并来源 {} -> {} ({} 个源)",
article.source_id, result.matched_url_hash, len(merged),
)
logger.debug("命中重复: {}", result.short_summary())
return result
def stats(self) -> DedupStats:
dedup/models.py
+29 -3
View File
@@ -4,9 +4,9 @@ from __future__ import annotations
from datetime import datetime
from enum import StrEnum
from typing import Literal
from typing import Literal, Self
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, model_validator
class DedupLayer(StrEnum):
@@ -18,7 +18,12 @@ class DedupLayer(StrEnum):
class Fingerprint(BaseModel):
"""单篇文章的指纹记录,持久化到 SQLite。"""
"""单篇文章的指纹记录,持久化到 SQLite。
source_ids: 同一内容组(去重后视为同一篇新闻)的全部来源列表,
第一位是主源(即本指纹的 source_id);重复文章命中时由
Deduper.ingest 自动合并,实现「一条唯一新闻记录多个源」。
"""
url_hash: str = Field(..., description="主键,与 Article.url_hash 一致")
content_hash: str = Field(..., description="标准化 content 的 SHA1[:16]")
@@ -28,6 +33,20 @@ class Fingerprint(BaseModel):
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):
@@ -39,6 +58,13 @@ class DedupResult(BaseModel):
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 层有值"
)
dedup/store.py
+40 -2
View File
@@ -3,10 +3,14 @@
注意:SimHash 是 64 位无符号整数,SQLite INTEGER 是 64 位有符号
(范围 [-2^63, 2^63-1])。直接存可能溢出/转负数,虽然 XOR 仍然
正确但语义混乱。这里统一存为 16 位 hex TEXT,避免符号问题。
source_ids 列存 JSON 数组文本(同一内容组全部来源);旧库无此列时
自动 ALTER TABLE 迁移,旧数据读取时回退为 [source_id]。
"""
from __future__ import annotations
import json
import sqlite3
from datetime import datetime, timedelta
from pathlib import Path
@@ -27,13 +31,17 @@ CREATE TABLE IF NOT EXISTS fingerprints (
url TEXT NOT NULL,
title TEXT NOT NULL,
publish_date TEXT,
ingested_at TEXT NOT NULL
ingested_at TEXT NOT NULL,
source_ids TEXT
);
CREATE INDEX IF NOT EXISTS idx_content_hash ON fingerprints(content_hash);
CREATE INDEX IF NOT EXISTS idx_publish_date ON fingerprints(publish_date);
CREATE INDEX IF NOT EXISTS idx_source_id ON fingerprints(source_id);
"""
# 兼容旧库:为已存在但缺少 source_ids 列的表补列
_ALTER_SQL = "ALTER TABLE fingerprints ADD COLUMN source_ids TEXT"
def _to_hex(simhash: int) -> str:
return f"{simhash:016x}"
@@ -43,6 +51,25 @@ def _from_hex(hex_str: str) -> int:
return int(hex_str, 16)
def _to_sources_json(source_ids: list[str]) -> str:
return json.dumps(source_ids, ensure_ascii=False)
def _from_sources_json(raw: str | None, fallback: str) -> list[str]:
"""解析 source_ids 列;NULL/损坏时回退 [主源]。"""
if not raw:
return [fallback]
try:
val = json.loads(raw)
except (TypeError, ValueError):
return [fallback]
if isinstance(val, list) and val:
# 保证主源在首位(兼容手改/旧数据)
cleaned = [s for s in val if s and s != fallback]
return [fallback, *cleaned]
return [fallback]
def _row_to_fp(row: sqlite3.Row) -> Fingerprint:
return Fingerprint(
url_hash=row["url_hash"],
@@ -53,6 +80,7 @@ def _row_to_fp(row: sqlite3.Row) -> Fingerprint:
title=row["title"],
publish_date=row["publish_date"],
ingested_at=datetime.fromisoformat(row["ingested_at"]),
source_ids=_from_sources_json(row["source_ids"], row["source_id"]),
)
@@ -67,8 +95,17 @@ class FingerprintStore:
)
self._conn.row_factory = sqlite3.Row
self._conn.executescript(_SCHEMA_SQL)
self._migrate_source_ids()
logger.debug("打开指纹库: {}", self.db_path)
def _migrate_source_ids(self) -> None:
"""旧库兼容:为缺少 source_ids 列的表补列(新库无需执行)。"""
try:
self._conn.execute(_ALTER_SQL)
logger.info("指纹库迁移:为 fingerprints 表新增 source_ids 列")
except sqlite3.OperationalError:
logger.debug("source_ids 列已存在,跳过迁移")
def close(self) -> None:
self._conn.close()
@@ -148,7 +185,7 @@ class FingerprintStore:
self._conn.execute(
"INSERT OR REPLACE INTO fingerprints "
"(url_hash, content_hash, simhash_hex, source_id, url, title, "
" publish_date, ingested_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
" publish_date, ingested_at, source_ids) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
(
fp.url_hash,
fp.content_hash,
@@ -158,6 +195,7 @@ class FingerprintStore:
fp.title,
fp.publish_date,
fp.ingested_at.isoformat(),
_to_sources_json(fp.source_ids),
),
)