import sys,json,csv
|
from pathlib import Path
|
from datetime import datetime
|
from collections import Counter
|
P=Path(__file__).parent
|
sys.path.insert(0,str(P.parent/'iqc-import-20261012/deps'))
|
import pymysql
|
s=json.loads((P/'before-log-business-actors.json').read_text(encoding='utf8'))
|
old_records=json.loads((P/'operation-log-supplement-records.json').read_text(encoding='utf8'))
|
record_by_trace={r['trace_id']:r for r in old_records}
|
roles={'mes_qc_iqc':('inspector_user_id','检验人员'),
|
'mes_qc_oqc':('inspector_user_id','检验人员'),
|
'crm_contract':('owner_user_id','合同负责人'),
|
'erp_sale_order':('sale_user_id','销售员')}
|
encode=lambda r:json.dumps(r,ensure_ascii=False,default=str,sort_keys=True)
|
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)
|
result_rows=[];changes=[];unresolved=[]
|
try:
|
with c.cursor() as q:
|
q.execute("SELECT GET_LOCK('codex-business-metadata-log-v1',0) locked")
|
assert q.fetchone()['locked']==1
|
q.execute('SELECT NOW() now');now=q.fetchone()['now']
|
q.execute('SELECT id,nickname FROM system_users WHERE deleted=0')
|
users={str(r['id']):r for r in q.fetchall()}
|
q.execute("SELECT * FROM system_operate_log WHERE trace_id LIKE 'BF-META-V1:%' FOR UPDATE")
|
logs=q.fetchall();by_trace={r['trace_id']:r for r in logs}
|
assert len(logs)==len(s['logs'])==len(old_records)==113
|
assert all(encode(by_trace[r['trace_id']])==encode(r) for r in s['logs']), 'Supplemented logs changed since snapshot'
|
q.execute('SELECT * FROM system_operate_log ORDER BY id');before=q.fetchall()
|
(P/'before-operation-log-actor-correction.json').write_text(encode(before),encoding='utf8')
|
for log in logs:
|
trace=log['trace_id'];old=record_by_trace[trace];extra=json.loads(log['extra'])
|
table=extra['source_table'];biz_id=log['biz_id']
|
actor_field,role=roles.get(table,('creator','建档人员（业务表未设专用人员字段）'))
|
time_field='inspect_date' if table in ['mes_qc_iqc','mes_qc_oqc'] else 'create_time'
|
q.execute('SELECT `'+actor_field+'` actor, `'+time_field+'` event_time FROM `'+table+'` WHERE id=%s AND deleted=0 FOR SHARE',(biz_id,))
|
source=q.fetchone()
|
if not source or str(source['actor']) not in users or source['event_time'] is None:
|
unresolved.append({'trace_id':trace,'source_table':table,'biz_id':biz_id,'reason':'业务人员或时间无法确认'})
|
continue
|
user=users[str(source['actor'])];name=user['nickname'].strip();event=source['event_time']
|
if not isinstance(event,datetime):event=datetime.combine(event,datetime.min.time())
|
assert event<=now
|
is_qc=time_field=='inspect_date'
|
when=event.strftime('%Y-%m-%d') if is_qc else str(event)
|
subtype='历史数据补录（检验记录）' if is_qc else '历史数据补录（业务归属）'
|
extra.update({'reconstructed':True,'evidence':'business_record_role_and_date' if is_qc else 'business_record_actor_field',
|
'actor_source_field':actor_field,'business_actor_user_id':user['id'],'business_actor_role':role,
|
'event_time_source_field':time_field,'event_time_precision':'day' if is_qc else 'second',
|
'corrected_at':str(now),'attribution':'依据业务人员字段归属，不证明原始请求操作者'})
|
action=(f'【历史数据补录／业务记录归属】{log["type"]}记录“{old["source_name"]}”（ID：{biz_id}）；'
|
f'{role}：{name}（{user["id"]}），依据字段：{actor_field}；'
|
f'{"检测日期" if is_qc else "建档时间"}：{when}。'
|
+('原数据仅有日期，日志零点为日期展示值，不代表实际操作时刻。' if is_qc else '')
|
+f'实际补录时间：{extra["recorded_at"]}，归属修正时间：{now}。仅依据业务记录补录，非原始操作请求日志。')
|
assert len(action)<=2000 and len(encode(extra))<=2000
|
q.execute('UPDATE system_operate_log SET user_id=%s,sub_type=%s,action=%s,extra=%s,create_time=%s,update_time=%s WHERE id=%s AND trace_id=%s',
|
(user['id'],subtype,action,encode(extra),event,now,log['id'],trace))
|
assert q.rowcount==1
|
updated=dict(old,user_id=user['id'],sub_type=subtype,action=action,extra=encode(extra),create_time=str(event),update_time=str(now),source_user_name=name,business_role=role,actor_source_field=actor_field,event_time_precision=extra['event_time_precision'])
|
result_rows.append(updated)
|
changes.append({'log_id':log['id'],'business_table':table,'business_id':biz_id,'before_user_id':log['user_id'],'after_user_id':user['id'],'actor':name,'role':role,'actor_source_field':actor_field,'before_time':str(log['create_time']),'after_time':str(event),'date_precision':extra['event_time_precision']})
|
assert not unresolved,encode(unresolved)
|
q.execute('SELECT * FROM system_operate_log ORDER BY id');after={r['id']:r for r in q.fetchall()}
|
target_ids={r['id'] for r in logs}
|
assert all(encode(r)==encode(after[r['id']]) for r in before if r['id'] not in target_ids), 'Original logs changed'
|
assert all(after[x['log_id']]['user_id']==x['after_user_id'] and str(after[x['log_id']]['create_time'])==x['after_time'] for x in changes)
|
c.commit()
|
result={'updated_supplemented_logs':len(changes),'changed_actor_count':sum(r['before_user_id']!=r['after_user_id'] for r in changes),
|
'changed_time_count':sum(r['before_time']!=r['after_time'] for r in changes),'by_business_actor':dict(Counter(r['actor'] for r in changes)),
|
'by_role':dict(Counter(r['role'] for r in changes)),'unresolved':unresolved,'original_logs_unchanged':len(before)-len(logs),'corrected_at':str(now)}
|
(P/'operation-log-actor-correction-result.json').write_text(encode(result),encoding='utf8')
|
(P/'operation-log-actor-corrections.json').write_text(encode(changes),encoding='utf8')
|
(P/'operation-log-supplement-records.json').write_text(encode(result_rows),encoding='utf8')
|
with (P/'操作日志补录清单.csv').open('w',encoding='utf-8-sig',newline='') as f:
|
w=csv.writer(f);w.writerow(['模块','业务ID','业务编号或名称','业务人员','人员ID','人员角色','人员来源字段','日志显示时间','时间精度','实际补录时间','日志类型','来源表','追踪编号'])
|
for r in result_rows:w.writerow([r['type'],r['biz_id'],r['source_name'],r['source_user_name'],r['user_id'],r['business_role'],r['actor_source_field'],r['create_time'],r['event_time_precision'],r['recorded_at'],r['sub_type'],r['source_table'],r['trace_id']])
|
print(encode(result))
|
finally:c.close()
|
