"""Extract traceable QC requirements; never infer missing sampling numbers or limits."""
|
import json, re, sys, collections, csv, hashlib
|
from pathlib import Path
|
sys.stdout.reconfigure(encoding='utf-8')
|
ROOT = Path('scripts/sop-qc')
|
docs = json.loads((ROOT/'documents.json').read_text(encoding='utf-8'))
|
NAMES = '''水分 含水率 灰分 总灰分 酸不溶性灰分 酸不溶灰分 酸不溶性灰 硫酸灰分 干燥失重 蛋白质 粗蛋白 总氮 脂肪 粗脂肪 总糖 还原糖 枸杞多糖 粗多糖 多糖 银耳多糖含量 玉米须粉含量 葡萄粉含量 葡萄浓缩粉含量 总黄酮 膳食纤维含量 总膳食纤维含量 水解度 低聚肽含量 低聚肽 低聚糖 多肽 羟脯氨酸 淀粉酶抑制剂活性 植物凝集素活性 透明质酸钠含量 壳寡糖含量 相对分子质量 可溶性固形物含量 可溶性固形物 pH值 pH 黏度 透光率 溶解度 酸度 总酸 酸价 过氧化值 二氧化硫残留量 二氧化硫残留 二氧化硫 亚硝酸盐 黄曲霉毒素B1 黄曲霉毒素 黄曲霉毒素B2 黄曲霉毒素G1 黄曲霉毒素G2 赭曲霉毒素A 展青霉素 米酵菌酸 脱氧雪腐镰刀菌烯醇 无机砷 总砷 总汞 甲基汞 铅 镉 汞 铜 砷 铬 苯 磁性金属物 六六六 滴滴涕 甲基毒死蜱 溴氰菊酯 氯化钠 铵盐 谷氨酸 钾 氨基酸态氮占总氮百分比 表没食子儿茶素没食子酸酯 咖啡因 橙皮苷 川陈皮素 橘皮素 绿原酸 酚酸类总量 木犀草苷 花青素 黄芪甲苷 虫草素和腺苷 腺苷 唾液酸 结合态唾液酸 总虾青素含量 吡咯并喹啉醌二钠盐含量 茶氨酸含量 人参总皂苷含量 低聚果糖 葡萄糖＋果糖＋蔗糖含量 葡萄糖＋果糖＋蔗糖 菊粉含量 XOS2-4含量 XOS含量 水苏糖含量 果浆含量 橄榄果汁含量 山竹果汁含量 花香浓缩液含量 酸樱桃粉含量 百香果果粉含量 酸枣仁粉含量 百合粉含量 分离乳清蛋白粉含量 DHA藻油粉含量 粒度 颗粒度 百粒重 硬度 细度 杂质总量 杂质度 含杂率 无机杂质 有机杂质 一般杂质 不溶于水杂质 不溶物 不完善粒质量分数 不完善粒 碎米 霉变率 霉变粒 整体合格率 去柄率 泡水次数 泥沙 含沙量 干湿比 干物质 净含量 盐分 总不合格果百分率 菌落总数 大肠菌群 沙门氏菌 金黄色葡萄球菌 志贺氏菌 霉菌和酵母菌 霉菌和酵母 霉菌酵母菌 霉菌酵母 酵母菌 霉菌 致病菌 农药残留量 农药残留 农药最大残留 农残 兽药残留 多氯联苯 N-二甲基亚硝胺 挥发油 阿魏酸 菊苣酸 丹参酮类 丹酚酸B 氨基酸 特征多肽 特征图谱 浸出物 水溶性浸出物 醇溶性浸出物 三萜及甾醇 鉴别 显微鉴别 薄层鉴别 斯皮诺素、酸枣仁皂苷A 性状 色泽和状态 色泽 状态 形态 形状 组织形态 滋味、气味 滋、气味 滋气味 气味 滋味 杂质和缺陷 药屑及杂质 杂质 果形和果实大小 品质 损伤和缺陷 容许度 瑕疵检查 不合格判定 产品等级分类 级别分类 含量测定 感官要求 感官指标 感官检查 感官检验 外观'''.split()
|
NAMES.sort(key=len, reverse=True)
|
ALIASES = {'pH':'pH值','含水率':'水分','酸不溶灰分':'酸不溶性灰分','酸不溶性灰':'酸不溶性灰分','可溶性固形物含量':'可溶性固形物','霉菌和酵母':'霉菌和酵母菌','霉菌酵母菌':'霉菌和酵母菌','霉菌酵母':'霉菌和酵母菌','感官要求':'感官检查','感官指标':'感官检查','感官检验':'感官检查'}
|
GROUPS = {'感官':2,'理化':4,'成分':5,'微生物':4,'污染物':5,'农兽药残留':5}
|
SENSORY = set('性状 色泽和状态 色泽 状态 形态 形状 组织形态 滋味、气味 滋、气味 滋气味 气味 滋味 杂质和缺陷 药屑及杂质 杂质 果形和果实大小 品质 损伤和缺陷 容许度 瑕疵检查 不合格判定 产品等级分类 级别分类 感官检查 外观'.split())
|
MICRO = set('菌落总数 大肠菌群 沙门氏菌 金黄色葡萄球菌 志贺氏菌 霉菌和酵母菌 酵母菌 霉菌 致病菌'.split())
|
POLLUTANTS = set('铅 镉 汞 总汞 总砷 砷 铜 铬 无机砷 甲基汞 二氧化硫残留量 二氧化硫残留 二氧化硫 亚硝酸盐 黄曲霉毒素B1 黄曲霉毒素 黄曲霉毒素B2 黄曲霉毒素G1 黄曲霉毒素G2 赭曲霉毒素A 展青霉素 米酵菌酸 脱氧雪腐镰刀菌烯醇 N-二甲基亚硝胺 苯'.split())
|
HEADING = re.compile(r'^\s*(\d{1,2}(?:\s*[.．]\s*\d{1,2})*)(?:[.．、]|\s)?\s*([^\d\s].*)$')
|
SKIP = re.compile('试剂|仪器|操作步骤|原理|样品的稀释|精密度|重复性|标准溶液|计算公式|标准曲线|理论板数|检出限|定量限|温度|取药材|取本品|取试样|换算系数|供试品溶液|对照品溶液')
|
REF = re.compile(r'(?:GB\s*(?:/\s*T)?|QB\s*/\s*T|SN\s*/\s*T|Q\s*/|T\s*/)[A-Za-z0-9/ .—－-]+|通则\s*\d{4}|SOP-QC[-－][^《》。，；;\n ]+')
|
|
def label(text):
|
text=re.sub(r'\s+', '', text).replace('黄曲霉毒素Ｂ','黄曲霉毒素B')
|
if text.startswith('标志性成分指标花青素'): return '花青素'
|
if text.startswith('N-乙酰神经氨酸含量'): return 'N-乙酰神经氨酸含量'
|
if text.startswith('酚酸类'):return '酚酸类总量'
|
if text.startswith('苯') and '并芘' in text[:12]:return '苯并[a]芘'
|
if text.startswith('霉菌及酵母菌'): return '霉菌和酵母菌'
|
if text.startswith('相对分子质量小于'):
|
return re.split(r'[：:（(]',text)[0]
|
if text.startswith(('γ－氨基丁酸','γ-氨基丁酸')):return 'γ-氨基丁酸含量'
|
if text.startswith('未熟果、过熟果、破损果'):return '未熟果、过熟果、破损果、果柄残留、坏果'
|
if text.startswith('苋菜红、胭脂红'):return '禁用色素检查'
|
for name in NAMES:
|
if text.startswith(name):
|
rest=text[len(name):]
|
if not rest or re.match(r'^[（(：:，,。/／≤≥＜＞<>=\d ]|^(?:限量|含量|检测|测定|依据|照|不得|平板|依GB|应符合|以Hg|以As|以Pb|以Cr|及酵母|具有|取|本品|应有)',rest):
|
return ALIASES.get(name,name)
|
return None
|
|
def category(name):
|
if name in SENSORY:return '感官'
|
if name in MICRO:return '微生物'
|
if name in POLLUTANTS or name=='禁用色素检查':return '污染物'
|
if re.search('农残|农药|兽药|六六六|滴滴涕|多氯联苯|毒死蜱|菊酯',name):return '农兽药残留'
|
if re.search('糖|蛋白|氮|肽|酸|素|苷|酮|酚|含量|成分|皂苷|浸出物|挥发油|虾青|相对分子|测定|三萜|甾醇|XOS|水解度',name):return '成分'
|
return '理化'
|
|
def metadata(d):
|
code=title=''
|
for b in d['blocks']:
|
for row in b.get('table',[]):
|
if len(row)>=2 and row[0]=='文件编号' and not code:code=row[1]
|
if len(row)>=2 and row[0]=='文件名称' and not title and '规程' in row[1]:title=row[1]
|
match=re.search(r'(\d{3})',code or d['file'])
|
number=match[1] if match else hashlib.sha256(d['file'].encode()).hexdigest()[:8]
|
if not title:title=re.sub(r'^.*?[）)]\s*\d{3}(?:-\d{2})?\s*','',Path(d['file']).stem)
|
material=re.sub(r'(?:原料)?(?:检验标准操作规程|检验操作规程|检验标准操作|检验规程|质量标准|检验标准)$','',title).strip()
|
return code or 'SOP-QC-YL-'+number,title,material,number
|
|
def criteria(segment,name):
|
result=[];result_context=False
|
for original in segment.splitlines():
|
line=original.strip()
|
if not line:continue
|
m=HEADING.match(line) if ' | ' not in line else None
|
if m and not (re.match(r'^[\u4e00-\u9fff]',m[2]) or label(m[2])):m=None
|
text=m[2] if m else line
|
if name not in SENSORY and '本品' in text and not text.startswith('本品') and not re.search('结果|标准|限量',text[:15]):
|
text=text[text.index('本品'):]
|
if re.search('标准缓冲|电极|滴定溶液|实验用水|理论板数|校准|空白对照|稀释度|菌落数在|JJG 119',text) and not re.search('本品|结果指标|结果与判定',text):continue
|
if m and text in ('理化指标','理化性质','微生物指标','仪器','试剂','操作步骤','计算','结果计算'):continue
|
if re.search('结果.{0,2}(判定|判断)|结果指标|标准限量|^标准[：:]|^限量[：:]',text):result_context=True
|
elif m and SKIP.search(text[:30]):result_context=False
|
if re.search('精密度|重复性|算术平均|理论板数|标准溶液|标准曲线|检出限|定量限|允许误差|适宜稀释度|保留.{0,8}(小数|有效数字)',text):continue
|
normative=bool(re.search(r'≤|≥|＜|＞|不得|不应|应符合|应无|无异味|无正常|无肉眼|限量|限度要求|采样方案|n\s*[=＝]|本品含|本品按|标准[：:]|结果.{0,2}(判定|判断)',text))
|
if name in SENSORY and len(text)<=1000 and not re.match(r'^(目的|范围|职责|相关|文件|理化|微生物|污染物|引用标准)',text):normative=True
|
if result_context and re.search('[%％]|mg|g/|MPN|CFU|[≤≥<>＜＞～~]',text):normative=True
|
specific=bool(re.search(r'本品|本产品|样品.{0,10}(?:含量|限量)|标准[：:]|限量|限度要求|应符合|采样方案|n\s*[=＝]|结果.{0,2}(判定|判断)|'+re.escape(name)+r'.{0,120}(?:≤|≥|＜|＞|不得|符合)',text))
|
if len(segment)<700 and normative:specific=True
|
if re.search(r'本品.*(?:[%％]|mg/kg|g/100g|[~～]|\d[.\d]*\s*[-－]\s*\d)',text):normative=True
|
if normative and (name in SENSORY or result_context or specific) and (name in SENSORY or result_context or not SKIP.search(text[:30])):
|
result.append(text)
|
result=list(dict.fromkeys(result))
|
return '\n'.join(result)
|
|
schemes=[];exceptions=[]
|
for d in docs:
|
code,title,material,number=metadata(d)
|
entries=[]
|
for bi,b in enumerate(d['blocks']):
|
if 'p' in b:
|
for li,line in enumerate(b['p'].splitlines()):entries.append({'text':line,'block':bi,'line':li})
|
else:
|
for ri,row in enumerate(b['table']):entries.append({'text':' | '.join(row),'block':bi,'row':ri,'cells':row})
|
# Exclude administrative history and references from inspection requirements.
|
end=next((i for i,e in enumerate(entries) if re.match(r'^\s*(相关文件|相关记录|文件变更|文件修订)',e['text'])),len(entries))
|
entries=entries[:end]
|
candidates=[]
|
in_definitions=False
|
for i,e in enumerate(entries):
|
if 'cells' in e:continue
|
m=HEADING.match(e['text'])
|
bracket=re.match(r'^[【〔\[]([^】〕\]]+)[】〕\]]\s*(.*)',e['text'])
|
if m and '.' not in m[1]:in_definitions='术语和定义' in m[2]
|
if in_definitions:continue
|
unnumbered=not m and not bracket and (re.match(r'^[^：:]{1,50}[：:（(]',e['text']) or e['text'].strip() in NAMES)
|
if not m and not bracket and not unnumbered:continue
|
text=m[2] if m else bracket[1]+'：'+bracket[2] if bracket else e['text']
|
name=label(text)
|
depth=len(re.findall(r'\d+',m[1])) if m else 1
|
if unnumbered and not m and name and (re.search('计算|按下式|误差|标准溶液|允许',e['text']) or any(re.search(r'^\s*式中',x['text']) for x in entries[max(0,i-5):i])):continue
|
if name and depth<=3:candidates.append({'start':i,'section':m[1].replace(' ','') if m else bracket[1] if bracket else '正文','name':name,'heading':e['text']})
|
for j,c in enumerate(candidates):
|
stop=candidates[j+1]['start'] if j+1<len(candidates) else len(entries)
|
for k in range(c['start']+1,stop):
|
if re.match(r'^\s*\d+(?:\.\d+)*[.、 ]?\s*(?:理化指标|理化性质|微生物|重金属|污染物|农药残留|相关文件|相关记录)',entries[k]['text']):
|
stop=k;break
|
c['segment']='\n'.join(e['text'] for e in entries[c['start']:stop])
|
# Prefer specific sensory subitems and named compound assays to their umbrella heading.
|
selected=[]
|
for j,c in enumerate(candidates):
|
if c['name']=='感官检查' and any(x['section'].startswith(c['section']+'.') and x['name'] in SENSORY and x['name']!='感官检查' for x in candidates[j+1:j+7]):continue
|
if c['name']=='含量测定':
|
children=[x for x in candidates[j+1:] if x['section'].startswith(c['section']+'.')]
|
if children:continue
|
selected.append(c)
|
# Inline contaminant lists occasionally have no numbered subheadings.
|
for i,e in enumerate(entries):
|
if '重金属及有害元素' in e['text'] and re.search('不得|≤|≥',e['text']):
|
for atom in re.split('[；;]',e['text']):
|
for metal in ['铅','镉','汞','铜','砷']:
|
m=re.search(metal+r'[^；;,，]*?(?:不得过|≤)\s*\d+(?:\.\d+)?\s*mg\s*/\s*kg',atom)
|
if m and not any(c['name']==metal for c in selected):selected.append({'start':i,'section':'正文','name':metal,'heading':e['text'],'segment':m[0]})
|
# True specification tables only; dilution examples and detection limits are excluded.
|
for bi,b in enumerate(d['blocks']):
|
rows=b.get('table',[])
|
if not rows or not any('block' in e and e['block']==bi for e in entries):continue
|
header=' | '.join(rows[0])
|
if ('项目' in header and re.search('采样|限量|指标|要求',header) and not re.search('检出限|定量限',header)):
|
for row in rows[1:]:
|
if not row:continue
|
name=label(row[0]);text=header+'\n'+' | '.join(row)
|
if name and not any(c['name']==name for c in selected):selected.append({'start':next(i for i,e in enumerate(entries) if e['block']==bi),'section':'表格','name':name,'heading':row[0],'segment':text})
|
items=[]
|
for c in sorted(selected,key=lambda c:c['start']):
|
requirement=criteria(c['segment'],c['name'])
|
if not requirement and len(c['segment'])<=1000 and c['name'] in ('鉴别','显微鉴别','薄层鉴别','特征图谱'):
|
requirement=c['segment']
|
refs=list(dict.fromkeys(m[0].strip(' .') for m in REF.finditer(c['heading']+'\n'+requirement)))
|
if not refs:refs=list(dict.fromkeys(m[0].strip(' .') for m in REF.finditer(c['segment'])))[:3]
|
flags=[]
|
if not requirement:flags.append('规程未提取到独立判定标准，按原文执行')
|
if re.search(r'n\s*[=＝]|采样方案|级|厂家|供应商|项目号|型式|不得检出',requirement):flags.append('复合、定性或条件判定，保留原文')
|
# Deliberately preserve all requirements as TEXT: the numeric schema cannot express strict
|
# inequalities, n/c/m/M plans or grade-dependent criteria without changing their meaning.
|
item={'name':c['name'],'group':category(c['name']),'section':c['section'],'heading':c['heading'],'requirement':requirement,'method_references':refs,'source_segment':c['segment'],'result_type':3,'threshold_min':None,'threshold_max':None,'flags':flags}
|
items.append(item)
|
# Same section title may repeat because of source numbering mistakes; combine requirements.
|
merged={}
|
for item in items:
|
key=item['name']
|
if key in merged:
|
old=merged[key];old['requirement']='\n'.join(dict.fromkeys([s for s in [old['requirement'],item['requirement']] if s]));old['source_segment']+='\n'+item['source_segment'];old['method_references']=list(dict.fromkeys(old['method_references']+item['method_references']))
|
else:merged[key]=item
|
items=list(merged.values())
|
file_material=re.sub(r'^.*?[）)]\s*\d{3}(?:-\d{2})?\s*','',Path(d['file']).stem)
|
file_material=re.sub(r'(?:原料)?(?:检验标准操作规程|检验操作规程|检验标准操作|检验规程|质量标准|检验标准).*$','',file_material).strip()
|
flags=[]
|
normalized=lambda x:re.sub(r'\s+','',x).translate(str.maketrans({'（':'(', '）':')'}))
|
if file_material and normalized(file_material)!=normalized(material):
|
flags.append('文件名品名与封面品名不一致：'+file_material+' / '+material)
|
if any(not i['requirement'] for i in items):flags.append('部分检验项未提取到独立合格标准')
|
if any(re.search(r'[＜<]\s*0\s*[(（]',i['requirement']) for i in items):flags.append('原文含小于零的计数要求，需复核')
|
if not items:exceptions.append({'file':d['file'],'reason':'未识别检验项'})
|
schemes.append({'file':d['file'],'source_sha256':d['sha256'],'source_code':code,'code':'SOP-QC-YL-'+number,'title':title,'material_name':material,'name':material+'原料质检方案（'+number+'）','types':[1],'status':1 if flags else 0,'review_flags':flags,'items':items})
|
|
for target in ['local','remote']:
|
materials=json.loads((ROOT/'local-materials.json').read_text(encoding='utf-8')) if target=='local' else json.loads((ROOT/'remote-database.json').read_text(encoding='utf-8'))['materials']
|
normalize=lambda s:re.sub(r'\s+','',s).translate(str.maketrans({'（':'(', '）':')'}))
|
for s in schemes:
|
s[target+'_material_matches']=[{'id':int(m['id']),'code':m['code'],'name':m['name']} for m in materials if str(m['status'])=='0' and normalize(m['name'])==normalize(s['material_name']) and not any('不一致' in flag for flag in s['review_flags'])]
|
plan={'source':str(Path(r'C:\Users\云\Desktop\SOP-QC-YL')),'schemes':schemes,'exceptions':exceptions,'policy':'按原文提取，全部使用文本结果及人工判定，不推断采样数量或简化复合限值；仅精确名称匹配关联现有物料。'}
|
(ROOT/'inspection-plan.json').write_text(json.dumps(plan,ensure_ascii=False,indent=2),encoding='utf-8')
|
with (ROOT/'检验项提取清单.csv').open('w',encoding='utf-8-sig',newline='') as f:
|
w=csv.writer(f);w.writerow(['方案编号','物料名称','来源文件','来源版本','检验项','分类','条款','判定要求原文','方法标准','注意事项'])
|
for s in schemes:
|
for i in s['items']:w.writerow([s['code'],s['material_name'],s['file'],s['source_code'],i['name'],i['group'],i['section'],i['requirement'],'；'.join(i['method_references']),'；'.join(i['flags'])])
|
print(json.dumps({'schemes':len(schemes),'unique_items':len(set(i['name'] for s in schemes for i in s['items'])),'scheme_items':sum(len(s['items']) for s in schemes),'exceptions':exceptions,'empty_requirements':sum(not i['requirement'] for s in schemes for i in s['items']),'exact_matches':{t:sum(bool(s[t+'_material_matches']) for s in schemes) for t in ['local','remote']}},ensure_ascii=False))
|
for s in schemes:print(s['code'],s['material_name'],':',','.join(i['name'] for i in s['items']))
|
