diff --git a/.python-version b/.python-version new file mode 100644 index 0000000..e4fba21 --- /dev/null +++ b/.python-version @@ -0,0 +1 @@ +3.12 diff --git a/flows/platform_order/sync_shop_dim.py b/flows/platform_order/sync_shop_dim.py new file mode 100644 index 0000000..9607e5a --- /dev/null +++ b/flows/platform_order/sync_shop_dim.py @@ -0,0 +1,144 @@ +# -*- coding: utf-8 -*- +"""shop_dim 同步 flow:teable(源) -> 本地 smt 库(目标)。 + +最小包装自 sync_shop_dim.py,对应 shop-dim-sync,每 5 分钟调度一次。 +源: PostgreSQL @ table.inkreach.cc:10432 db=teable schema=bsehWCxpH6HoHUdkjUd + 物理表 tblUF7OrBxeRKUyWBc2 (teable 里名为 "shop_dim" 的表) +目标: PostgreSQL @ 192.168.124.92:5432 db=smt public.shop_dim + +本地直接跑: python flows/platform_order/sync_shop_dim.py +""" + +import time +from typing import Any + +import psycopg2 +from prefect import flow + +SRC = dict(host="table.inkreach.cc", port=10432, dbname="teable", + user="read_only_role_bsehWCxpH6HoHUdkjUd", + password="ygBuh8Au5J-PjuNQASos0", connect_timeout=25) +SRC_TABLE = '"bsehWCxpH6HoHUdkjUd"."tblUF7OrBxeRKUyWBc2"' + +DST = dict(host="192.168.124.92", port=5432, dbname="smt", + user="postgres", password="Inkreach123456.", connect_timeout=15) + +COL_MAP = [ + ("Guo_Jia", "country"), + ("Dian_Pu_Ming_Cheng", "shop_name"), + ("DBDian_Pu_Ming", "db_store_name"), + ("Ping_Tai", "platform"), + ("Lei_Xing", "shop_type"), + ("Deng_Ji", "tier"), + ("Fu_Ze_Ren", "owner"), + ("Pin_Lei", "category"), + ("Xia_Dian_Shi_Jian", "opened_at"), + ("Mu_Biao_Jian_Shu", "sep_target_qty"), +] + + +def log_msg(msg): + print(msg) + + +def connect(cfg, tries=5): + last = None + for i in range(tries): + try: + return psycopg2.connect(**cfg) + except Exception as e: + last = e + log_msg(f" 连接重试 {i+1}/{tries}: {e}") + time.sleep(3) + raise ConnectionError(f"连接失败: {last}") + + +def norm(v, col): + if v is None: + return None + if col == "opened_at": + return v.date() if hasattr(v, "date") else v + if col == "sep_target_qty": + try: + return int(round(float(v))) + except Exception: + return None + s = str(v).strip() + return s if s != "" else None + + +@flow(name="shop-dim-sync", log_prints=True) +def shop_dim_sync(dry_run: bool = False, prune: bool = True) -> dict[str, Any]: + log_msg(f"[1/4] 读取源表 {SRC_TABLE} ...") + sc = connect(SRC) + scur = sc.cursor() + sel = ", ".join('"%s"' % c for c, _ in COL_MAP) + scur.execute("SELECT %s FROM %s" % (sel, SRC_TABLE)) + raw = scur.fetchall() + scur.close(); sc.close() + src_rows = [ + dict(zip([d for _, d in COL_MAP], [norm(v, c) for v, (_, c) in zip(row, COL_MAP)])) + for row in raw + ] + log_msg(f" 源记录: {len(src_rows)} 行") + + log_msg("[2/4] 连接目标 smt.shop_dim ...") + dc = connect(DST) + dcur = dc.cursor() + dcur.execute("SELECT country, db_store_name FROM shop_dim WHERE db_store_name IS NOT NULL") + existing = {(r[0], r[1]) for r in dcur.fetchall()} + log_msg(f" 目标现有: {len(existing)} 个 (country, db_store_name)") + + src_keys = {(r["country"], r["db_store_name"]) for r in src_rows} + to_update = src_keys & existing + to_insert = src_keys - existing + to_delete = existing - src_keys + log_msg(f"[3/4] 计划: 更新 {len(to_update)} | 新增 {len(to_insert)} | 待删 {len(to_delete)}") + + if dry_run: + log_msg("[dry-run] 不执行写入。") + if to_delete: + log_msg(" 待删示例: " + str(list(to_delete)[:10])) + dcur.close(); dc.close() + return {"sourceRows": len(src_rows), "dryRun": True} + + cols = [d for _, d in COL_MAP] + upd = 0 + ins = 0 + for r in src_rows: + vals = [r[c] for c in cols] + if (r["country"], r["db_store_name"]) in existing: + set_clause = ", ".join(f"{c}=%s" for c in cols) + dcur.execute( + f"UPDATE shop_dim SET {set_clause} WHERE country=%s AND db_store_name=%s", + vals + [r["country"], r["db_store_name"]]) + upd += dcur.rowcount + else: + ph = ", ".join(["%s"] * len(cols)) + dcur.execute(f"INSERT INTO shop_dim ({', '.join(cols)}) VALUES ({ph})", vals) + ins += dcur.rowcount + + dele = 0 + if prune and to_delete: + for country, dbn in to_delete: + dcur.execute("DELETE FROM shop_dim WHERE country=%s AND db_store_name=%s", (country, dbn)) + dele += dcur.rowcount + + dc.commit() + log_msg(f"[4/4] 完成: 更新 {upd} | 新增 {ins} | 删除 {dele}") + dcur.execute("SELECT COUNT(*) FROM shop_dim WHERE db_store_name IS NOT NULL") + total = dcur.fetchone()[0] + log_msg(f" 目标现有总数: {total}") + dcur.close(); dc.close() + + return { + "sourceRows": len(src_rows), + "updated": upd, + "inserted": ins, + "deleted": dele, + "total": total, + } + + +if __name__ == "__main__": + shop_dim_sync.serve(name="shop-dim-sync", cron="*/5 * * * *") diff --git a/prefect.yaml b/prefect.yaml index 4547703..1d1c04a 100644 --- a/prefect.yaml +++ b/prefect.yaml @@ -108,6 +108,13 @@ deployments: timezone: Asia/Shanghai work_pool: name: inkreach-pool +- name: shop-dim-sync + entrypoint: flows/platform_order/sync_shop_dim.py:shop_dim_sync + schedules: + - cron: '*/5 * * * *' + timezone: Asia/Shanghai + work_pool: + name: inkreach-pool - name: label-ocr entrypoint: flows/label_ocr.py:label_ocr schedules: