"""Build a reviewable, source-backed migration without inventing measured values."""
|
import sys,json,re,csv,collections,copy
|
from pathlib import Path
|
from decimal import Decimal as D
|
sys.stdout.reconfigure(encoding='utf-8')
|
P=Path(__file__).parent;db=json.loads((P/'audit.json').read_text(encoding='utf8'))
|
def active(t):return [x for x in db[t]['rows'] if x.get('deleted') not in (1,"b'\\x01'")]
|
def dump(name,v):(P/name).write_text(json.dumps(v,ensure_ascii=False,indent=2,default=str),encoding='utf8')
|
inds={x['id']:copy.deepcopy(x) for x in active('mes_qc_indicator')};changes=[];review=[];units=set()
|
def change(t,row,v):
|
v={k:z for k,z in v.items() if row.get(k)!=z}
|
if v:changes.append({'table':t,'id':row['id'],'before':{k:row.get(k) for k in v},'after':v});row.update(v)
|
aliases={285:47,409:47}
|
allinds={x['id']:x for x in db['mes_qc_indicator']['rows']}
|
for old,new in aliases.items():assert allinds[old]['name']==inds[new]['name']=='大肠菌群'
|
specs=copy.deepcopy(active('mes_qc_template_indicator'))
|
for row in specs:
|
if row['indicator_id'] in aliases:
|
row['_source_unit']=allinds[row['indicator_id']]['unit_text']
|
change('mes_qc_template_indicator',row,{'indicator_id':aliases[row['indicator_id']]})
|
group_names={1:'感官与性状',2:'理化与物性',3:'营养与活性成分',4:'微生物',5:'污染物与重金属',6:'农兽药残留'}
|
for iid,name in group_names.items():
|
row=inds[iid];assert row['item_type']==2
|
change('mes_qc_indicator',row,{'name':name,'sort_order':iid*10,'result_type':None,'result_specification':None,'unit_text':None,'formula_text':'按本组各子项的适用标准分别检验和判定；分组标题不录入实测值。'})
|
micro=re.compile('菌落|大肠|沙门|葡萄球菌|志贺|霉菌|酵母|致病菌|微生物')
|
pollution=re.compile('^(?:铅|镉|汞|砷|铜|铬|总砷|总汞|无机砷|甲基汞)|二氧化硫|亚硝酸盐|黄曲霉|赭曲|展青霉|米酵菌|苯并|异丙醇|脱氧雪腐|磁性金属|重金属')
|
sensory=re.compile('性状|色泽|状态|形态|形状|外观|滋味|气味|香气|香味|感官|鉴别|标签|原料和|适用范围|品质|损伤|瑕疵|级别|容许度')
|
composition=re.compile('含量|蛋白|多糖|总糖|脂肪|黄酮|花青素|膳食|皂苷|氨基酸|虾青素|唾液酸|咖啡因|黄芪|虫草|腺苷|低聚|菊粉|茶氨酸|酚酸|木犀|橙皮|川陈|橘皮|绿原|丹参|阿魏|浸出|挥发油|DHA|总氮')
|
byind=collections.defaultdict(list)
|
for s in specs:byind[s['indicator_id']].append(s)
|
def requirement(s):
|
method=s.get('check_method') or ''
|
if method.startswith('要求：'):return method[3:].split('；方法：')[0].strip()
|
return (s.get('remark') or '').strip()
|
numeric_units={'%','mg/kg','g/kg','g/100g','mg/100g','g/100mL','mg/mL','CFU/g','CFU/mL','MPN/g','ug/kg','μg/kg','ppm','g/100粒','粒','泡','目','U/g','IU/mg','mPa·s','(°)·dm²·kg⁻¹'}
|
num=r'[+-]?\d+(?:\.\d+)?(?:[eE][+-]?\d+)?'
|
def normalized(text):
|
text=text.replace('％','%').replace('−','-').replace('℃','°C')
|
text=text.replace('ug/kg','μg/kg')
|
text=re.sub(r'('+num+r')\s*[×xX*]\s*10\s*\^\s*[（(]?([+-]?\d+)[)）]?',lambda m:format(D(m[1])*(D(10)**int(m[2])),'f'),text)
|
text=re.sub(r'10\s*\^\s*[（(]?([+-]?\d+)[)）]?',lambda m:format(D(10)**int(m[1]),'f'),text)
|
for a,b in [('不得超过','≤'),('不得过','≤'),('不超过','≤'),('不得大于','≤'),('不大于','≤'),('不得低于','≥'),('不低于','≥'),('不少于','≥'),('不小于','≥'),('>=','≥'),('<=','≤')]:text=text.replace(a,b)
|
return text
|
def limits(req):
|
s=normalized(req)
|
# Multiple grades, sampling plans, strict inequalities, censoring and procedural text need manual interpretation.
|
flags=[]
|
if re.search('标准品|标准物质|标准储备|试剂',s):return None,None,None,'试剂或标准品条件不能作为产品验收限值'
|
if re.search('特优|特级|甲级|乙级|一级|二级|三级|等级|n\\s*=|n＝|c\\s*=|c＝|鲜果|干果|不得检出|未检出|检出限|每批|分别|溶液',s) or len(s)>170:return None,None,None,'分级、抽样、检出条件或长文本需人工确认'
|
if re.search(r'(?<![≤≥])[<>](?!=)',s):return None,None,None,'严格大于/小于条件保留原文'
|
matches=re.findall(r'([≤≥])\s*('+num+')',s)
|
if len(matches)==1:
|
op,value=matches[0];v=D(value);return str(v),str(v) if op=='≥' else None,str(v) if op=='≤' else None,None
|
if len(matches)==2 and {x[0] for x in matches}=={'≤','≥'}:
|
low=D(next(x[1] for x in matches if x[0]=='≥'));high=D(next(x[1] for x in matches if x[0]=='≤'))
|
if low<=high:return str((low+high)/2),str(low),str(high),'标准值为允许区间中心参考值，不替代原文限值'
|
if len(matches)>1:return None,None,None,'多个限值不能合并为单一规格'
|
m=re.search('('+num+r')\s*[～~至]\s*('+num+')',s)
|
if not m:m=re.fullmatch(r'\s*('+num+r')\s*[-—]\s*('+num+r')\s*',s)
|
if m:
|
low,high=D(m[1]),D(m[2])
|
if low<=high:return str((low+high)/2),str(low),str(high),'标准值为允许区间中心参考值，不替代原文限值'
|
m=re.search('('+num+r')\s*[±]\s*('+num+')',s)
|
if m:
|
base,delta=D(m[1]),D(m[2]);return str(base),str(base-delta),str(base+delta),None
|
return None,None,None,'原文无可唯一提取的数值限值'
|
def unit_for(row,req):
|
s=normalized(req)
|
if re.search('标准品|标准物质|标准储备|试剂',s):return None
|
if len(s)>170 or re.search('溶液|仪器|原理|试剂',s):return row.get('unit_text') or None
|
if row['name'].startswith('粒度') and '/50g' in s:return '粒/50g'
|
for u in sorted(numeric_units,key=len,reverse=True):
|
if u in s:return 'μg/kg' if u=='ug/kg' else u
|
old=row.get('unit_text') or ''
|
if old:return 'μg/kg' if old=='ug/kg' else old
|
if re.search('pH|相对密度|折光指数|剪切性能|相对分子质量',row['name']):return '无量纲'
|
return None
|
for row in inds.values():
|
if row['item_type']==2:continue
|
name=row['name'];parent=row['parent_id']
|
if parent not in (407,):
|
parent=4 if micro.search(name) else 6 if re.search('农残|农药|兽药|六六六|滴滴涕|毒死蜱|氯联苯|菊酯',name) else 5 if pollution.search(name) else 1 if sensory.search(name) else 3 if composition.search(name) else 2
|
reqs=[requirement(s) for s in byind[row['id']]]
|
has_simple=any(limits(r)[0] is not None for r in reqs)
|
scalar_hint=bool(re.search('含量|花青素|水解度|透光率|溶解度|干湿比|净含量|硫酸灰分|硬度|总膳食|还原糖|pH|相对密度|折光指数|剪切性能',name))
|
numeric=(row.get('unit_text') in numeric_units or has_simple or scalar_hint) and (not sensory.search(name) or row.get('unit_text')=='%') and not re.search('续文|沙门氏菌|金黄色葡萄球菌|志贺氏菌|致病菌|农药残留|农残|兽药残留',name)
|
rt=2 if numeric and (micro.search(name) or name in ['泡水次数','粒度','颗粒度']) else 1 if numeric else 3
|
if rt==2 and (str(row.get('unit_text') or '').startswith('MPN') or any('MPN' in r for r in reqs) or any(str(s.get('_source_unit') or '').startswith('MPN') for s in byind[row['id']])):rt=1
|
if rt==2:
|
# Preserve recorded decimal measurements instead of rounding or dropping them during the type migration.
|
for detail in active('mes_qc_indicator_result_detail'):
|
if aliases.get(detail['indicator_id'],detail['indicator_id'])!=row['id'] or not detail['value']:continue
|
m=re.match(r'^\s*('+num+r')(?=\s|$)',detail['value'])
|
if m and D(m[1])!=D(m[1]).to_integral_value():rt=1;break
|
values={'parent_id':parent,'result_type':rt}
|
# Numeric fields have no dictionary/file specification; exact standard text belongs to each scheme.
|
if rt!=3:values['result_specification']=None
|
meaningful=[r for r in reqs if r and not r.startswith('10月8日至14日业务链')]
|
if meaningful and all(re.search('标准品|标准物质|标准储备|试剂',r) for r in meaningful):values['unit_text']=None
|
if rt in [1,2] and not row.get('unit_text'):
|
found={unit_for(row,r) for r in reqs}-{None}
|
if len(found)==1:values['unit_text']=next(iter(found))
|
change('mes_qc_indicator',row,values)
|
for gid in list(group_names)+[407]:
|
children=sorted([x for x in inds.values() if x['parent_id']==gid],key=lambda x:(x.get('sort_order') or 999999,x['id']))
|
for index,row in enumerate(children,1):change('mes_qc_indicator',row,{'sort_order':index*10})
|
for s in specs:
|
row=inds[s['indicator_id']];req=requirement(s)
|
if row['item_type']==2:
|
v={'standard_value':None,'threshold_min':None,'threshold_max':None,'unit_measure_id':None};change('mes_qc_template_indicator',s,v);continue
|
if row['result_type'] in [1,2]:
|
std,low,high,note=limits(req)
|
# For clone schemes, refer to the source indicator's original scheme when the clone remark is only provenance.
|
if not req or req.startswith('10月8日至14日业务链'):
|
candidates=[x for x in byind[row['id']] if x['id']!=s['id'] and requirement(x) and not requirement(x).startswith('10月8日至14日业务链')]
|
# Clone IDs 285/286 explicitly reference source template 12; raw 284 has no quantitative acceptance threshold.
|
if s['template_id'] in [285,286]:candidates=[x for x in candidates if x['template_id']==12]
|
else:candidates=[]
|
if len(candidates)==1:req=requirement(candidates[0]);std,low,high,note=limits(req)
|
unit=unit_for(row,req) or s.get('_source_unit')
|
if re.search('标准品|标准物质|标准储备|试剂',req):
|
unit=None
|
change('mes_qc_template_indicator',s,{'unit_measure_id':None,'remark':'规格待确认：原资料条目属于试剂或标准品条件，不作为产品验收标准；产品限值和检测单位须按适用产品标准确认。'})
|
if unit and '或' in unit:unit=None
|
if unit:units.add(unit)
|
v={'standard_value':std,'threshold_min':low,'threshold_max':high}
|
# Unit IDs are allocated by the transaction; symbolic unit names never enter database ID columns.
|
if unit:v['_unit_name']=unit
|
clean={k:v for k,v in v.items() if k!='_unit_name'};change('mes_qc_template_indicator',s,clean)
|
if unit:s['_unit_name']=unit
|
else:change('mes_qc_template_indicator',s,{'unit_measure_id':None})
|
review.append({'template_id':s['template_id'],'indicator_id':row['id'],'name':row['name'],'type':row['result_type'],'unit':unit,'standard':std,'lower':low,'upper':high,'requirement':req,'note':note})
|
else:
|
# Qualitative requirements must not carry a leftover quantitative limit.
|
change('mes_qc_template_indicator',s,{'standard_value':None,'threshold_min':None,'threshold_max':None,'unit_measure_id':None})
|
# Preserve the original acceptance wording and append interpretation information without changing provenance.
|
if row['result_type'] in [1,2] and note and std is None:
|
prior=s.get('remark') or '';suffix='；规格核对：'+note
|
if suffix not in prior and len(prior)+len(suffix)<=255:change('mes_qc_template_indicator',s,{'remark':prior+suffix})
|
unit_specs=[{'table':'mes_qc_template_indicator','id':s['id'],'unit_name':s['_unit_name']} for s in specs if s.get('_unit_name')]
|
specmap={(s['template_id'],s['indicator_id']):s for s in specs}
|
line_updates=[];line_state={}
|
for kind in ['iqc','ipqc','oqc','rqc']:
|
heads={x['id']:x for x in active('mes_qc_'+kind)};rr=copy.deepcopy(active('mes_qc_'+kind+'_line'))
|
for row in rr:
|
if row['indicator_id'] in aliases:change('mes_qc_'+kind+'_line',row,{'indicator_id':aliases[row['indicator_id']]})
|
head=heads.get(row[kind+'_id']);s=specmap.get((head['template_id'],row['indicator_id'])) if head else None
|
if s:
|
change('mes_qc_'+kind+'_line',row,{'standard_value':s['standard_value'],'min_threshold':s['threshold_min'],'max_threshold':s['threshold_max'],'check_method':s['check_method'],'remark':s['remark']})
|
if s.get('_unit_name'):unit_specs.append({'table':'mes_qc_'+kind+'_line','id':row['id'],'unit_name':s['_unit_name']});row['_unit_name']=s['_unit_name']
|
else:change('mes_qc_'+kind+'_line',row,{'unit_measure_id':None})
|
line_state[kind]=rr
|
details=copy.deepcopy(active('mes_qc_indicator_result_detail'));unconvertible=[];converted=[]
|
result_headers={x['id']:x for x in active('mes_qc_indicator_result')}
|
qc_headers={i:{x['id']:x for x in active('mes_qc_'+k)} for i,k in enumerate(['iqc','ipqc','oqc','rqc'],1)}
|
def numeric_value(value,rt,expected_unit):
|
if value is None or value.strip()=='':return value
|
s=normalized(value.strip()).replace(',','')
|
m=re.fullmatch('('+num+r')(?:\s*('+'|'.join(re.escape(u) for u in sorted(units,key=len,reverse=True))+'))?',s)
|
if not m:return None
|
if m[2] and expected_unit:
|
equivalent=lambda u:'%' if u in ['%','g/100g'] else u
|
if equivalent(m[2])!=equivalent(expected_unit):return None
|
v=D(m[1]);
|
if rt==2 and v!=v.to_integral_value():return None
|
return format(v,'f').rstrip('0').rstrip('.') if '.' in format(v,'f') else format(v,'f')
|
for row in details:
|
if row['indicator_id'] in aliases:change('mes_qc_indicator_result_detail',row,{'indicator_id':aliases[row['indicator_id']]})
|
ind=inds.get(row['indicator_id']);old=row['value']
|
if not ind or not old:continue
|
if ind['item_type']==2:
|
value=None;reason='分组标题不录入实测结果'
|
elif ind['result_type'] in [1,2]:
|
rh=result_headers[row['result_id']];qh=qc_headers[rh['qc_type']].get(rh['qc_id']);s=specmap.get((qh['template_id'],ind['id'])) if qh else None
|
expected_unit=s.get('_unit_name') if s else ind.get('unit_text')
|
value=numeric_value(old,ind['result_type'],expected_unit);reason='原值不含可转换的实测数值，待补测'
|
if value==old:continue
|
else:continue
|
if value is None:
|
note='原记录值：'+old+'；'+reason
|
prior=row['remark'] or ''
|
if len(prior)+len(note)+1<=255:newremark=(prior+'；' if prior else '')+note
|
else:newremark=prior # Full originals are also retained in the backup and review list.
|
change('mes_qc_indicator_result_detail',row,{'value':None,'remark':newremark});unconvertible.append({'id':row['id'],'indicator_id':ind['id'],'name':ind['name'],'old_value':old,'reason':reason})
|
else:change('mes_qc_indicator_result_detail',row,{'value':value});converted.append(row['id'])
|
plan={'changes':changes,'unit_assignments':unit_specs,'units':sorted(units),'indicators':list(inds.values()),'specs':specs,'line_state':line_state,'details':details,'standards_review':review,'unconvertible':unconvertible,'converted':converted}
|
dump('plan.json',plan)
|
with (P/'方案规格核对.csv').open('w',encoding='utf-8-sig',newline='') as f:
|
writer=csv.DictWriter(f,fieldnames=list(review[0]));writer.writeheader();writer.writerows(review)
|
dump('待补测结果.json',unconvertible)
|
print(json.dumps({'changes':dict(collections.Counter(x['table'] for x in changes)),'result_types':dict(collections.Counter(str(x['result_type']) for x in inds.values())),'units':sorted(units),'unconvertible':len(unconvertible),'converted':len(converted),'specs_review':len(review),'unique_numeric_specs':sum(x['standard'] is not None for x in review)},ensure_ascii=False))
|
