John Doe
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>
77 lines
3.4 KiB
Markdown
77 lines
3.4 KiB
Markdown
# 采集管理 (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/image` 9 项的命中比例。
|
||
- **源权重**：贡献该商品的源中最高 `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`。
|
||
|
||
```bash
|
||
# 增量更新 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`。
|
||
|
||
## 测试
|
||
|
||
```bash
|
||
cd ingestion && pip install -e ".[dev]"
|
||
ruff check . && ruff format --check . && pytest -q
|
||
```
|
||
|
||
纯函数测试（质量/冲突/增量分页）始终运行；依赖库的测试（水位/质量落库/去重/GS1 补全）
|
||
在无数据库或未应用 M4 迁移时自动跳过。
|