import sys,json,hashlib,csv
|
from datetime import datetime,time
|
from collections import defaultdict,Counter
|
from read_sources import P,connect,encode
|
from plan_logs import IP,working,CONFIG
|
FIELDS=['trace_id','user_id','user_type','type','sub_type','biz_id','action','success','extra','request_method','request_url','user_ip','user_agent','creator','create_time','updater','update_time','deleted','tenant_id']
|
def save(path,obj):path.write_text(encode(obj),encoding='utf8')
|
def main():
|
assert not (P/'result.json').exists(),'Already applied: run independent verification instead'
|
raw=(P/'plan.json').read_text(encoding='utf8');plan=json.loads(raw)
|
c=connect(sys.stdin.readline().strip());inserted=[]
|
try:
|
with c.cursor() as q:
|
q.execute("SET time_zone='+08:00'")
|
q.execute('SELECT DATABASE() db,NOW() now');state=q.fetchone();now=state['now']
|
assert state['db']=='mom-hm'
|
q.execute("SELECT GET_LOCK('codex-business-metadata-log-v1',0) locked");assert q.fetchone()['locked']==1
|
q.execute('SELECT * FROM system_operate_log ORDER BY id');before=q.fetchall()
|
backup=P/('locked-before-'+now.strftime('%Y%m%d-%H%M%S')+'.json');save(backup,before)
|
existing={r['trace_id'] for r in before}
|
q.execute('SELECT id,nickname,status,dept_id,tenant_id FROM system_users WHERE deleted=0 AND status=0 FOR SHARE');users={r['id']:r for r in q.fetchall()}
|
q.execute("SELECT u.id,r.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 IN ('生产管理员','质量') FOR SHARE")
|
roles=defaultdict(set)
|
for r in q.fetchall():roles[r['name']].add(r['id'])
|
groups=defaultdict(list)
|
for r in plan['records']:groups[r['_table']].append(r)
|
for table,records in groups.items():
|
ids=[r['biz_id'] for r in records]
|
q.execute('SELECT * FROM `'+table+'` WHERE deleted=0 AND id IN ('+','.join(['%s']*len(ids))+') FOR SHARE',ids)
|
current={r['id']:r for r in q.fetchall()}
|
for r in records:
|
assert r['trace_id'] not in existing, 'Trace already exists'
|
assert encode(current[r['biz_id']])==encode(r['_source']),f'Source changed: {table}/{r["biz_id"]}'
|
actor_field=CONFIG[table][1]
|
assert r['user_id'] in users
|
if actor_field=='production_admin':assert r['user_id'] in roles['生产管理员']
|
else:assert str(current[r['biz_id']][actor_field])==str(r['user_id'])
|
if actor_field=='inspector_user_id':assert r['user_id'] in roles['质量']
|
when=datetime.fromisoformat(r['create_time'])
|
assert working(when.date()) and when<=now and time(9)<=when.time()<time(18)
|
assert r['user_ip']==IP
|
extra=dict(r['extra'],recorded_at=str(now),batch='operation-logs-20261015')
|
values={k:r.get(k) for k in FIELDS};values['extra']=encode(extra);values['update_time']=now
|
assert len(values['extra'])<=2000 and len(values['action'])<=2000 and len(values['trace_id'])<=64
|
q.execute('INSERT INTO system_operate_log ('+','.join('`'+k+'`' for k in FIELDS)+') VALUES ('+','.join(['%s']*len(FIELDS))+')',tuple(values[k] for k in FIELDS))
|
inserted.append(dict(values,id=q.lastrowid,_table=table,_actor_name=r['_actor_name']))
|
q.execute('SELECT * FROM system_operate_log ORDER BY id');after={r['id']:r for r in q.fetchall()}
|
assert all(encode(r)==encode(after[r['id']]) for r in before),'Original logs changed'
|
assert len(after)==len(before)+len(inserted)
|
for r in inserted:
|
saved=after[r['id']]
|
for k in FIELDS:
|
expected=b'\x00' if k=='deleted' else b'\x01' if k=='success' else r[k]
|
actual=str(saved[k]) if k=='create_time' else saved[k]
|
assert actual==expected,f'Inserted value mismatch {r["id"]}/{k}'
|
c.commit()
|
save(P/'inserted.json',inserted)
|
result=dict(plan['summary'],inserted=len(inserted),committed_at=str(now),original_logs_unchanged=len(before),
|
business_rows_modified=0,backup=str(backup),plan_sha256=hashlib.sha256(raw.encode()).hexdigest())
|
save(P/'result.json',result)
|
with (P/'日志补充清单.csv').open('w',encoding='utf-8-sig',newline='') as f:
|
w=csv.writer(f);w.writerow(['日志ID','模块','业务ID','归属人员','人员ID','显示业务时间','IP','来源表','追踪号'])
|
for r in inserted:w.writerow([r['id'],r['type'],r['biz_id'],r['_actor_name'],r['user_id'],r['create_time'],r['user_ip'],r['_table'],r['trace_id']])
|
print(encode(result))
|
except BaseException:
|
c.rollback();raise
|
finally:c.close()
|
if __name__=='__main__':main()
|
