"""Atomic migration of QC configuration, existing line standards and reviewed reports."""
|
import sys,json,re,hashlib,collections,copy
|
from pathlib import Path
|
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;apply='--apply' in sys.argv
|
audit=json.loads((P/'audit.json').read_text(encoding='utf8'));plan=json.loads((P/'plan.json').read_text(encoding='utf8'));reports=json.loads((P/'report-output.json').read_text(encoding='utf8'))
|
upgrade=json.loads((P/'report-template-upgrade.json').read_text(encoding='utf8'))
|
assert len(reports)==12
|
def canon(v):return json.loads(json.dumps(v,ensure_ascii=False,default=str))
|
def fingerprint(v):return hashlib.sha256(json.dumps(v,ensure_ascii=False,sort_keys=True,default=str).encode()).hexdigest()
|
def dump(name,v):(P/name).write_text(json.dumps(v,ensure_ascii=False,indent=2,default=str),encoding='utf8')
|
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,autocommit=False,read_timeout=90)
|
allowed={'mes_qc_indicator','mes_qc_template_indicator','mes_qc_iqc_line','mes_qc_ipqc_line','mes_qc_oqc_line','mes_qc_rqc_line','mes_qc_indicator_result_detail','qc_report_instance','qc_report_template','system_storage_attachment'}
|
merged={};new_units=[];sql_log=[];before={};unitids={}
|
for ch in plan['changes']:
|
assert ch['table'] in allowed
|
merged.setdefault((ch['table'],ch['id']),{}).update(ch['after'])
|
with c.cursor() as q:
|
try:
|
q.execute('SELECT GET_LOCK(%s,10)',('qc-indicator-standard-optimize',));assert list(q.fetchone().values())==[1]
|
for t in allowed|{'mes_md_unit_measure','qc_report_template_version'}:
|
q.execute('SELECT * FROM `'+t+'` ORDER BY id FOR UPDATE');before[t]=q.fetchall()
|
# Refuse to apply a stale plan; inactive rows are preserved too.
|
for t in allowed-{'system_storage_attachment'}:
|
assert canon(before[t])==audit[t]['rows'],('Data changed after audit; regenerate plan',t)
|
q.execute('SHOW FULL COLUMNS FROM system_storage_attachment');attachment_fields={x['Field'] for x in q.fetchall()}
|
for name in plan['units']:
|
found=[x for x in before['mes_md_unit_measure'] if x['name']==name and x['deleted'] in (0,b'\x00') and x['status']==0]
|
if found:unitids[name]=found[0]['id'];continue
|
code='QC_UNIT_'+hashlib.sha256(name.encode()).hexdigest()[:12].upper()
|
assert not any(x['code']==code for x in before['mes_md_unit_measure'])
|
sql='INSERT INTO mes_md_unit_measure (code,name,primary_flag,primary_id,change_rate,status,remark,creator,updater) VALUES (%s,%s,1,NULL,1,0,%s,%s,%s)'
|
args=(code,name,'质量检测专用计量单位；依据现有方案标准及检验记录单位规范化。','1','1')
|
q.execute(sql,args);unitids[name]=q.lastrowid;new_units.append({'id':q.lastrowid,'name':name,'code':code});sql_log.append(q.mogrify(sql,args)+';')
|
for assignment in plan['unit_assignments']:
|
merged.setdefault((assignment['table'],assignment['id']),{})['unit_measure_id']=unitids[assignment['unit_name']]
|
existing_versions=before['qc_report_template_version']
|
assert not any(x['template_id']==upgrade['template_id'] and x['version']==upgrade['version'] for x in existing_versions)
|
old_version=next(x for x in existing_versions if x['template_id']==upgrade['template_id'] and x['version']==upgrade['source_version'] and x['deleted'] in (0,b'\x00'))
|
sql='INSERT INTO qc_report_template_version (template_id,version,schema_json,status,description,creator,updater) VALUES (%s,%s,%s,%s,%s,%s,%s)'
|
args=(upgrade['template_id'],upgrade['version'],json.dumps(upgrade['schema'],ensure_ascii=False,separators=(',',':')),old_version['status'],'质检数据规范化：展示完整标准要求、允许区间和检测单位；保留原模板版本。','1','1')
|
q.execute(sql,args);new_version_id=q.lastrowid;sql_log.append(q.mogrify(sql,args)+';')
|
merged[('qc_report_template',upgrade['template_id'])]={'current_version':upgrade['version']}
|
for report in reports:
|
assert report['html'].strip() and '{{' not in report['html']
|
merged[('qc_report_instance',report['id'])]={'data_snapshot':json.dumps(report['snapshot'],ensure_ascii=False,separators=(',',':')),'render_html':report['html'],'template_version':report['template_version']}
|
# Invalidate only stale PDF attachment bindings, retaining physical blobs and the previous report in the backup.
|
reportids={x['id'] for x in reports};old_attachments=[]
|
for row in before['system_storage_attachment']:
|
if row.get('record_type')=='qc_report_instance' and str(row.get('record_id')) in {str(x) for x in reportids} and row['deleted'] in (0,b'\x00'):
|
merged[('system_storage_attachment',row['id'])]={'deleted':1};old_attachments.append(row['id'])
|
protected_tables=['mes_qc_iqc','mes_qc_ipqc','mes_qc_oqc','mes_qc_rqc','mes_qc_indicator_result','mes_qc_template','mes_qc_template_item','erp_sale_order','erp_sale_order_items','crm_contract','crm_sale_quotation','crm_receivable_plan','mes_pro_mps','mes_pro_work_order','mes_pro_task','mes_pro_feedback','mes_wm_batch','mes_wm_material_stock','mes_wm_transaction','mes_wm_product_sales','mes_wm_product_sales_line']
|
protected={}
|
for t in protected_tables:q.execute('SELECT * FROM `'+t+'` ORDER BY id');protected[t]=fingerprint(q.fetchall())
|
dump('before-'+('apply' if apply else 'preview')+'.json',{'tables':before,'protected_hashes':protected})
|
schema={t:{x['Field']:x for x in audit[t]['columns']} for t in allowed if t in audit}
|
q.execute('SHOW FULL COLUMNS FROM system_storage_attachment');schema['system_storage_attachment']={x['Field']:x for x in q.fetchall()}
|
byid={t:{x['id']:x for x in rr} for t,rr in before.items()};changes=[];batches=collections.defaultdict(list)
|
for (t,rid),v in merged.items():
|
row=byid[t][rid];v={k:z for k,z in v.items() if row[k]!=z}
|
if not v:continue
|
assert set(v)<=set(schema[t])
|
for k,z in v.items():
|
if z is None:assert schema[t][k]['Null']=='YES'
|
if isinstance(z,str):
|
m=re.match(r'(?:var)?char\((\d+)\)',schema[t][k]['Type']);assert not m or len(z)<=int(m[1]),(t,rid,k,len(z))
|
if schema[t][k]['Type'].startswith('decimal') and z is not None:v[k]=D(z)
|
v={k:z for k,z in v.items() if row[k]!=z}
|
if not v:continue
|
changes.append({'table':t,'id':rid,'before':{k:row[k] for k in v},'after':v})
|
batches[(t,tuple(v.items()))].append(rid)
|
for (t,values),ids in batches.items():
|
for start in range(0,len(ids),400):
|
chunk=ids[start:start+400]
|
sql='UPDATE `'+t+'` SET '+','.join('`'+k+'`=%s' for k,z in values)+',update_time=update_time WHERE id IN ('+','.join(['%s']*len(chunk))+')'
|
args=tuple(z for k,z in values)+tuple(chunk);q.execute(sql,args);assert q.rowcount==len(chunk),(t,q.rowcount,len(chunk));sql_log.append(q.mogrify(sql,args)+';')
|
expected={(x['table'],x['id']):x['after'] for x in changes}
|
for t,rr in before.items():
|
if t in ['mes_md_unit_measure','qc_report_template_version']:continue
|
q.execute('SELECT * FROM `'+t+'` ORDER BY id');after=q.fetchall();assert len(after)==len(rr)
|
for old,new in zip(rr,after):
|
exp={**old,**expected.get((t,old['id']),{})}
|
for k,z in new.items():
|
if isinstance(z,D):assert z==D(exp[k]),(t,old['id'],k)
|
elif schema[t].get(k,{}).get('Type','').startswith('bit') and k in expected.get((t,old['id']),{}):assert z==bytes([exp[k]])
|
else:assert z==exp[k],(t,old['id'],k)
|
for t,h in protected.items():q.execute('SELECT * FROM `'+t+'` ORDER BY id');assert fingerprint(q.fetchall())==h,t
|
q.execute('SELECT id,parent_id,item_type,result_type FROM mes_qc_indicator WHERE deleted=0');ind={x['id']:x for x in q.fetchall()}
|
for row in ind.values():
|
if row['item_type']==2:assert row['result_type'] is None and row['parent_id']==0
|
else:assert row['result_type'] in [1,2,3] and row['parent_id'] in ind and ind[row['parent_id']]['item_type']==2
|
for t in ['mes_qc_template_indicator','mes_qc_iqc_line','mes_qc_ipqc_line','mes_qc_oqc_line','mes_qc_indicator_result_detail']:
|
q.execute('SELECT indicator_id FROM `'+t+'` WHERE deleted=0');assert all(x['indicator_id'] in ind for x in q.fetchall()),t
|
q.execute('SELECT * FROM qc_report_template_version ORDER BY id');versions=q.fetchall();assert len(versions)==len(existing_versions)+1
|
assert versions[:-1]==existing_versions
|
result={'applied':apply,'changes':changes,'counts':dict(collections.Counter(x['table'] for x in changes)),'new_units':new_units,'unit_ids':unitids,'report_template_version':{'id':new_version_id,'template_id':upgrade['template_id'],'version':upgrade['version']},'reports_updated':len(reports),'invalidated_pdf_attachments':old_attachments,'protected_hashes':protected,'verified_before_commit':True}
|
dump('result-'+('applied' if apply else 'preview')+'.json',result)
|
(P/('applied.sql' if apply else 'preview.sql')).write_text('START TRANSACTION;\n'+'\n'.join(sql_log)+'\nCOMMIT;\n',encoding='utf8')
|
if apply:c.commit()
|
else:c.rollback()
|
print(json.dumps({k:result[k] for k in ['applied','counts','new_units','reports_updated','invalidated_pdf_attachments','verified_before_commit']},ensure_ascii=False))
|
except:c.rollback();raise
|
finally:q.execute('SELECT RELEASE_LOCK(%s)',('qc-indicator-standard-optimize',))
|
c.close()
|
