import json,re,csv
|
from pathlib import Path
|
from datetime import datetime,timedelta
|
from decimal import Decimal
|
P=Path(__file__).parent
|
d=json.loads((P/'remote-snapshot.json').read_text(encoding='utf8'))
|
s=json.loads((P/'source.json').read_text(encoding='utf8'))[0]['rows']
|
def norm(s):return re.sub(r'[\s《》（）()\-]','',s or '')
|
def subject(s):
|
s=re.sub(r'^.*?\d{3}', '',s)
|
return norm(re.sub(r'(原料)?质检方案.*$|(原料)?质量标准.*$', '',s))
|
materials={str(m['code']):m for m in d['mdm_item']}
|
users={u['nickname'].strip():u for u in d['system_users']}
|
supplements=json.loads((P/'supplement-mapping.json').read_text(encoding='utf8')) if (P/'supplement-mapping.json').exists() else []
|
mapped={(x['source_item_code'],x['source_template']):x for x in supplements}
|
result=[]
|
for n,r in enumerate(s[1:],2):
|
code,name,item,vendor,batch,template,received,qualified,unqualified,check,inspector,inspect,receive,remark=r
|
if not any(r):continue
|
date=datetime.fromisoformat(inspect)
|
rec=date-timedelta(days=15)
|
problems=[]
|
if code=='Y202608006':problems.append('编号重复：按用户要求暂不导入这两行')
|
source_item=str(item)
|
mapping=mapped.get((source_item,template))
|
if mapping:item=mapping['item_code']
|
m=materials.get(str(item));u=users.get(inspector.strip())
|
if not m:problems.append('物料编码不存在')
|
if not u:problems.append('检测人员不存在')
|
candidates=[t for t in d['mes_qc_template'] if norm(t['name'])==norm(template)]
|
match='名称'
|
if not candidates:
|
# Template names begin with the subject, not a document code.
|
candidates=[t for t in d['mes_qc_template'] if norm(re.sub(r'原料质检方案.*$','',t['name']))==subject(template)]
|
match='内容名称'
|
if mapping:
|
candidates=[t for t in d['mes_qc_template'] if t['id']==mapping['template_id']]
|
match='补充核对映射'
|
if len(candidates)!=1:problems.append('质检方案不存在' if not candidates else '质检方案匹配不唯一')
|
t=candidates[0] if len(candidates)==1 else None
|
if t and (t['status']!=0 or '1' not in t['types'].split(',')):problems.append('方案未启用来料检验')
|
if Decimal(str(received))!=Decimal(str(qualified))+Decimal(str(unqualified)):problems.append('数量不平')
|
if m and t and not any(x['item_id']==m['id'] and x['template_id']==t['id'] for x in d['mes_qc_template_item']):problems.append('物料未关联该方案')
|
result.append(dict(row=n,code=code,name=name,item_code=item,item_id=m['id'] if m else None,vendor_name=vendor,vendor_batch=str(batch),source_template=template,template_id=t['id'] if t else None,template_name=t['name'] if t else None,template_match=match,received_quantity=received,qualified_quantity=qualified,unqualified_quantity=unqualified,check_quantity=received,check_result={'校验通过':1,'校验不通过':2}[check],inspector_user_id=u['id'] if u else None,inspect_date=str(date),receive_date=str(rec),remark=remark,status=0,issues=problems))
|
(P/'plan.json').write_text(json.dumps(result,ensure_ascii=False,indent=2),encoding='utf8')
|
with (P/'mapping-review.csv').open('w',encoding='utf-8-sig',newline='') as f:
|
q=csv.writer(f);q.writerow(['Excel行号','编号','物料编码','原质检方案','匹配方案','检测日期','来料日期','问题'])
|
for x in result:q.writerow([x['row'],x['code'],x['item_code'],x['source_template'],x['template_name'],x['inspect_date'],x['receive_date'],'；'.join(x['issues'])])
|
from collections import Counter
|
print('total',len(result),'ready',sum(not x['issues'] for x in result),'issues',Counter(p for x in result for p in x['issues']))
|
