"""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()