"""Append explicitly labelled metadata reconstructions, never original request logs."""
|
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
|
|
MODULES = {
|
'crm_contract': 'CRM 合同',
|
'erp_sale_order': 'ERP 销售订单',
|
'mes_pro_mps': 'MES 生产主计划',
|
'mes_pro_route': 'MES 工艺路线',
|
'mes_pro_process': 'MES 工序',
|
'mes_md_workstation': 'MES 工作站',
|
'mes_pd_process_param': 'MES 生产参数',
|
'mes_pd_process_param_template': 'MES 参数模板',
|
'mes_md_item': 'MES 物料',
|
'mes_qc_iqc': 'MES 来料检验单',
|
'mes_qc_oqc': 'MES 出货检验单',
|
'mes_md_product_bom': 'MES 产品BOM',
|
}
|
snapshot = json.loads((P / 'before-log-sources.json').read_text(encoding='utf8'))
|
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,
|
connect_timeout=10, read_timeout=60, autocommit=False)
|
|
def serial(value):
|
return json.dumps(value, ensure_ascii=False, default=str, sort_keys=True)
|
|
def sample(rows, n=10):
|
# Ensure each real source account is represented, then spread across IDs.
|
selected = {}
|
for row in rows:
|
if not any(x['creator'] == row['creator'] for x in selected.values()):
|
selected[row['id']] = row
|
for i in range(min(n, len(rows))):
|
row = rows[round(i * (len(rows) - 1) / max(1, min(n, len(rows)) - 1))]
|
if len(selected) < n:
|
selected[row['id']] = row
|
for row in rows:
|
if len(selected) >= n:
|
break
|
selected[row['id']] = row
|
return sorted(selected.values(), key=lambda r: r['id'])
|
|
inserted, skipped = [], []
|
try:
|
with c.cursor() as q:
|
q.execute('SELECT DATABASE() db, NOW() now')
|
state = q.fetchone()
|
assert state['db'] == 'mom-hm'
|
now = state['now']
|
q.execute("SELECT GET_LOCK('codex-business-metadata-log-v1', 0) locked")
|
assert q.fetchone()['locked'] == 1, 'Another supplementation is running'
|
q.execute('SELECT id,nickname FROM system_users WHERE deleted=0')
|
users = {str(u['id']): u for u in q.fetchall()}
|
q.execute('SELECT * FROM system_operate_log ORDER BY id')
|
before = q.fetchall()
|
(P / 'before-operation-log-append.json').write_text(serial(before), encoding='utf8')
|
existing = {r['trace_id'] for r in before}
|
for table, module in MODULES.items():
|
q.execute('SHOW COLUMNS FROM `' + table + '`')
|
columns = {r['Field'] for r in q.fetchall()}
|
tenant_select = 'tenant_id' if 'tenant_id' in columns else '0 AS tenant_id'
|
eligible = []
|
for row in snapshot[table]:
|
if str(row['creator']) not in users:
|
continue
|
when = datetime.fromisoformat(row['create_time'])
|
if when > now or when > datetime.fromisoformat(row['update_time']):
|
skipped.append({'table': table, 'id': row['id'], 'reason': '时间异常'})
|
continue
|
eligible.append(row)
|
for row in sample(eligible):
|
trace = f"BF-META-V1:{table}:{row['id']}"
|
if trace in existing:
|
continue
|
q.execute('SELECT creator,create_time,update_time,' + tenant_select + ' FROM `' + table +
|
'` WHERE id=%s AND deleted=0 FOR SHARE', (row['id'],))
|
current = q.fetchone()
|
if not current or str(current['creator']) != str(row['creator']) or str(current['create_time']) != row['create_time']:
|
skipped.append({'table': table, 'id': row['id'], 'reason': '源记录已变化'})
|
continue
|
user = users[str(row['creator'])]
|
name = user['nickname'].strip()
|
label = row.get('no') or row.get('code') or row.get('name') or row.get('param_name') or row.get('template_name') or str(row['id'])
|
extra = serial({'reconstructed': True, 'evidence': 'record_creator_and_create_time',
|
'source_table': table, 'source_id': row['id'],
|
'source_creator': row['creator'], 'source_create_time': row['create_time'],
|
'source_update_time': str(current['update_time']),
|
'recorded_at': str(now), 'request_evidence_available': False,
|
'attribution': '源记录建档账号；不能据此认定真实操作人或请求'})
|
action = (f'【历史数据补录／元信息重建】{module}记录“{label}”（ID：{row["id"]}）；'
|
f'源建档账号：{name}（{user["id"]}），源建档时间：{row["create_time"]}。'
|
f'补录时间：{now}。仅还原记录元信息，非原始操作请求日志；不推定审核、修改或其他操作。')
|
assert len(trace) <= 64 and len(action) <= 2000 and len(extra) <= 2000
|
inserted.append(dict(trace_id=trace, user_id=user['id'], user_type=2,
|
type=module, sub_type='历史数据补录（建档元信息）', biz_id=row['id'],
|
action=action, success=1, extra=extra, request_method=None,
|
request_url=None, user_ip=None, user_agent=None, creator='1',
|
create_time=row['create_time'], updater='1', update_time=now,
|
deleted=0, tenant_id=current['tenant_id'], source_table=table,
|
source_name=str(label), source_user_name=name, recorded_at=str(now)))
|
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']
|
if inserted:
|
sql = 'INSERT INTO system_operate_log (' + ','.join('`'+f+'`' for f in fields) + ') VALUES (' + ','.join('%s' for _ in fields) + ')'
|
q.executemany(sql, [tuple(r[f] for f in fields) for r in inserted])
|
q.execute("SELECT * FROM system_operate_log WHERE trace_id LIKE 'BF-META-V1:%'")
|
reconstructed = q.fetchall()
|
assert len({r['trace_id'] for r in reconstructed}) == len(reconstructed)
|
assert all(r['sub_type'] == '历史数据补录（建档元信息）' and json.loads(r['extra'])['reconstructed'] for r in reconstructed)
|
assert all(r['request_url'] is None and r['user_ip'] is None and r['user_agent'] is None for r in reconstructed)
|
q.execute('SELECT * FROM system_operate_log ORDER BY id')
|
after = {r['id']: r for r in q.fetchall()}
|
assert all(serial(r) == serial(after[r['id']]) for r in before), 'Original logs changed'
|
c.commit()
|
result = {'inserted': len(inserted), 'by_module': dict(Counter(r['type'] for r in inserted)),
|
'by_source_account': dict(Counter(r['source_user_name'] for r in inserted)),
|
'historical_start': min((r['create_time'] for r in inserted), default=None),
|
'historical_end': max((r['create_time'] for r in inserted), default=None),
|
'recorded_at': str(now), 'original_logs_verified_unchanged': len(before),
|
'skipped': skipped, 'reconstruction_total': len(reconstructed)}
|
(P / 'operation-log-supplement-result.json').write_text(serial(result), encoding='utf8')
|
(P / 'operation-log-supplement-records.json').write_text(serial(inserted), 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 inserted:
|
w.writerow([r['type'],r['biz_id'],r['source_name'],r['source_user_name'],r['user_id'],r['create_time'],r['recorded_at'],r['sub_type'],r['source_table'],r['trace_id']])
|
print(serial(result))
|
finally:
|
c.close()
|
