@@ -0,0 +1,144 @@
# -*- coding: utf-8 -*-
""" shop_dim 同步 flow：
最小包装自 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 * * * * " )