"""Copy only this import to the authorized remote database; map every business FK."""
|
from pathlib import Path
|
import sys,json,re
|
from decimal import Decimal as D
|
sys.stdout.reconfigure(encoding='utf-8')
|
sys.path.insert(0,str(Path('scripts/iqc-import-20261012/deps').resolve()))
|
import pymysql
|
P=Path(__file__).parent;E=json.loads((P/'sync-export.json').read_text(encoding='utf-8'));data=E['data'];refs=E['references'];APPLY='--apply' in sys.argv
|
c=pymysql.connect(host='36.213.150.237',port=10000,user='root',password=sys.stdin.readline().strip(),database='mom-hm',charset='utf8mb4',cursorclass=pymysql.cursors.DictCursor,connect_timeout=15,read_timeout=90,autocommit=False)
|
maps={};before=[];sqls=[];created={};schemas={};deferred=[];code_map={};result={}
|
def dump(name,v):(P/name).write_text(json.dumps(v,ensure_ascii=False,indent=2,default=str),encoding='utf-8')
|
def rows(table):q.execute('select * from `'+table+'` where deleted=0');return q.fetchall()
|
def schema(table):
|
if table not in schemas:q.execute('show full columns from `'+table+'`');schemas[table]={r['Field']:r for r in q.fetchall()}
|
return schemas[table]
|
def raw_update(table,rid,values):
|
q.execute('select * from `'+table+'` where id=%s for update',(rid,));before.append({'table':table,'row':q.fetchone()})
|
sql='update `'+table+'` set '+','.join('`'+k+'`=%s' for k in values)+' where id=%s';args=list(values.values())+[rid]
|
q.execute(sql,args);sqls.append(q.mogrify(sql,args)+';')
|
def insert(table,values,source_id=None):
|
v={k:z for k,z in values.items() if k!='id'};f=schema(table);assert not set(v)-set(f),(table,'missing columns',set(v)-set(f))
|
for k,z in list(v.items()):
|
if isinstance(z,str) and z in ("b'\\x00'","b'\\x01'"):v[k]=int(z=="b'\\x01'")
|
if isinstance(v[k],str):
|
m=re.match(r'varchar\((\d+)\)',f[k]['Type']);assert not m or len(v[k])<=int(m[1]),(table,k,len(v[k]),m[1])
|
sql='insert into `'+table+'` (`'+'`,`'.join(v)+'`) values('+','.join(['%s']*len(v))+')';args=list(v.values());q.execute(sql,args);rid=q.lastrowid
|
created.setdefault(table,[]).append(rid);sqls.append(q.mogrify(sql,args)+';')
|
if source_id is not None:maps.setdefault(table,{})[source_id]=rid
|
return rid
|
def mapped(table,rid):
|
if rid is None:return None
|
if rid==0:return 0
|
assert rid in maps.get(table,{}),(table,'unmapped reference',rid)
|
return maps[table][rid]
|
FK={
|
'crm_sale_quotation':{'customer_id':'crm_customer','owner_user_id':'system_users','contract_id':'crm_contract'},
|
'crm_contract':{'customer_id':'crm_customer','owner_user_id':'system_users','sign_user_id':'system_users','order_id':'erp_sale_order'},
|
'erp_sale_order':{'customer_id':'crm_customer','contract_id':'crm_contract','sale_user_id':'system_users'},
|
'crm_receivable_plan':{'customer_id':'crm_customer','contract_id':'crm_contract','owner_user_id':'system_users'},
|
'crm_sale_quotation_product':{'quotation_id':'crm_sale_quotation','item_id':'mdm_item','item_unit_id':'mdm_unit_measure'},
|
'crm_contract_product':{'contract_id':'crm_contract','item_id':'mdm_item','item_unit_id':'mdm_unit_measure'},
|
'erp_sale_order_items':{'order_id':'erp_sale_order','product_id':'mdm_item','product_unit_id':'mdm_unit_measure','mps_id':'mes_pro_mps'},
|
'mes_pro_work_order':{'product_id':'mes_md_item','route_id':'mes_pro_route','client_id':'crm_customer'},
|
'mes_pro_mps':{'sale_order_id':'erp_sale_order','sale_order_item_id':'erp_sale_order_items','product_id':'mes_md_item','unit_measure_id':'mes_md_unit_measure','client_id':'crm_customer','work_order_id':'mes_pro_work_order'},
|
'mes_pro_work_order_process':{'work_order_id':'mes_pro_work_order','process_id':'mes_pro_process','param_template_id':'mes_pd_process_param_template','next_process_id':'mes_pro_process'},
|
'mes_pro_work_order_bom':{'work_order_id':'mes_pro_work_order','item_id':'mes_md_item','process_id':'mes_pro_process'},
|
'mes_pro_task':{'work_order_id':'mes_pro_work_order','workstation_id':'mes_md_workstation','route_id':'mes_pro_route','process_id':'mes_pro_process','param_template_id':'mes_pd_process_param_template','item_id':'mes_md_item','client_id':'crm_customer'},
|
'mes_pro_feedback':{'work_order_id':'mes_pro_work_order','task_id':'mes_pro_task','workstation_id':'mes_md_workstation','route_id':'mes_pro_route','process_id':'mes_pro_process','item_id':'mes_md_item','feedback_user_id':'system_users','approve_user_id':'system_users'},
|
'mes_pro_work_order_param_record':{'work_order_id':'mes_pro_work_order','task_id':'mes_pro_task','template_id':'mes_pd_process_param_template'},
|
'mes_pro_work_order_param_record_detail':{'record_id':'mes_pro_work_order_param_record','unit_measure_id':'mes_md_unit_measure'},
|
'mes_pro_task_assignee':{'task_id':'mes_pro_task','user_id':'system_users'},
|
'mes_pro_route_process':{'route_id':'mes_pro_route','process_id':'mes_pro_process','param_template_id':'mes_pd_process_param_template','next_process_id':'mes_pro_process'},
|
'mes_pro_route_product':{'route_id':'mes_pro_route','item_id':'mes_md_item'}}
|
def prepare(table,row):
|
v=dict(row)
|
for k,t in FK.get(table,{}).items():
|
if v.get(k) is not None:
|
if v[k] not in maps.get(t,{}):deferred.append((table,row['id'],k,t,v[k]));v[k]=None
|
else:v[k]=mapped(t,v[k])
|
for k in ['creator','updater']:
|
if str(v.get(k,'')).isdigit():v[k]=str(mapped('system_users',int(v[k])))
|
if v.get('output_item_id'):v['output_item_id']=','.join(str(mapped('mes_md_item',int(i.strip()))) for i in v['output_item_id'].split(','))
|
if table=='mes_pro_work_order_param_record':v['record_no']='TASK-'+str(v['task_id'])
|
if table=='mes_pro_work_order_param_record_detail':
|
# Template detail ids are explicitly verified below against parameter code and template.
|
v['template_detail_id']=mapped('mes_pd_process_param_template_detail',v.get('template_detail_id'))
|
return v
|
|
try:
|
with c.cursor() as q:
|
q.execute('select database() db,get_lock(%s,15) locked',('ecommerce-production-sync-20261015',));state=q.fetchone();assert state['db']=='mom-hm' and state['locked']==1
|
q.execute('select count(*) n from crm_sale_quotation where deleted=0 and remark like %s',('%来源：10月15日电商未缴库计划%',));assert q.fetchone()['n']==0,'Remote import already exists; do not duplicate'
|
# Inspect field width before any writes. Widen only the known template/snapshot mismatch.
|
f=schema('mes_pro_work_order_param_record_detail')['remark'];width=int(re.search(r'\((\d+)\)',f['Type']).group(1))
|
if width<500:
|
if not APPLY:print('Schema prerequisite: remark varchar(255) -> varchar(500)')
|
else:
|
dump('remote-schema-before.json',f)
|
q.execute("alter table mes_pro_work_order_param_record_detail modify column remark varchar(500) character set utf8mb4 collate utf8mb4_unicode_ci null comment '备注（参数快照）'")
|
schemas.pop('mes_pro_work_order_param_record_detail',None)
|
protected={}
|
for t in ['crm_receivable','mes_wm_transaction','mes_wm_material_stock','mes_wm_product_receipt','mes_wm_product_sales']:
|
q.execute('select count(*) n from `'+t+'`');protected[t]=q.fetchone()['n']
|
# Match shared masters by stable code (units by trimmed name), never by local primary key alone.
|
for table,key in [('system_users','nickname'),('crm_customer','name'),('mdm_unit_measure','name'),('mes_md_unit_measure','name'),('mes_pro_process','code'),('mes_md_workstation','code'),('mes_dv_machinery','code'),('mes_pd_process_param_template','template_code'),('mes_md_auto_code_rule','code'),('mes_pro_route','code'),('mdm_item','code'),('mes_md_item','code')]:
|
remote=rows(table);lookup={}
|
for v in remote:lookup.setdefault(str(v.get(key) or '').strip(),[]).append(v)
|
maps.setdefault(table,{})
|
for local in refs[table]:
|
match=lookup.get(str(local.get(key) or '').strip(),[])
|
if len(match)==1:maps[table][local['id']]=match[0]['id']
|
if table in ['system_users','mdm_unit_measure','mes_md_unit_measure']:
|
# Users/unit ids may share names; exact identity is safe when both id and name agree.
|
for local in refs[table]:
|
exact=next((v for v in remote if v['id']==local['id'] and str(v.get(key) or '').strip()==str(local.get(key) or '').strip()),None)
|
if exact:maps[table][local['id']]=exact['id']
|
assert mapped('crm_customer',1613)
|
for uid in [244,291]:assert mapped('system_users',uid)
|
new_codes={'12100271','UC000017','U2100282'}
|
original_updates=E['local_result']['updates']
|
for table in ['mdm_item','mes_md_item']:
|
for row in data[table]:
|
rid=maps.get(table,{}).get(row['id'])
|
if rid:
|
values={k:row[k] for k in ['name','specification','unit_measure_id']}
|
unit_table='mdm_unit_measure' if table=='mdm_item' else 'mes_md_unit_measure';values['unit_measure_id']=mapped(unit_table,row['unit_measure_id'])
|
q.execute('select * from `'+table+'` where id=%s for update',(rid,));remote=q.fetchone()
|
diff={k:v for k,v in values.items() if remote.get(k)!=v}
|
if diff:
|
originals=[x['before'] for x in original_updates if x['table']==table and x['id']==row['id']]
|
assert originals or row['code'] in new_codes,(table,row['code'],'no authorized source change')
|
for k in diff:
|
permitted=[o.get(k) for o in originals]+[row.get(k)]
|
assert remote.get(k) in permitted,(table,row['code'],k,'remote changed independently',remote.get(k),permitted)
|
raw_update(table,rid,diff)
|
else:
|
assert row['code'] in new_codes,(table,row['code'],'missing existing master')
|
v=dict(row);v['unit_measure_id']=mapped('mdm_unit_measure' if table=='mdm_item' else 'mes_md_unit_measure',row['unit_measure_id'])
|
if table=='mes_md_item':v['mdm_item_id']=mapped('mdm_item',row['mdm_item_id'])
|
insert(table,v,row['id'])
|
for row in data['mes_pro_route']:
|
if row['id'] not in maps['mes_pro_route']:insert('mes_pro_route',row,row['id'])
|
for row in data['mes_pro_route_process']:
|
v=prepare('mes_pro_route_process',row);q.execute('select id from mes_pro_route_process where route_id=%s and process_id=%s and deleted=0',(v['route_id'],v['process_id']));existing=q.fetchone()
|
if not existing:insert('mes_pro_route_process',v,row['id'])
|
else:maps.setdefault('mes_pro_route_process',{})[row['id']]=existing['id']
|
for row in data['mes_pro_route_product']:
|
v=prepare('mes_pro_route_product',row);q.execute('select id from mes_pro_route_product where route_id=%s and item_id=%s and deleted=0',(v['route_id'],v['item_id']));existing=q.fetchone()
|
if not existing:insert('mes_pro_route_product',v,row['id'])
|
# Verify parameter detail references using template + code, including the full source remark.
|
q.execute('select * from mes_pd_process_param_template_detail where deleted=0');remote_details=q.fetchall()
|
audit=json.loads((P/'audit.json').read_text(encoding='utf-8'));maps['mes_pd_process_param_template_detail']={}
|
for record in data['mes_pro_work_order_param_record_detail']:
|
src=next(x for x in audit['mes_pd_process_param_template_detail'] if x['id']==record['template_detail_id'])
|
matches=[x for x in remote_details if x['template_id']==mapped('mes_pd_process_param_template',src['template_id']) and x['param_code']==src['param_code']]
|
assert len(matches)==1,('parameter detail mismatch',src['id']);maps['mes_pd_process_param_template_detail'][src['id']]=matches[0]['id']
|
# Check codes before writes; unrelated remote documents must never be replaced.
|
code_tables={'crm_sale_quotation':'no','crm_contract':'no','erp_sale_order':'no','mes_pro_work_order':'code','mes_pro_mps':'code','mes_pro_task':'code','mes_pro_feedback':'code'}
|
for table,col in code_tables.items():
|
for row in data[table]:
|
q.execute('select id from `'+table+'` where `'+col+'`=%s',(row[col],));assert not q.fetchone(),(table,row[col],'code conflict; regenerate against remote before sync')
|
for table in ['crm_sale_quotation','crm_contract','erp_sale_order','crm_receivable_plan','crm_sale_quotation_product','crm_contract_product','erp_sale_order_items','mes_pro_work_order','mes_pro_mps','mes_pro_work_order_process','mes_pro_work_order_bom','mes_pro_task','mes_pro_feedback','mes_pro_work_order_param_record','mes_pro_task_assignee','mes_pro_work_order_param_record_detail']:
|
for row in data[table]:insert(table,prepare(table,row),row['id'])
|
for table,local_id,key,target,local_fk in deferred:raw_update(table,mapped(table,local_id),{key:mapped(target,local_fk)})
|
for row in data['crm_permission']:
|
table={9:'crm_sale_quotation',5:'crm_contract',8:'crm_receivable_plan'}[row['biz_type']]
|
v=dict(row);v['biz_id']=mapped(table,row['biz_id']);v['user_id']=mapped('system_users',row['user_id']);insert('crm_permission',v,row['id'])
|
for row in data['mes_md_auto_code_record']:
|
v=dict(row);v['rule_id']=mapped('mes_md_auto_code_rule',row['rule_id']);q.execute('select id from mes_md_auto_code_record where rule_id=%s and result=%s and deleted=0',(v['rule_id'],v['result']));reservation=q.fetchone()
|
if reservation:maps.setdefault('mes_md_auto_code_record',{})[row['id']]=reservation['id']
|
else:insert('mes_md_auto_code_record',v,row['id'])
|
# Cross-check every product and every financial chain within the same transaction.
|
for row in data['mes_pro_mps']:
|
q.execute('select m.quantity,m.unit_measure_id,l.count,l.product_unit_id,l.mps_id,w.quantity wo_qty,w.parent_id from mes_pro_mps m join erp_sale_order_items l on l.id=m.sale_order_item_id join mes_pro_work_order w on w.id=m.work_order_id where m.id=%s',(mapped('mes_pro_mps',row['id']),));v=q.fetchone();assert v['quantity']==v['count']==v['wo_qty']==D(str(row['quantity'])) and v['unit_measure_id']==v['product_unit_id'] and v['mps_id']==mapped('mes_pro_mps',row['id']) and v['parent_id']==0
|
total=D(0)
|
for group in E['local_result']['groups'].values():
|
values=[]
|
for table,field,totalfield in [('crm_sale_quotation','quotation_id','total_price'),('crm_contract','contract_id','total_price'),('erp_sale_order','order_id','total_price'),('crm_receivable_plan','receivable_plan_id','price')]:
|
q.execute('select `'+totalfield+'` amount from `'+table+'` where id=%s',(mapped(table,group[field]),));values.append(q.fetchone()['amount'])
|
assert all(x==D(group['amount']) for x in values);total+=values[0]
|
assert total==D('2062353.18')
|
for table,count in protected.items():q.execute('select count(*) n from `'+table+'`');assert q.fetchone()['n']==count,(table,'unrelated inventory/receipt changed')
|
# Inspect active resource collisions rather than silently overwriting an online schedule.
|
new_tasks=created['mes_pro_task'];q.execute('select * from mes_pro_task where deleted=0 and (status is null or status not in (4,5))');active=q.fetchall()
|
for t in active:
|
if t['id'] not in new_tasks:continue
|
for other in active:
|
if other['id']==t['id'] or other['workstation_id']!=t['workstation_id']:continue
|
assert not (other['start_time'] and other['end_time'] and t['start_time']<other['end_time'] and other['start_time']<t['end_time']),('online resource conflict',t['code'],other['code'])
|
result={'applied':APPLY,'host':'36.213.150.237:10000','database':'mom-hm','total':str(total),'maps':maps,'created':created,'protected_counts':protected,'source_codes':{t:[x[col] for x in data[t]] for t,col in code_tables.items()}}
|
dump('remote-before-apply.json' if APPLY else 'remote-before-preview.json',before)
|
dump('remote-result-applied.json' if APPLY else 'remote-result-preview.json',result)
|
(P/('remote-applied.sql' if APPLY else 'remote-preview.sql')).write_text('START TRANSACTION;\n'+'\n'.join(sqls)+'\n'+('COMMIT;' if APPLY else 'ROLLBACK;'),encoding='utf-8')
|
if APPLY:c.commit()
|
else:c.rollback()
|
print(json.dumps({'applied':APPLY,'counts':{t:len(v) for t,v in created.items()},'total':str(total)},ensure_ascii=False))
|
finally:
|
c.rollback()
|
with c.cursor() as q:q.execute('select release_lock(%s)',('ecommerce-production-sync-20261015',))
|
c.close()
|
