diff --git a/flows/platform_order/platform_order_upper_sync.py b/flows/platform_order/platform_order_upper_sync.py new file mode 100644 index 0000000..c6e0aaf --- /dev/null +++ b/flows/platform_order/platform_order_upper_sync.py @@ -0,0 +1,65 @@ +# -*- coding: utf-8 -*- +"""平台订单 platform 字段大写同步:把 platform_order_data.platform 统一转为大写。 + +优化自 platform_order_upper_sync.py,对应 platform-order-upper-sync。 +""" + +import time + +import psycopg2 +from prefect import flow + +# 数据库连接参数 +conn_params = { + "host": "192.168.124.92", + "port": 5432, + "user": "postgres", + "password": "Inkreach123456.", + "database": "smt", +} + + +def log_msg(msg, level="info"): + ts = time.strftime('%H:%M:%S') + tag = {"success": "[OK]", "error": "[ERR]"}.get(level, "") + print(f"[{ts}] {tag} {msg}") + + +def update_platform_upper() -> int: + conn = None + cur = None + try: + conn = psycopg2.connect(**conn_params) + cur = conn.cursor() + + # 仅更新需要变更的行,避免每次都全表写入 + update_sql = """ + UPDATE platform_order_data + SET platform = UPPER(platform) + WHERE platform IS NOT NULL + AND platform <> UPPER(platform); + """ + cur.execute(update_sql) + affected_rows = cur.rowcount + conn.commit() + log_msg(f"platform 大写转换完成,共 {affected_rows} 行受影响", "success") + return affected_rows + except Exception as e: + log_msg(f"platform 大写转换失败:{e}", "error") + if conn: + conn.rollback() + raise + finally: + if cur: + cur.close() + if conn: + conn.close() + + +@flow(name="platform-order-upper-sync", log_prints=True) +def platform_order_upper_sync() -> int: + return update_platform_upper() + + +if __name__ == "__main__": + platform_order_upper_sync.serve(name="platform-order-upper-sync", cron="*/2 * * * *") diff --git a/flows/platform_order/sdspod.py b/flows/platform_order/sdspod.py index a46bc90..e6f36d2 100644 --- a/flows/platform_order/sdspod.py +++ b/flows/platform_order/sdspod.py @@ -359,10 +359,18 @@ def sd_format_item(item): create_time = item.get("gmtCreated", "") parcel_list = item.get("parcelList", []) if parcel_list and isinstance(parcel_list, list): - qty = sum(p.get("totalQty", 0) for p in parcel_list) + def safe_int(val): + try: + return int(val) + except (ValueError, TypeError): + return 0 + + qty = sum(safe_int(p.get("totalQty", 0)) for p in parcel_list) else: qty = 0 + + return [ to_chinese_country(country), to_chinese_country(country), diff --git a/prefect.yaml b/prefect.yaml index 3fce7f3..4547703 100644 --- a/prefect.yaml +++ b/prefect.yaml @@ -97,7 +97,14 @@ deployments: - name: platform-order-sdspod entrypoint: flows/platform_order/sdspod.py:sdspod_platform_sync schedules: - - cron: '*/30 * * * *' + - cron: 0 * * * * + timezone: Asia/Shanghai + work_pool: + name: inkreach-pool +- name: platform-order-upper-sync + entrypoint: flows/platform_order/platform_order_upper_sync.py:platform_order_upper_sync + schedules: + - cron: '*/2 * * * *' timezone: Asia/Shanghai work_pool: name: inkreach-pool