1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
| import sys,json,hashlib
| 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
| b=json.loads((P/'before-apply.json').read_text(encoding='utf8'));r=json.loads((P/'result-applied.json').read_text(encoding='utf8'));assert r['applied']
| changes={(x['table'],x['id']):x['after'] for x in r['changes']}
| canon=lambda v: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()
| 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)
| with c.cursor() as q:
| q.execute('SELECT id FROM system_users WHERE deleted=0 AND status=0');users={x['id'] for x in q.fetchall()}
| total=0
| for t,oldrows in b['qc'].items():
| q.execute('SELECT * FROM '+t+' WHERE deleted=0 ORDER BY id');rows=q.fetchall();assert len(rows)==len(oldrows)
| for old,new in zip(oldrows,rows):
| assert canon(new)=={**old,**changes.get((t,old['id']),{})},(t,old['id'])
| assert new['inspect_date'] and new['inspector_user_id'] in users
| if (t,new['id']) in changes:assert new['inspect_date'].date()==new['out_date'].date()
| total+=len(rows)
| for t,h in b['source_hashes'].items():q.execute('SELECT * FROM '+t+' WHERE deleted=0 ORDER BY id');assert fingerprint(q.fetchall())==h
| result={'verified':True,'quality_records_checked':total,'completed_records':len(changes),'remaining_missing_dates':0,'remaining_missing_inspectors':0,'dates_match_shipping':True,'other_fields_and_source_documents_unchanged':True}
| (P/'verification.json').write_text(json.dumps(result,ensure_ascii=False,indent=2),encoding='utf8')
| print(json.dumps(result,ensure_ascii=False));c.close()
|
|
