import sys,json
|
from pathlib import Path
|
P=Path(__file__).parent
|
sys.path.insert(0,str(P.parent/'iqc-import-20261012/deps'))
|
import pymysql
|
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)
|
before=json.loads((P/'before-operation-log-append.json').read_text(encoding='utf8'))
|
expected=json.loads((P/'operation-log-supplement-records.json').read_text(encoding='utf8'))
|
correction_path=P/'mdm-material-log-role-assignment-changes.json'
|
corrections={r['id']:r for r in json.loads(correction_path.read_text(encoding='utf8'))} if correction_path.exists() else {}
|
with c.cursor() as q:
|
q.execute('SELECT * FROM system_operate_log ORDER BY id')
|
rows=q.fetchall(); byid={r['id']:r for r in rows}; bytrace={r['trace_id']:r for r in rows}
|
normalize=lambda r:json.loads(json.dumps(r,default=str,ensure_ascii=False))
|
for original in before:
|
current=normalize(byid[original['id']])
|
if original['id'] not in corrections:
|
assert current==original
|
else:
|
change=corrections[original['id']]
|
assert current['user_id']==change['after_user_id']==244
|
assert current['action']==change['action'] and current['extra']==change['extra']
|
evidence=json.loads(current['extra'])
|
assert evidence['original_request_user_id']==original['user_id']
|
assert evidence['original_action']==original['action'] and evidence['original_extra']==original['extra']
|
modified={'user_id','action','extra','updater','update_time'}
|
assert all(current[k]==v for k,v in original.items() if k not in modified)
|
for e in expected:
|
r=bytrace[e['trace_id']]
|
assert r['user_id']==e['user_id'] and str(r['create_time'])==e['create_time']
|
assert r['action']==e['action'] and json.loads(r['extra'])['recorded_at']==e['recorded_at']
|
assert r['request_method'] is None and r['request_url'] is None and r['user_ip'] is None
|
mdm=[r for r in rows if r['type']=='MDM 物料' and r['deleted'] in (0,False,b'\x00')]
|
if corrections:assert len(mdm)==17 and all(r['user_id']==244 for r in mdm)
|
print(json.dumps({'verified_committed_logs':len(expected),'original_logs_unchanged':len(before)-len(corrections),'original_logs_with_explicit_role_adjustment':len(corrections),'mdm_material_logs':len(mdm),'database_log_count':len(rows)},ensure_ascii=False))
|
c.close()
|
