import sys,json
|
from pathlib import Path
|
from collections import defaultdict
|
from datetime import datetime,date
|
P=Path(__file__).parent
|
sys.stdout.reconfigure(encoding='utf8')
|
sys.path.insert(0,str(P.parent/'iqc-import-20261012/deps'))
|
import pymysql
|
from validate_plan import validate
|
assert not (P/'result.json').exists(),'Already completed; use verify_timing.py.'
|
snapshot=json.loads((P/'snapshot.json').read_text(encoding='utf8'))
|
plan=json.loads((P/'plan.json').read_text(encoding='utf8'))
|
canon=lambda x:json.loads(json.dumps(x,ensure_ascii=False,default=str))
|
c=pymysql.connect(host='36.213.150.237',port=10000,user='root',password=sys.stdin.readline().strip(),database='mom-hm',charset='utf8mb4',autocommit=False,cursorclass=pymysql.cursors.DictCursor,read_timeout=120)
|
try:
|
with c.cursor() as q:
|
q.execute("SET time_zone='+08:00'")
|
q.execute('SELECT DATABASE() db,NOW() now,GET_LOCK(%s,10) locked',('production-timing-20261014',));state=q.fetchone();assert state['db']=='mom-hm' and state['locked']==1
|
before={};byid={}
|
for t in snapshot['tables']:
|
q.execute('SELECT * FROM `'+t+'` WHERE deleted=0 ORDER BY id FOR UPDATE');before[t]=q.fetchall();byid[t]={x['id']:x for x in before[t]}
|
original={t:{r['id']:r for r in v['rows']} for t,v in snapshot['tables'].items()}
|
for change in plan['changes']:
|
t=change['table'];id=change['id'];assert canon(byid[t][id])==original[t][id],('Scoped row changed since planning',t,id)
|
datecols={x['Field'] for x in snapshot['tables'][t]['columns'] if x['Type'].startswith(('date','timestamp'))};assert set(change['after'])<=datecols,'Only date fields may be changed'
|
backup=P/('locked-before-'+state['now'].strftime('%Y%m%d-%H%M%S')+'.json')
|
backup.write_text(json.dumps({'state':state,'tables':before},ensure_ascii=False,default=str,indent=2),encoding='utf8')
|
# Check the exact proposed full state against the latest database baseline first.
|
proposed={t:[{**r,**next((x['after'] for x in plan['changes'] if x['table']==t and x['id']==r['id']),{})} for r in rr] for t,rr in before.items()}
|
validate(before,proposed,plan,state['now'])
|
groups=defaultdict(list)
|
for x in plan['changes']:groups[x['table'],tuple(x['after'].items())].append(x['id'])
|
for (t,fields),ids in groups.items():
|
args=[]
|
cols={x['Field']:x['Type'] for x in snapshot['tables'][t]['columns']}
|
for k,v in fields:args.append(None if v is None else date.fromisoformat(v) if cols[k]=='date' else datetime.fromisoformat(v))
|
sql='UPDATE `'+t+'` SET '+','.join('`'+k+'`=%s' for k,v in fields)
|
if 'update_time' not in dict(fields):sql+=',update_time=update_time'
|
q.execute(sql+' WHERE deleted=0 AND id IN ('+','.join(['%s']*len(ids))+')',[*args,*ids]);assert q.rowcount==len(ids)
|
after={}
|
for t in before:
|
q.execute('SELECT * FROM `'+t+'` WHERE deleted=0 ORDER BY id');after[t]=q.fetchall()
|
verification=validate(before,after,plan,state['now'])
|
c.commit()
|
result={**verification,'committed':True,'at':str(state['now']),'backup':backup.name,'quantity_basis':plan['quantity_basis'],'approved_upstream_exception':'26000495报价合同前移至2026-10-12','no_amount_quantity_status_result_or_code_changes':True}
|
(P/'result.json').write_text(json.dumps(result,ensure_ascii=False,indent=2),encoding='utf8')
|
print(json.dumps(result,ensure_ascii=False))
|
except Exception:c.rollback();raise
|
finally:
|
with c.cursor() as q:q.execute('SELECT RELEASE_LOCK(%s)',('production-timing-20261014',))
|
c.close()
|
