import sys,json,csv
|
from collections import defaultdict
|
from read_links import P,connect,encode
|
from plan_links import build
|
BASE=['crm_customer','erp_sale_order','mes_pro_mps','mes_pro_work_order','mes_pro_task','mes_wm_batch']
|
def save(path,obj):path.write_text(encode(obj),encoding='utf8')
|
def main():
|
assert not (P/'result.json').exists(),'Already applied'
|
plan=json.loads((P/'plan.json').read_text(encoding='utf8'))
|
c=connect(sys.stdin.readline().strip())
|
try:
|
with c.cursor() as q:
|
q.execute("SET time_zone='+08:00'")
|
q.execute('SELECT DATABASE() db,NOW() now');state=q.fetchone();assert state['db']=='mom-hm'
|
q.execute("SELECT GET_LOCK('codex-customer-chain-repair-20261015',0) locked");assert q.fetchone()['locked']==1
|
before={}
|
targets={x['table'] for x in plan['changes']}
|
for table in BASE:
|
q.execute('SELECT * FROM `'+table+'` WHERE deleted=0 ORDER BY id'+(' FOR UPDATE' if table in targets else ' FOR SHARE'))
|
before[table]=q.fetchall()
|
backup=P/('locked-before-'+state['now'].strftime('%Y%m%d-%H%M%S')+'.json');save(backup,before)
|
original={t:{r['id']:r for r in rows} for t,rows in before.items()}
|
expected={t:{id:dict(r) for id,r in rows.items()} for t,rows in original.items()}
|
for change in plan['changes']:
|
table,id,field=change['table'],change['id'],change['field']
|
assert encode(original[table][id])==encode(change['source_before']),f'Source changed {table}/{id}'
|
assert change['after'] in original['crm_customer'] and change['before'] not in original['crm_customer']
|
expected[table][id][field]=change['after']
|
for change in plan['changes']:
|
for ev in change['evidence']:
|
assert expected[ev['table']][ev['id']][ev['field']]==change['after'],'Upstream customer changed'
|
for change in plan['changes']:
|
r=original[change['table']][change['id']]
|
q.execute('UPDATE `'+change['table']+'` SET `'+change['field']+'`=%s,update_time=%s WHERE id=%s AND deleted=0 AND `'+change['field']+'` <=> %s',
|
(change['after'],r['update_time'],change['id'],change['before']))
|
assert q.rowcount==1
|
for table in BASE:
|
q.execute('SELECT * FROM `'+table+'` WHERE deleted=0 ORDER BY id');after={r['id']:r for r in q.fetchall()}
|
assert encode(after)==encode(expected[table]),'Unexpected table/field change: '+table
|
c.commit()
|
result=dict(plan['summary'],committed_at=str(state['now']),backup=str(backup),changed_fields=['client_id'],
|
business_dates_status_quantities_amounts_preserved=True,customers_unchanged=True)
|
save(P/'result.json',result)
|
with (P/'客户关联修复清单.csv').open('w',encoding='utf-8-sig',newline='') as f:
|
w=csv.writer(f);w.writerow(['业务表','业务ID','客户ID','客户名称','关联依据'])
|
for r in plan['changes']:w.writerow([r['table'],r['id'],r['after'],r['customer_name'],encode(r['evidence'])])
|
save(P/'待确认出货检验单.json',plan['unresolved']);print(encode(result))
|
except BaseException:c.rollback();raise
|
finally:c.close()
|
if __name__=='__main__':main()
|
