restore main tree to deployed 3bafd30 (HEL-529 校准)

Revert the full revert 6f99ee9 so main's tree content is byte-identical
to 3bafd30, which is the commit currently deployed and healthy on the
8766 xiaobai-datahub container (image hel531-3bafd30). No history
rewrite, no force push; this is a normal forward commit on top of main.

Co-authored-by: multica-agent <github@multica.ai>

xiaobai-datahub/datahub/pipeline.py

@@ -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 = ?",