From b0e1b5066de93ecf70003cd823806cde2c9b5d5b Mon Sep 17 00:00:00 2001 From: wangyunlong <3080433063@qq.com> Date: Fri, 9 Oct 2026 13:45:30 +0800 Subject: [PATCH] =?UTF-8?q?feat(content):=20=E8=87=AA=E8=90=A5=E4=B8=89?= =?UTF-8?q?=E5=B9=B3=E5=8F=B0=E6=8C=87=E6=A0=87=E5=85=A5=E5=BA=93=E7=AE=A1?= =?UTF-8?q?=E7=BA=BF=E5=AE=8C=E5=96=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit sync_self_video_metrics 按平台隔离入库与字段口径调整，配套自营指标验收测试 更新。 --- .../data/tools/sync_self_video_metrics.py | 188 +++++++++++------- .../test_self_content_metrics.py | 77 +++++++ 2 files changed, 193 insertions(+), 72 deletions(-) diff --git a/src/gyxx_flow/modules/content_marketing/data/tools/sync_self_video_metrics.py b/src/gyxx_flow/modules/content_marketing/data/tools/sync_self_video_metrics.py index f25dab0..4f7c191 100644 --- a/src/gyxx_flow/modules/content_marketing/data/tools/sync_self_video_metrics.py +++ b/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()})的产物; " - "补历史请显式传 --metric-date 并加 --allow-stale:\n " - + "\n ".join(stale) + f"⚠️ 以下来源文件不是业务日期({metric_date.isoformat()})的产物, " + f"已跳过对应 {dropped} 行(补历史用 --metric-date + " + "--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: - ensure_table(cur) - for row in all_rows.values(): - # 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, - ) - conn.commit() - print(f"[sync-self] 完成: 已幂等写入 {len(all_rows)} 行到 {TABLE_NAME}") + written, failed = upsert_rows(conn, list(all_rows.values())) + if failed: + print(f"[sync-self] 完成: 写入 {written} 行,单行失败跳过 {failed} 行 " + f"(见上方 ⚠️ 日志)") + return 0 if written else 1 + print(f"[sync-self] 完成: 已幂等写入 {written} 行到 {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()) diff --git a/tests/modules/content_marketing/test_self_content_metrics.py b/tests/modules/content_marketing/test_self_content_metrics.py index 709a670..d5525a3 100644 --- a/tests/modules/content_marketing/test_self_content_metrics.py +++ b/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 列表头(无 达人昵称/抖音号/蝉妈妈链接)