"""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']