"""Assign only our reconstructed logs to a requested current role; retain source evidence."""
|
import sys,json,csv
|
from pathlib import Path
|
from collections import Counter
|
P=Path(__file__).parent
|
sys.path.insert(0,str(P.parent/'iqc-import-20261012/deps'))
|
import pymysql
|
encode=lambda r:json.dumps(r,ensure_ascii=False,default=str,sort_keys=True)
|
snapshot=json.loads((P/'before-log-production-role.json').read_text(encoding='utf8'))
|
records=json.loads((P/'operation-log-supplement-records.json').read_text(encoding='utf8'))
|
bytrace={r['trace_id']:r for r in records}
|
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)
|
changes=[]
|
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 u.id,u.nickname,r.id role_id,r.name role_name FROM system_users u JOIN system_user_role ur ON ur.user_id=u.id JOIN system_role r ON r.id=ur.role_id WHERE u.deleted=0 AND u.status=0 AND ur.deleted=0 AND r.deleted=0 AND r.status=0 AND r.name=%s AND u.tenant_id=0 AND r.tenant_id=0 FOR SHARE',('生产管理员',))
|
members={r['id']:r for r in q.fetchall()}
|
# Keep attribution to an existing source business account rather than distributing invented operators.
|
assert 244 in members, 'Existing business account is no longer an enabled production administrator'
|
member=members[244];name=member['nickname'].strip()
|
q.execute('SELECT * FROM system_operate_log ORDER BY id');before=q.fetchall()
|
(P/'before-operation-log-role-assignment.json').write_text(encode(before),encoding='utf8')
|
q.execute("SELECT * FROM system_operate_log WHERE trace_id LIKE 'BF-META-V1:%' FOR UPDATE")
|
logs=q.fetchall();snapshot_bytrace={r['trace_id']:r for r in snapshot['logs']}
|
assert len(logs)==len(records)==113
|
assert all(encode(r)==encode(snapshot_bytrace[r['trace_id']]) for r in logs), 'Supplemented logs changed since snapshot'
|
for log in logs:
|
if not log['type'].startswith(('MES ','MDM ')):continue
|
record=bytrace[log['trace_id']];extra=json.loads(log['extra'])
|
source_name=record['source_user_name'];source_id=extra.get('business_actor_user_id',log['user_id'])
|
source_role=extra.get('business_actor_role',record.get('business_role','建档人员'))
|
extra.update({'role_assignment':{'requested_role':'生产管理员','role_id':member['role_id'],
|
'assigned_user_id':member['id'],'assigned_user_name':name,'assigned_at':str(now),
|
'basis':'用户指定角色归属；采用既有业务建档账号；非历史请求操作者证明'},
|
'source_business_actor_user_id':source_id,'source_business_actor_name':source_name,
|
'source_business_actor_role':source_role,'attribution':'日志按当前生产管理员角色归属；保留原业务人员字段，非原始请求日志'})
|
date_precision=extra.get('event_time_precision','second')
|
when=str(log['create_time'])[:10] if date_precision=='day' else str(log['create_time'])
|
date_label='检测日期' if date_precision=='day' else '源建档时间'
|
action=(f'【历史数据补录／指定角色归属】{log["type"]}记录“{record["source_name"]}”（ID：{log["biz_id"]}）；'
|
f'日志归属人员：{name}（{member["id"]}），当前角色：生产管理员；'
|
f'源业务人员：{source_name}（{source_id}，{source_role}）；{date_label}：{when}。'
|
+('原数据仅有日期，日志零点不代表实际操作时刻。' if date_precision=='day' 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,update_time=%s WHERE id=%s AND trace_id=%s',
|
(member['id'],'历史数据补录（角色归属）',action,encode(extra),now,log['id'],log['trace_id']))
|
assert q.rowcount==1
|
record.update(user_id=member['id'],source_user_name=name,business_role='生产管理员（日志归属）',
|
actor_source_field='system_user_role.role_id='+str(member['role_id']),sub_type='历史数据补录（角色归属）',
|
action=action,extra=encode(extra),update_time=str(now),original_business_actor_name=source_name,
|
original_business_actor_user_id=source_id,original_business_actor_role=source_role)
|
changes.append({'log_id':log['id'],'trace_id':log['trace_id'],'module':log['type'],
|
'before_user_id':log['user_id'],'after_user_id':member['id'],'source_business_actor':source_name})
|
assert len(changes)==93
|
q.execute('SELECT * FROM system_operate_log ORDER BY id');after={r['id']:r for r in q.fetchall()}
|
changed_ids={r['log_id'] for r in changes}
|
assert all(encode(r)==encode(after[r['id']]) for r in before if r['id'] not in changed_ids)
|
assert all(after[r['log_id']]['user_id'] in members for r in changes)
|
assert all(encode(after[r['id']]['create_time'])==encode(r['create_time']) for r in before)
|
c.commit()
|
result={'updated_logs':len(changes),'changed_user_count':sum(r['before_user_id']!=r['after_user_id'] for r in changes),
|
'assigned_user_id':member['id'],'assigned_user_name':name,'role_id':member['role_id'],
|
'by_module':dict(Counter(r['module'] for r in changes)),'untouched_logs':len(before)-len(changes),
|
'source_business_data_modified':False,'log_times_unchanged':True,'assigned_at':str(now)}
|
(P/'operation-log-role-assignment-result.json').write_text(encode(result),encoding='utf8')
|
(P/'operation-log-role-assignment-changes.json').write_text(encode(changes),encoding='utf8')
|
(P/'operation-log-supplement-records.json').write_text(encode(records),encoding='utf8')
|
with (P/'操作日志补录清单.csv').open('w',encoding='utf-8-sig',newline='') as f:
|
w=csv.writer(f);w.writerow(['模块','业务ID','业务编号或名称','日志归属人员','归属人员ID','归属角色','归属依据','源业务人员','源业务人员ID','源业务角色','日志显示时间','时间精度','实际补录时间','日志类型','来源表','追踪编号'])
|
for r in records: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.get('original_business_actor_name',r['source_user_name']),r.get('original_business_actor_user_id',r['user_id']),r.get('original_business_actor_role',r['business_role']),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()
|
