Merge pull request 'Develop' (#5) from develop into main

Reviewed-on: #5

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 * * * *")

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),

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