from pathlib import Path
|
from datetime import datetime,date,time
|
from collections import Counter
|
import json,hashlib
|
from read_sources import encode
|
P=Path(__file__).parent
|
IP='118.213.155.222'
|
# Fixed weekends remain excluded, including official make-up workdays.
|
HOLIDAYS={date(2026,9,d) for d in range(25,28)}|{date(2026,10,d) for d in range(1,8)}
|
def working(d):return d.weekday()<5 and d not in HOLIDAYS
|
# module, actor source, event source, representative record cap
|
CONFIG={
|
'crm_customer':('CRM 客户','owner_user_id','create_time',30),
|
'crm_business':('CRM 商机','owner_user_id','create_time',30),
|
'crm_sale_quotation':('CRM 销售报价','owner_user_id','quotation_time',100),
|
'crm_contract':('CRM 合同','owner_user_id','order_date',100),
|
'crm_invoice':('CRM 开票','owner_user_id','invoice_time',100),
|
'crm_receivable':('CRM 回款','owner_user_id','return_time',100),
|
'crm_receivable_plan':('CRM 回款计划','owner_user_id','create_time',100),
|
'erp_sale_order':('ERP 销售订单','sale_user_id','order_time',100),
|
'mes_pro_mps':('MES 生产主计划','production_admin','create_time',100),
|
'mes_pro_work_order':('MES 生产工单','production_admin','create_time',100),
|
'mes_pro_task':('MES 生产任务','production_admin','create_time',60),
|
'mes_pro_feedback':('MES 生产报工','feedback_user_id','feedback_time',60),
|
'mes_qc_iqc':('MES 来料检验单','inspector_user_id','inspect_date',100),
|
'mes_qc_ipqc':('MES 过程检验单','inspector_user_id','inspect_date',100),
|
'mes_qc_oqc':('MES 出货检验单','inspector_user_id','inspect_date',100),
|
'mes_wm_sales_notice':('MES 发货通知','creator','sales_date',100),
|
'mes_wm_product_sales':('MES 销售出库','creator','sales_date',100),
|
'mes_wm_product_produce':('MES 生产入库','production_admin','produce_date',100),
|
'mes_wm_product_issue':('MES 生产领料','production_admin','issue_date',100),
|
'mes_wm_misc_receipt':('MES 其他入库','creator','receipt_date',100),
|
'mdm_item':('MDM 物料','production_admin','create_time',30),
|
'mes_md_item':('MES 物料','production_admin','create_time',30),
|
'mes_md_product_bom':('MES 产品BOM','production_admin','create_time',30),
|
'mes_pro_process':('MES 工序','production_admin','create_time',30),
|
'mes_pro_route':('MES 工艺路线','production_admin','create_time',30),
|
'mes_md_workstation':('MES 工作站','production_admin','create_time',30),
|
'mes_pd_process_param':('MES 生产参数','production_admin','create_time',30),
|
'mes_pd_process_param_template':('MES 参数模板','production_admin','create_time',30),
|
'mes_dv_repair':('MES 设备维修','accepted_user_id','require_date',100),
|
'mes_dv_check_record':('MES 设备点检','user_id','check_time',40),
|
'mes_dv_mainten_record':('MES 设备保养','user_id','mainten_time',40),
|
}
|
def plan(snapshot):
|
now=datetime.fromisoformat(snapshot['state']['now'])
|
users={str(x['id']):x for x in snapshot['users'] if x['status']==0}
|
production={str(x['id']) for x in snapshot['roles'] if x['role_name']=='生产管理员'}
|
quality={str(x['id']) for x in snapshot['roles'] if x['role_name']=='质量'}
|
assert '244' in production and '244' in users
|
known=set(); traces={x['trace_id'] for x in snapshot['logs']}
|
for log in snapshot['logs']:
|
if log['deleted']!="b'\\x00'":continue
|
try:
|
extra=json.loads(log['extra'] or '{}')
|
if isinstance(extra,dict) and extra.get('source_table'):
|
known.add((extra['source_table'],int(extra.get('source_id',log['biz_id']))))
|
except (ValueError,TypeError):pass
|
# Match only creation/recording logs; an unrelated update is not a duplicate.
|
if any(x in log['sub_type'] for x in ['创建','新增','新建','添加','历史数据补录','业务记录补充']):
|
for tab,cfg in CONFIG.items():
|
if log['type']==cfg[0]:known.add((tab,log['biz_id']))
|
rows=[];skips=[];limited={}
|
for tab,(module,actor_field,time_field,cap) in CONFIG.items():
|
candidates=[]
|
for source in snapshot['tables'].get(tab,{}).get('rows',[]):
|
key=(tab,source['id']);reason=None
|
when=source.get(time_field)
|
actor='244' if actor_field=='production_admin' else str(source.get(actor_field))
|
trace=f'BF-20261015:{tab}:{source["id"]}'
|
if key in known or trace in traces:reason='已有对应业务日志'
|
elif actor not in users:reason='业务人员缺失或停用'
|
elif actor_field=='inspector_user_id' and actor not in quality:reason='检验人员无有效质量角色'
|
elif not when:reason='缺少实际业务日期'
|
else:
|
when=datetime.fromisoformat(when)
|
if when.date()<date(2026,9,1):reason='不在九月至今范围'
|
elif when>now:reason='未来业务日期'
|
elif not working(when.date()):reason='业务日期为双休或法定假期'
|
if reason:
|
skips.append({'table':tab,'id':source['id'],'reason':reason});continue
|
# Keep exact business timestamps. Only midnight/date-only values receive a
|
# documented display time on the same day, never a fabricated request time.
|
date_only=when.time()==time(0)
|
if date_only:
|
clock=time(9+(source['id']%8),source['id']%60,source['id']%59)
|
when=datetime.combine(when.date(),clock)
|
if when>now:
|
skips.append({'table':tab,'id':source['id'],'reason':'当日显示时刻尚未到达'});continue
|
if not time(9)<=when.time()<time(18):
|
skips.append({'table':tab,'id':source['id'],'reason':'业务时间不在九至十八点范围'});continue
|
name=str(source.get('no') or source.get('code') or source.get('name') or source.get('param_name') or source.get('template_name') or source['id'])
|
user=users[actor]
|
role='生产管理员' if actor_field=='production_admin' else actor_field
|
extra={'reconstructed':True,'source_table':tab,'source_id':source['id'],
|
'source_creator':source.get('creator'),'actor_source_field':actor_field,
|
'event_time_source_field':time_field,'source_event_time':source[time_field],
|
'event_time_precision':'day' if date_only else 'second',
|
'display_time_assigned':date_only,'ip_source':'user_specified','request_evidence_available':False,
|
'attribution':'按业务字段或指定角色归属，非原始请求证据'}
|
action=f'{module}记录“{name}”（ID：{source["id"]}），归属人员：{user["nickname"].strip()}；业务日期：{source[time_field]}。'
|
if actor_field=='production_admin':action+='按生产管理员角色归属。'
|
action+='依据现有业务记录补充，非原始请求日志。'
|
candidates.append({'trace_id':trace,'user_id':user['id'],'user_type':2,'type':module,
|
'sub_type':'业务记录补充','biz_id':source['id'],'action':action,'success':1,
|
'extra':extra,'request_method':None,'request_url':None,'user_ip':IP,'user_agent':None,
|
'creator':str(user['id']),'create_time':str(when),'updater':str(user['id']),
|
'deleted':0,'tenant_id':source.get('tenant_id',user['tenant_id']),
|
'_source':source,'_table':tab,'_actor_name':user['nickname'].strip(), '_actor_role':role})
|
candidates.sort(key=lambda r:(r['create_time'],r['biz_id']))
|
if len(candidates)>cap:
|
# Spread across the full period, with newest and oldest records represented.
|
indices={round(i*(len(candidates)-1)/(cap-1)) for i in range(cap)}
|
selected=[r for i,r in enumerate(candidates) if i in indices]
|
limited[tab]={'eligible':len(candidates),'selected':len(selected)}
|
else:selected=candidates
|
rows+=selected
|
result={'target':'36.213.150.237:10000/mom-hm','cutoff':str(now),'records':rows,'skips':skips,'representative_limits':limited,
|
'summary':{'insert_count':len(rows),'by_module':dict(Counter(x['type'] for x in rows)),
|
'by_actor':dict(Counter(x['_actor_name'] for x in rows)),
|
'skip_reasons':dict(Counter(x['reason'] for x in skips)),
|
'earliest':min((x['create_time'] for x in rows),default=None),'latest':max((x['create_time'] for x in rows),default=None),
|
'display_times_for_date_only':sum(x['extra']['display_time_assigned'] for x in rows),'ip':IP}}
|
return result
|
if __name__=='__main__':
|
s=json.loads((P/'snapshot.json').read_text(encoding='utf8'));p=plan(s)
|
(P/'plan.json').write_text(encode(p),encoding='utf8')
|
print(encode(p['summary']))
|
