import sys,json,hashlib,collections,csv
|
from pathlib import Path
|
from decimal import Decimal as D
|
sys.stdout.reconfigure(encoding='utf-8')
|
sys.path.insert(0,str(Path('scripts/iqc-import-20261012/deps').resolve()))
|
import pymysql
|
P=Path(__file__).parent
|
b=json.loads((P/'before-apply.json').read_text(encoding='utf8'));r=json.loads((P/'result-applied.json').read_text(encoding='utf8'));assert r['applied']
|
canon=lambda v:json.loads(json.dumps(v,ensure_ascii=False,default=str))
|
def fingerprint(v):return hashlib.sha256(json.dumps(v,ensure_ascii=False,sort_keys=True,default=str).encode()).hexdigest()
|
changes={(x['table'],x['id']):x['after'] for x in r['changes']}
|
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,autocommit=False)
|
checked=0
|
with c.cursor() as q:
|
for t,oldrows in b['tables'].items():
|
q.execute('SELECT * FROM `'+t+'` ORDER BY id');newrows=q.fetchall();current={x['id']:x for x in newrows}
|
expected_extra=len(r['new_units']) if t=='mes_md_unit_measure' else 1 if t=='qc_report_template_version' else 0
|
assert len(newrows)==len(oldrows)+expected_extra,t
|
for old in oldrows:
|
expected={**old,**changes.get((t,old['id']),{})};actual=current[old['id']]
|
for k,v in actual.items():
|
if isinstance(v,D):assert v==D(expected[k]),(t,old['id'],k)
|
elif isinstance(v,bytes) and isinstance(expected[k],int):assert v==bytes([expected[k]])
|
else:assert canon(v)==expected[k],(t,old['id'],k)
|
checked+=len(oldrows)
|
for t,h in r['protected_hashes'].items():q.execute('SELECT * FROM `'+t+'` ORDER BY id');assert fingerprint(q.fetchall())==h,t
|
q.execute('SELECT * FROM mes_qc_indicator WHERE deleted=0');inds={x['id']:x for x in q.fetchall()}
|
q.execute('SELECT * FROM mes_md_unit_measure WHERE deleted=0');units={x['id']:x for x in q.fetchall()}
|
for t in ['mes_qc_template_indicator','mes_qc_iqc_line','mes_qc_ipqc_line','mes_qc_oqc_line']:
|
q.execute('SELECT * FROM '+t+' WHERE deleted=0')
|
for row in q.fetchall():
|
assert row['indicator_id'] in inds
|
if row.get('unit_measure_id') is not None:assert row['unit_measure_id'] in units
|
lo=row.get('threshold_min',row.get('min_threshold'));hi=row.get('threshold_max',row.get('max_threshold'))
|
if lo is not None and hi is not None:assert lo<=hi
|
q.execute('SELECT * FROM mes_qc_indicator_result_detail WHERE deleted=0')
|
for row in q.fetchall():
|
ind=inds[row['indicator_id']]
|
if ind['item_type']==2:assert row['value'] is None or not row['value'].strip()
|
elif ind['result_type'] in [1,2] and row['value'] and row['value'].strip():
|
value=D(row['value']);assert value.is_finite()
|
if ind['result_type']==2:assert value==value.to_integral_value()
|
q.execute('SELECT * FROM qc_report_template_version WHERE id=%s',(r['report_template_version']['id'],));version=q.fetchone();assert version and version['version']=='v1.2'
|
q.execute('SELECT current_version FROM qc_report_template WHERE id=14');assert q.fetchone()['current_version']=='v1.2'
|
reports=json.loads((P/'report-output.json').read_text(encoding='utf8'));findings=[]
|
for report in reports:
|
for item in report['snapshot']['inspectionItems']:
|
if item.get('group',{}).get('span',1)>1 and not item.get('group',{}).get('childName'):continue
|
if item['result']!='PASS':findings.append({'report_id':report['id'],'report_no':report['report_no'],'indicator':item['itemName'],'actual_value':item['actualValue'],'unit':item['unit'],'requirement':item['requirement'],'result':item['resultText'],'remark':item['remark']})
|
with (P/'报告异常与待核对.csv').open('w',encoding='utf-8-sig',newline='') as f:
|
writer=csv.DictWriter(f,fieldnames=list(findings[0]));writer.writeheader();writer.writerows(findings)
|
types=dict(collections.Counter(str(x['result_type']) for x in inds.values()))
|
failed_reports=[x['id'] for x in reports if x['snapshot']['report']['result']=='FAIL']
|
result={'verified':True,'rows_checked':checked,'result_types':types,'report_template_version':'v1.2','reports_updated':len(reports),'reports_with_out_of_range_values':failed_reports,'operational_tables_unchanged':len(r['protected_hashes']),'typed_values_valid':True,'references_valid':True,'bounds_valid':True}
|
(P/'verification.json').write_text(json.dumps(result,ensure_ascii=False,indent=2),encoding='utf8')
|
(P/'质检优化核对清单.md').write_text('# 质检检测项、方案与报告优化\n\n已同步远程 `mom-hm`，原始快照、变更SQL及逐项标准核对表在同目录保存。\n\n- 分组：感官与性状、理化与物性、营养与活性成分、微生物、污染物与重金属、农兽药残留；保留甲醇专项分组。\n- 结果类型：205个浮点、6个整数、69个文本、7个无结果值的分组项。\n- 22种检测单位：百分比、质量分数、污染物浓度、微生物计数及物性单位；不改产品包装单位。\n- 标准上下限按现有SOP/国标资料原文提取；标准品/试剂纯度、分级、抽样方案及检出条件不合并成一个通用验收限值。\n- 原来只有“合格、正常”等文字的数值项，保留原文到备注及备份，数值留空待补录或确认，不生成假的实测数字。\n- 12份有效来源报告重新生成；1份来源已删除的报告保留原冻结数据。物料报告模板升级到v1.2，显示完整标准要求并改善列宽，保留原版本。\n- 6份旧PDF附件关联作废，实体文件保留；重新导出PDF即可取得更新报告。\n- 4份报告（ID 5、13、15、18）有现存数值超出对应标准，见“报告异常与待核对.csv”，需人工复核；原质检单人工结论未改动。\n- 原销售、生产、库存、质量单头的数量和状态未变，独立连接核验通过。\n- 后端28项相关测试、实际报告渲染和前端生产构建通过；报告判定和数值输入的代码修复需发布新版前后端后生效。\n',encoding='utf8')
|
print(json.dumps(result,ensure_ascii=False));c.close()
|
