a35bcd6647
- OFF incremental fetch via search API + persistent watermark (ingest_state, migration 0004) - GS1 barcode supplement adapter (offline mapping + GS1-style API) filling only gaps with field-level provenance - Non-GTIN dedup with canonical selection + merge_log; field-level conflict resolution (source trust > recency) - Quality scoring (0.4 completeness + 0.3 source trust + 0.2 multi-source + 0.1 freshness) wired into load/merge - Jobs: update_off, dedup, schedule; docs/ingestion-management.md - 19 new tests (pure + DB-integration), ruff clean Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
3.4 KiB
3.4 KiB
采集管理 (M4)
M4 在 M2(Open Food Facts 首次导入)基础上,补齐"持续运营"所需的采集能力:
增量更新、第二数据源补全(GS1)、去重合并与字段级冲突解决、数据质量评分,
以及把这些串起来的定时调度。全部为 Python 侧(ingestion/),只写库、可单测。
组成
| 能力 | 模块 | 说明 |
|---|---|---|
| 增量采集 | adapters/openfoodfacts.py: fetch_modified_since() |
按 last_modified_t 拉取自上次水位后变更的商品 |
| 采集水位 | etl/state.py + ingest_state 表 |
每个源持久化 last_modified_t,只前进不回退 |
| GS1 补全 | adapters/gs1.py + etl/supplement.py |
用权威条码源补缺失字段(品牌/厂商/GPC/产地/净含量),不覆盖已有值 |
| 去重合并 | etl/dedup.py |
非 GTIN 重复(同名+品牌+净含量)合并到质量最高的主记录 |
| 冲突解决 | etl/merge.py |
多源同字段按"源权重 > 新鲜度"择优,保留字段级溯源 |
| 质量评分 | etl/quality.py |
0~1 分,落到 product.quality_score |
| 定时调度 | jobs/schedule.py |
固定周期跑"增量 + 去重"一轮,零额外依赖 |
质量评分
锁定公式(各分量均归一到 0~1):
quality = 0.4 * 完整度 + 0.3 * 源权重 + 0.2 * 多源一致 + 0.1 * 新鲜度
- 完整度:
name/gtin/brand/category/net_content/country/nutriments/ingredients/image9 项的命中比例。 - 源权重:贡献该商品的源中最高
source.trust_weight(OFF=0.7,GS1=0.9)。 - 多源一致:源数量代理——单源 0.5、两源 0.8、三源及以上 1.0(单源无法互证)。
- 新鲜度:最近一次
product_source.fetched_at的时间衰减(≤30d=1.0 … >730d=0.2)。
load_record() 与 merge_products() 写入后都会调 update_quality() 重算。
增量水位
ingest_state(迁移 0004)每源一行,记录 last_modified_t、last_run_at、stats。
set_watermark() 用 GREATEST(...) 保证水位只前进,避免乱序/中断的运行回退进度。
运行
前置:docker compose up -d postgres 且迁移已 up(含 0004)。DSN 默认读 OPENGOODS_DATABASE_URL。
# 增量更新 OFF(从持久化水位开始;--since 可覆盖)
python -m opengoods.jobs.update_off --max-pages 5
python -m opengoods.jobs.update_off --since 1700000000
# 去重合并(--dry-run 只报告不写库)
python -m opengoods.jobs.dedup --dry-run
python -m opengoods.jobs.dedup --actor nightly
# 定时调度:单轮 / 周期循环(增量 + 去重)
python -m opengoods.jobs.schedule --once
python -m opengoods.jobs.schedule --interval 3600
GS1 补全
GS1 为付费、分区域的授权数据,适配器支持两种模式:
- 离线(默认):从本地 JSON 映射
{gtin: {...}}查(GS1Adapter.from_file(path)), 供测试与内网环境使用。 - 在线:传
base_url+client(+api_key),GET {base_url}/{gtin},按 Verified-by-GS1 风格字段解析。
补全只填空缺字段并在 product_source 记字段级溯源,源标记为 gs1。
测试
cd ingestion && pip install -e ".[dev]"
ruff check . && ruff format --check . && pytest -q
纯函数测试(质量/冲突/增量分页)始终运行;依赖库的测试(水位/质量落库/去重/GS1 补全) 在无数据库或未应用 M4 迁移时自动跳过。