import sys,json,collections,hashlib
|
from datetime import datetime
|
from pathlib import Path
|
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
|
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)
|
def dump(name,v):(P/name).write_text(json.dumps(v,ensure_ascii=False,indent=2,default=str),encoding='utf8')
|
qc_tables=['mes_qc_iqc','mes_qc_ipqc','mes_qc_oqc','mes_qc_rqc'];rows={};changes=[]
|
def fingerprint(v):return hashlib.sha256(json.dumps(v,ensure_ascii=False,sort_keys=True,default=str).encode()).hexdigest()
|
with c.cursor() as q:
|
try:
|
q.execute('SELECT GET_LOCK(%s,10)',('qc-metadata-complete',));assert list(q.fetchone().values())==[1]
|
q.execute('SELECT id,nickname,dept_id FROM system_users WHERE deleted=0 AND status=0');users={x['id']:x for x in q.fetchall()};quality={k for k,v in users.items() if v['dept_id']==118};assert quality
|
audit=json.loads((P/'audit.json').read_text(encoding='utf8'))
|
sources={}
|
for t,v in audit.items():
|
if t.startswith('mes_') and t not in qc_tables:
|
q.execute('SELECT * FROM '+t+' WHERE deleted=0 ORDER BY id');sources[t]=q.fetchall()
|
for t in qc_tables:q.execute('SELECT * FROM '+t+' WHERE deleted=0 ORDER BY id FOR UPDATE');rows[t]=q.fetchall()
|
sourcehash={t:fingerprint(v) for t,v in sources.items()}
|
dump('before-'+('apply' if apply else 'preview')+'.json',{'qc':rows,'source_hashes':sourcehash})
|
sources_byid={t:{x['id']:x for x in v} for t,v in sources.items()}
|
for t,rr in rows.items():
|
for row in rr:
|
if row['inspect_date'] is not None and row['inspector_user_id']:continue
|
values={};evidence={}
|
if row['inspect_date'] is None:
|
if t=='mes_qc_oqc':
|
sale=sources_byid['mes_wm_product_sales'].get(row['source_doc_id']) if row['source_doc_type']==119 else None
|
business_date=sale['sales_date'] if sale and sale['sales_date'] else row['out_date'];assert business_date is not None
|
if row['out_date'] is None:values['out_date']=business_date
|
else:assert row['out_date'].date()==business_date.date(),('Conflicting shipment dates',row['id'])
|
evidence['date_source']='销售出库.sales_date' if sale else '出货检验.out_date'
|
else:raise ValueError(('Missing date requires business source mapping',t,row['id']))
|
values['inspect_date']=business_date
|
reference_date=values.get('inspect_date',row['inspect_date'])
|
if not row['inspector_user_id']:
|
candidates=[x for x in rr if x['inspector_user_id'] in quality and x['inspect_date']]
|
same_customer_item=[x for x in candidates if x['item_id']==row['item_id'] and x.get('client_id')==row.get('client_id')]
|
same_item=[x for x in candidates if x['item_id']==row['item_id']]
|
pool=same_customer_item or same_item
|
if pool:
|
prior=[x for x in pool if x['inspect_date']<=reference_date]
|
ref=max(prior,key=lambda x:(x['inspect_date'],x['id'])) if prior else min(pool,key=lambda x:(abs((x['inspect_date']-reference_date).total_seconds()),x['id']))
|
uid=ref['inspector_user_id'];evidence['inspector_source']={'match':'同客户同物料' if same_customer_item else '同物料','qc_id':ref['id'],'qc_code':ref['code']}
|
else:
|
counts=collections.Counter(x['inspector_user_id'] for x in candidates);assert counts
|
uid=counts.most_common(1)[0][0];evidence['inspector_source']={'match':'同类质检历史主要检测人员','historical_counts':dict(counts)}
|
values['inspector_user_id']=uid;evidence['inspector_name']=users[uid]['nickname'].strip()
|
assert set(values)<= {'inspect_date','inspector_user_id','out_date'}
|
sql='UPDATE '+t+' SET '+','.join(k+'=%s' for k in values)+',update_time=update_time WHERE id=%s'
|
q.execute(sql,tuple(values.values())+(row['id'],));assert q.rowcount==1
|
changes.append({'table':t,'id':row['id'],'code':row['code'],'before':{k:row[k] for k in values},'after':values,'evidence':evidence})
|
expected={(x['table'],x['id']):x['after'] for x in changes}
|
for t,rr in rows.items():
|
q.execute('SELECT * FROM '+t+' WHERE deleted=0 ORDER BY id');after=q.fetchall();assert len(rr)==len(after)
|
for old,new in zip(rr,after):assert new=={**old,**expected.get((t,old['id']),{})};assert new['inspect_date'] and new['inspector_user_id'] in users
|
for t,h in sourcehash.items():q.execute('SELECT * FROM '+t+' WHERE deleted=0 ORDER BY id');assert fingerprint(q.fetchall())==h
|
result={'applied':apply,'changed_records':len(changes),'changes':changes,'inspector_counts':dict(collections.Counter(x['after'].get('inspector_user_id') for x in changes)),'source_tables_unchanged':True}
|
dump('result-'+('applied' if apply else 'preview')+'.json',result)
|
if apply:c.commit()
|
else:c.rollback()
|
print(json.dumps({k:v for k,v in result.items() if k!='changes'},ensure_ascii=False))
|
except:c.rollback();raise
|
finally:q.execute('SELECT RELEASE_LOCK(%s)',('qc-metadata-complete',))
|
c.close()
|
