feat(content): 自营三平台指标入库管线完善

sync_self_video_metrics 按平台隔离入库与字段口径调整，配套自营指标验收测试
更新。

src/gyxx_flow/modules/content_marketing/data/tools/sync_self_video_metrics.py

@@ -426,6 +426,21 @@ def ensure_table(cur) -> None:
         )
 
 
+def drop_stale_rows(
+    all_rows: dict[tuple, tuple], stale_paths: list[Path]
+) -> int:
+    """删除来源文件已陈旧的行,返回删除数(按行内 source_file 文件名匹配)。"""
+    stale_names = {Path(path).name for path in stale_paths}
+    source_index = ROW_FIELDS.index("source_file")
+    stale_keys = [
+        key for key, row in all_rows.items()
+        if str(row[source_index]) in stale_names
+    ]
+    for key in stale_keys:
+        del all_rows[key]
+    return len(stale_keys)
+
+
 def main() -> int:
     parser = argparse.ArgumentParser(
         description="同步自营内容(抖音/B站/小红书)全量指标到 cmt_self_video_metrics"
@@ -504,20 +519,25 @@ def main() -> int:
         print("❌ 没有任何平台的来源文件可用")
         return 1
     stale = [
-        str(path)
+        path
         for path in sources
         if business_date_bound
         and not args.allow_stale
         and not args.use_file_mtime
         and not is_current_artifact(path, metric_date)
     ]
-    if stale and not args.allow_stale:
+    if stale:
+        # 陈旧防护按来源文件粒度跳过: 某平台来源过期只丢弃该平台的行,
+        # 不拖累其他平台当天的新鲜数据(如抖音导出失败时不拦 B站/小红书)。
+        dropped = drop_stale_rows(all_rows, stale)
         print(
-            f"❌ 以下来源文件不是业务日期({metric_date.isoformat()})的产物; "
+            f"⚠️ 以下来源文件不是业务日期({metric_date.isoformat()})的产物, "
-            "补历史请显式传 --metric-date 并加 --allow-stale:\n  "
+            f"已跳过对应 {dropped} 行(补历史用 --metric-date + "
-            + "\n  ".join(stale)
+            "--allow-stale):\n  " + "\n  ".join(str(p) for p in stale)
         )
-        return 1
+        if not all_rows:
+            print("❌ 跳过陈旧来源后没有可入库的内容指标行")
+            return 1
 
     print(
         f"[sync-self] 平台={'+'.join(selected_platforms)} 待写={len(all_rows)} 行; "
@@ -543,74 +563,98 @@ def main() -> int:
     from gyxx_flow.modules.content_marketing.data.tools.db import get_db_config
 
     with psycopg.connect(**get_db_config()) as conn:
-        with conn.cursor() as cur:
+        written, failed = upsert_rows(conn, list(all_rows.values()))
-            ensure_table(cur)
+    if failed:
-            for row in all_rows.values():
+        print(f"[sync-self] 完成: 写入 {written} 行,单行失败跳过 {failed} 行 "
-                # SQL 为静态字面量,29 个值全部经 %s 占位符参数绑定传入。
+              f"(见上方 ⚠️ 日志)")
-                cur.execute(
+        return 0 if written else 1
-                    """
+    print(f"[sync-self] 完成: 已幂等写入 {written} 行到 {TABLE_NAME}")
-                    INSERT INTO cmt_self_video_metrics (
-                        platform, account_alias, account_name,
-                        platform_account_id, video_id, title, chanmama_url,
-                        video_url, duration, publish_time, metric_date,
-                        exposure, like_count, comment_count, share_count,
-                        est_sales_count, est_sales_amount, note_type,
-                        creation_method, publish_status, mounted_products,
-                        homepage_redirect_count, live_redirect_count,
-                        payment_amount, payment_orders, coin_count,
-                        favorite_count, danmaku_count, page_metrics,
-                        source_file,
-                        observed_at, created_at, updated_at
-                    ) VALUES (
-                        %s, %s, %s,
-                        %s, %s, %s, %s,
-                        %s, %s, %s, %s,
-                        %s, %s, %s, %s,
-                        %s, %s, %s,
-                        %s, %s, %s,
-                        %s, %s,
-                        %s, %s, %s,
-                        %s, %s, %s,
-                        %s,
-                        CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
-                    )
-                    ON CONFLICT (platform, account_name, video_id, metric_date)
-                    DO UPDATE SET
-                        account_alias = EXCLUDED.account_alias,
-                        platform_account_id = EXCLUDED.platform_account_id,
-                        title = EXCLUDED.title,
-                        chanmama_url = EXCLUDED.chanmama_url,
-                        video_url = EXCLUDED.video_url,
-                        duration = EXCLUDED.duration,
-                        publish_time = EXCLUDED.publish_time,
-                        exposure = EXCLUDED.exposure,
-                        like_count = EXCLUDED.like_count,
-                        comment_count = EXCLUDED.comment_count,
-                        share_count = EXCLUDED.share_count,
-                        est_sales_count = EXCLUDED.est_sales_count,
-                        est_sales_amount = EXCLUDED.est_sales_amount,
-                        note_type = EXCLUDED.note_type,
-                        creation_method = EXCLUDED.creation_method,
-                        publish_status = EXCLUDED.publish_status,
-                        mounted_products = EXCLUDED.mounted_products,
-                        homepage_redirect_count = EXCLUDED.homepage_redirect_count,
-                        live_redirect_count = EXCLUDED.live_redirect_count,
-                        payment_amount = EXCLUDED.payment_amount,
-                        payment_orders = EXCLUDED.payment_orders,
-                        coin_count = EXCLUDED.coin_count,
-                        favorite_count = EXCLUDED.favorite_count,
-                        danmaku_count = EXCLUDED.danmaku_count,
-                        page_metrics = EXCLUDED.page_metrics,
-                        source_file = EXCLUDED.source_file,
-                        observed_at = CURRENT_TIMESTAMP,
-                        updated_at = CURRENT_TIMESTAMP
-                    """,
-                    row,
-                )
-        conn.commit()
-    print(f"[sync-self] 完成: 已幂等写入 {len(all_rows)} 行到 {TABLE_NAME}")
     return 0
 
 
+def upsert_rows(conn, rows: list[tuple]) -> tuple[int, int]:
+    """逐行 upsert,返回 (成功数, 失败数)。
+
+    每行放在独立保存点里执行: 单行失败(如超长值违反约束)只回滚该行,
+    不拖累其他平台/其他内容的写入——这是「一个平台失败不影响全库」的
+    最后一道隔离。
+    """
+    written = 0
+    failed = 0
+    with conn.cursor() as cur:
+        ensure_table(cur)
+        for row in rows:
+            try:
+                with conn.transaction():
+                    # SQL 为静态字面量,29 个值全部经 %s 占位符参数绑定传入。
+                    cur.execute(
+                        """
+                        INSERT INTO cmt_self_video_metrics (
+                            platform, account_alias, account_name,
+                            platform_account_id, video_id, title, chanmama_url,
+                            video_url, duration, publish_time, metric_date,
+                            exposure, like_count, comment_count, share_count,
+                            est_sales_count, est_sales_amount, note_type,
+                            creation_method, publish_status, mounted_products,
+                            homepage_redirect_count, live_redirect_count,
+                            payment_amount, payment_orders, coin_count,
+                            favorite_count, danmaku_count, page_metrics,
+                            source_file,
+                            observed_at, created_at, updated_at
+                        ) VALUES (
+                            %s, %s, %s,
+                            %s, %s, %s, %s,
+                            %s, %s, %s, %s,
+                            %s, %s, %s, %s,
+                            %s, %s, %s,
+                            %s, %s, %s,
+                            %s, %s,
+                            %s, %s, %s,
+                            %s, %s, %s,
+                            %s,
+                            CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
+                        )
+                        ON CONFLICT (platform, account_name, video_id, metric_date)
+                        DO UPDATE SET
+                            account_alias = EXCLUDED.account_alias,
+                            platform_account_id = EXCLUDED.platform_account_id,
+                            title = EXCLUDED.title,
+                            chanmama_url = EXCLUDED.chanmama_url,
+                            video_url = EXCLUDED.video_url,
+                            duration = EXCLUDED.duration,
+                            publish_time = EXCLUDED.publish_time,
+                            exposure = EXCLUDED.exposure,
+                            like_count = EXCLUDED.like_count,
+                            comment_count = EXCLUDED.comment_count,
+                            share_count = EXCLUDED.share_count,
+                            est_sales_count = EXCLUDED.est_sales_count,
+                            est_sales_amount = EXCLUDED.est_sales_amount,
+                            note_type = EXCLUDED.note_type,
+                            creation_method = EXCLUDED.creation_method,
+                            publish_status = EXCLUDED.publish_status,
+                            mounted_products = EXCLUDED.mounted_products,
+                            homepage_redirect_count = EXCLUDED.homepage_redirect_count,
+                            live_redirect_count = EXCLUDED.live_redirect_count,
+                            payment_amount = EXCLUDED.payment_amount,
+                            payment_orders = EXCLUDED.payment_orders,
+                            coin_count = EXCLUDED.coin_count,
+                            favorite_count = EXCLUDED.favorite_count,
+                            danmaku_count = EXCLUDED.danmaku_count,
+                            page_metrics = EXCLUDED.page_metrics,
+                            source_file = EXCLUDED.source_file,
+                            observed_at = CURRENT_TIMESTAMP,
+                            updated_at = CURRENT_TIMESTAMP
+                        """,
+                        row,
+                    )
+            except Exception as exc:
+                failed += 1
+                print(f"  ⚠️ 单行写入失败已隔离跳过: {type(exc).__name__}: {exc}")
+            else:
+                written += 1
+    conn.commit()
+    return written, failed
+
+
 if __name__ == "__main__":
     raise SystemExit(main())

tests/modules/content_marketing/test_self_content_metrics.py

@@ -466,6 +466,83 @@ def test_collect_xiaohongshu_rows_from_artifact(tmp_path, monkeypatch):
     assert row[fields.index("page_metrics")].obj["笔记阅读数"] == "1.2万"
 
 
+def test_upsert_rows_isolates_single_row_failure():
+    """单行写入失败只跳过该行,不拖累整批(保存点隔离)。"""
+    import contextlib
+
+    class _Cursor:
+        def fetchone(self):
+            return (1,)
+
+        def __getattr__(self, name):
+            def record(*args, **kwargs):
+                params = args[1] if len(args) > 1 else kwargs.get("params")
+                if params is None or len(params) != len(sync_tool.ROW_FIELDS):
+                    return
+                # 模拟毒行: 标题以 POISON 开头的行违反约束
+                title = params[sync_tool.ROW_FIELDS.index("title")]
+                if str(title).startswith("POISON"):
+                    raise RuntimeError("value too long")
+            return record
+
+    class _Conn:
+        def __init__(self):
+            self.cursor_stub = _Cursor()
+            self.committed = False
+
+        @contextlib.contextmanager
+        def cursor(self):
+            yield self.cursor_stub
+
+        @contextlib.contextmanager
+        def transaction(self):
+            yield
+
+        def commit(self):
+            self.committed = True
+
+    metric_date = dt.date(2026, 10, 1)
+    rows = [
+        sync_tool.make_row("bilibili", metric_date, "daily.json",
+                           account_name="光影行星", video_id="BV1"),
+        sync_tool.make_row("douyin", metric_date, "a.xlsx",
+                           account_name="店铺A", video_id="1",
+                           title="POISON 这行会失败"),
+        sync_tool.make_row("xiaohongshu", metric_date, "daily.json",
+                           account_name="光影行星GYXX旗舰店", video_id="66e"),
+    ]
+    conn = _Conn()
+
+    written, failed = sync_tool.upsert_rows(conn, rows)
+
+    assert (written, failed) == (2, 1)
+    assert conn.committed
+
+
+def test_drop_stale_rows_only_removes_matching_sources():
+    """陈旧防护按来源文件跳过: 抖音旧导出只清掉抖音行,不拦 B站/小红书。"""
+    stale_excel = Path("chanmama/旧导出_20260823-20260921.xlsx")
+    rows = {
+        ("douyin", STORE_FLAGSHIP, "1", dt.date(2026, 10, 1)): sync_tool.make_row(
+            "douyin", dt.date(2026, 10, 1), stale_excel.name, video_id="1"
+        ),
+        ("bilibili", "光影行星", "BV1", dt.date(2026, 10, 1)): sync_tool.make_row(
+            "bilibili", dt.date(2026, 10, 1), "daily.json", video_id="BV1"
+        ),
+        ("xiaohongshu", XHS_SHOP, "2", dt.date(2026, 10, 1)): sync_tool.make_row(
+            "xiaohongshu", dt.date(2026, 10, 1), "daily.json", video_id="2"
+        ),
+    }
+
+    dropped = sync_tool.drop_stale_rows(rows, [stale_excel])
+
+    assert dropped == 1
+    assert set(rows) == {
+        ("bilibili", "光影行星", "BV1", dt.date(2026, 10, 1)),
+        ("xiaohongshu", XHS_SHOP, "2", dt.date(2026, 10, 1)),
+    }
+
+
 def test_douyin_history_backfill_uses_file_mtime_and_filename_account(tmp_path):
     """历史回填: 按导出落盘日期各留快照,旧版12列文件按文件名前缀补账号。"""
     # 旧版 12 列表头(无 达人昵称/抖音号/蝉妈妈链接)