feat(HEL-529): 按定稿100%重做三页数据中枢 + 数据修正1-5
视觉:admin/app.js 按已确认打样 hub-kimi.html 逐行重写三页 DOM 与动效
(sparkline/缓存年龄秒增/延迟变化闪烁/EVENT TAPE 预装滚动/分组接口表/
更新频率列/时钟冒号 blink/雷达 blip),CSS 补真实调用脉冲 node-ping。
数据修正:
1) pipeline._stage 批内按暂存表业务键确定性去重(保留最后一条),
修复人气榜/龙虎榜自 09-07 起每日 UNIQUE constraint 落库失败;
2) overview.anomalies 收敛为「最新批次未成功且当日未发布」的当前异常,
历史已恢复批次留在审计明细;
3) source_catalog 接口补真实批次分组 + 观测 join(接口名或数据集名
双向匹配,标注 observed/observed_basis),消除「全部未配置/0/30」误报;
4) lineage 逐数据集按其服务接口过滤健康行(接口级状态),
provisional 已配置无观测显示「已配置 · 待观测」,仅 iFinD 为未配置;
5) lineage 补 update_freq 真实频率字段。
测试:新增 tests/test_hel529_rework.py(8 项),全套 191 项通过
(1 项环境依赖失败在基线 d9358ab 上同样复现,与本改动无关)。
Co-authored-by: multica-agent <github@multica.ai>
This commit is contained in:
@@ -99,6 +99,58 @@ STAGING_INSERT = {
|
||||
**EXTENDED_STAGING_INSERT,
|
||||
}
|
||||
|
||||
# Staging-table business keys (PRIMARY KEY minus the constant batch_id),
|
||||
# mirroring the PRIMARY KEY clauses declared in datahub/db.py and
|
||||
# datahub/datasets_ext.py. Used only to collapse within-batch duplicates so
|
||||
# a single upstream response cannot fail the whole batch on a UNIQUE
|
||||
# constraint (HEL-529: 2026-09-14 dc_hot returned 4 duplicate ts_codes and
|
||||
# hm_detail 43 duplicate (ts_code, hm_name) keys in one response, which has
|
||||
# blocked popularity/dragon_tiger publishing every day since 09-07).
|
||||
STAGING_KEY_FIELDS = {
|
||||
"stocks": ("ts_code", "trade_date"),
|
||||
"daily": ("ts_code", "trade_date"),
|
||||
"valuation": ("ts_code", "trade_date"),
|
||||
"moneyflow": ("ts_code", "trade_date"),
|
||||
"auction": ("ts_code", "trade_date"),
|
||||
"index_daily": ("ts_code", "trade_date"),
|
||||
"limit_events": ("ts_code", "trade_date", "limit_type"),
|
||||
"popularity": ("ts_code", "trade_date", "source"),
|
||||
"dragon_tiger": ("ts_code", "trade_date", "hm_name"),
|
||||
"sector_daily": ("ts_code", "trade_date", "family"),
|
||||
}
|
||||
|
||||
|
||||
def _dedupe_staging_rows(dataset: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
"""Collapse within-batch duplicates on the staging table's business key.
|
||||
|
||||
Deterministic: keeps the LAST occurrence of each key (the same row the
|
||||
eod copy's INSERT OR REPLACE would keep), preserves first-seen order, and
|
||||
never touches rows across batches. Datasets without a declared business
|
||||
key are returned unchanged.
|
||||
"""
|
||||
fields = STAGING_KEY_FIELDS.get(dataset)
|
||||
if not fields:
|
||||
return rows
|
||||
seen: dict[tuple, int] = {}
|
||||
out: list[dict[str, Any]] = []
|
||||
dropped = 0
|
||||
for row in rows:
|
||||
key = tuple(row.get(field) for field in fields)
|
||||
if key in seen:
|
||||
out[seen[key]] = row
|
||||
dropped += 1
|
||||
else:
|
||||
seen[key] = len(out)
|
||||
out.append(row)
|
||||
if dropped:
|
||||
LOGGER.info(
|
||||
"staging dedupe: %s collapsed %d duplicate rows within batch (kept last)",
|
||||
dataset,
|
||||
dropped,
|
||||
extra={"hub": {"dataset": dataset, "deduped": dropped}},
|
||||
)
|
||||
return out
|
||||
|
||||
EOD_COPY = {
|
||||
"stocks": (
|
||||
"INSERT OR REPLACE INTO eod_stocks "
|
||||
@@ -1720,6 +1772,7 @@ class Pipeline:
|
||||
|
||||
def _stage(self, dataset: str, batch_id: str, rows: list[dict[str, Any]]) -> None:
|
||||
sql, mapper = STAGING_INSERT[dataset]
|
||||
rows = _dedupe_staging_rows(dataset, rows)
|
||||
with self.db.write() as connection:
|
||||
connection.execute(
|
||||
f"DELETE FROM {DATASET_TABLES[dataset][1]} WHERE batch_id = ?",
|
||||
|
||||
Reference in New Issue
Block a user