kangshifu-board/scripts/build-industry-policy-usage.py

"""Build a source-bound policy review from local originals; never writes a database."""
import argparse
from collections import Counter
from datetime import datetime, timezone
import hashlib
from html.parser import HTMLParser
import json
from pathlib import Path
import re
import shutil
import sqlite3

ROOT = Path(__file__).resolve().parents[1]
GENERATION = '82b104fc024b49a75d2f5376bf9778ced6f63a9433b5fc5a6daa3c458d05d5ea'


def sha(raw):
    return hashlib.sha256(raw).hexdigest()


def norm(value):
    return re.sub(r'\s+', '', value)


class PageText(HTMLParser):
    def __init__(self):
        super().__init__(convert_charrefs=True)
        self.parts, self.hidden = [], []

    def handle_starttag(self, tag, attrs):
        if tag in ('script', 'style'):
            self.hidden.append(tag)

    def handle_endtag(self, tag):
        if self.hidden and self.hidden[-1] == tag:
            self.hidden.pop()

    def handle_data(self, text):
        if not self.hidden:
            self.parts.append(text)


def text_of(raw):
    p = PageText()
    p.feed(raw.decode('utf-8-sig'))
    return re.sub(r'\s+', ' ', ' '.join(p.parts)).strip()


def save(path, data):
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(data, ensure_ascii=False, indent=2)+'\n', encoding='utf-8')


# Page citations below were visually read from the exact, hash-matched PDFs.
# Table values remain structured cells rather than invented contiguous quotations.
NOTES = {
    1: dict(topic='工艺分类', applies='符合GB17400范围的方便面；实际干燥工艺需生产资料确认',
            inputs=['干燥工艺记录', '产品配方与食用方法'],
            steps=['先核范围及产品定义，再据生产记录区分油炸和非油炸。', '分别保存面饼与调料包的对象；非油炸标签不能替代营养含量证据。'],
            limits='工艺定义不证明低脂或低能量。'),
    2: dict(topic='面饼理化检验', applies='油炸/非油炸面饼；酸价和过氧化值为油炸面饼、以脂肪计',
            inputs=['已核干燥工艺', '面饼样品与批次', '检测报告及方法版本'],
            steps=['按工艺核水分上限：油炸10.0、非油炸14.0 g/100g。', '油炸面饼核酸价≤1.8 mg KOH/g、过氧化值≤0.25 g/100g，保留以脂肪计。', '分别回查表中GB5009.3、GB5009.229、GB5009.227及适用版本。'],
            limits='本条只核表2，不代替整份产品标准或商品检测。'),
    3: dict(topic='污染物和致病菌的交叉引用', applies='GB17400产品；须继续核目标标准中的食品类别、对象及采样条件',
            inputs=['配方及食用方式', '检验对象和采样方案', '引用标准原文及适用版本'],
            steps=['回查GB2762中带馅（料）面米制品的对应项目。', '回查GB29921中方便面米制品的项目及适用范围。', '保存目标标准页码、限量、单位、采样n/c/m/M；未补齐时不出合格结论。'],
            limits='引用标准号不是完成了目标标准逐项检验。'),
    4: dict(topic='添加剂与营养强化剂', applies='具体配方、食品类别和物质；面饼与料包的对象须明确',
            inputs=['完整配方与各包组成', '添加剂通用名及功能', '用量/残留量与质量规格'],
            steps=['添加剂查GB2760：先定本标准食品类别，再查附录A对应物质、用量、限制及带入条件。', '强化剂查GB14880：同时核附录A使用范围和量、附录B化合物来源；特殊膳食另核适用规定。', '将物质、食品类别、工艺、用量、目标表页码逐项留档。'],
            limits='不把新品种公告标题或其他食品的许可当方便面配方许可。'),
    5: dict(topic='钠含量声称', applies='采用GB28050-2011核查的预包装食品；先核生产日期及适用版本',
            inputs=['营养表/检测证据', '每份质量或体积', '可食部及干品/冲调后、料包范围'],
            steps=['记录原标示及单位，统一每100g或每100mL基准。', '分别核无钠≤5、极低钠≤40、低钠≤120 mg/100g或100mL。', '保留同页表注：以份计量也须符合每100g（mL）的要求；符合钠声称时可按表述用盐字替代钠。'],
            limits='减小每份量不会自动达到含量声称要求；偏咸评价不是钠检测证据。'),
    6: dict(topic='膳食纤维声称', applies='采用GB28050-2011核查的预包装食品；固体/液体/能量基准分别记录',
            inputs=['纤维检测或营养表', '固体/液体状态', '每100g、每100mL或每420kJ基准'],
            steps=['来源/含有阈值为3g/100g（固体）、1.5g/100mL（液体）或1.5g/420kJ。', '高/富含/良好来源阈值为6g/100g（固体）、3g/100mL（液体）或3g/420kJ。', '核限制性条件：总膳食纤维，或可溶性、不溶性、单体成分任一项符合；保留检测对象与选用基准。', '使用份标示时仍保留表下注释，不靠缩小份量获得声称。'],
            limits='全谷物、粗粮或杂粮标题不能证明纤维含量。'),
    7: dict(topic='减少钠的比较声称', applies='同类或同一属类、消费者熟知且容易理解的参考食品比较',
            inputs=['参考食品身份及同类依据', '双方同基准含量证据', '比较日期及食品状态'],
            steps=['保留参考食品和候选食品的原件与计量基准。', '按（参考值−候选值）÷参考值×100%记录差异，核表C.3的减少25%以上要求。', '含量声称与比较声称分别留档，另核拟使用版本和措辞。'],
            limits='减少钠不等于低钠；半包粉包不等于全产品钠减半。'),
    8: dict(topic='标签换版：无/不含等措辞', applies='GB7718-2025的解释资料；原标准封面实施日2027-03-16',
            inputs=['目标生产日期', '拟用标签及词语', '实施与过渡安排'],
            steps=['定位问答33列明的三类对象，并回读GB7718-2025第4.4.2.1条。', '分别保存食品添加剂、污染物、禁止添加/不应存在物质的词语核查。', '按生产批次和实施安排确定标签方案。'],
            limits='问答是解释资料；不把未来版当作2026年所有商品已强制实施。'),
    9: dict(topic='标签换版：配料、图片和致敏物', applies='GB7718-2025问答31、35、38对应情形；先核版本和例外',
            inputs=['配料比例及来源', '标签图片与真实配料关系', '八类致敏物质清单'],
            steps=['问答31：核特别强调配料/成分时的定量标示与例外。', '问答35：区分真实配料与仅香精香料调味，核照片/图案及相邻提示。', '问答38：核直接作为配料使用的八类致敏物及标示形式；与可能交叉带入的预防性提示分开。'],
            limits='归档摘要只作索引；实际标签逐项回读正式标准和解释条件。'),
    10: dict(topic='全谷物创新背景', applies='面向成人膳食结构的指导信息，不是单份方便面的配方比例',
             inputs=['原料完整籽粒结构证据', '配方比例', '口感与保存验证'],
             steps=['先核全谷物定义，区分完整结构与仅粗粮/杂粮名称。', '50—100g指成人每日全谷物摄入建议，不能直接写成单份产品目标。', '把日常搭配、口感、保存条件列为研发验证项。'],
             limits='全谷物指导不能替代高纤维等营养声称标准。'),
    11: dict(topic='营养标签选择背景', applies='蔬菜水果章节中的加工制品选择信息；推广到方便面属于研究解释',
             inputs=['章节语境', '营养表原件', '同基准糖/钠/脂肪含量'],
             steps=['保留第8条前句“鼓励多食新鲜的蔬菜、水果”及后句加工制品语境。', '将营养成分作为可研究比较维度，具体声称回到适用标准。'],
             limits='原摘要省略蔬菜水果语境；本次补充适用说明，不改线上原条款。'),
    12: dict(topic='三减研发和标签识读背景', applies='营养指导员在餐饮与食品健康服务中的工作指导',
             inputs=['服务对象及场景', '食品营养信息', '研发与消费者沟通验证计划'],
             steps=['区分餐饮油盐糖用量管理、协助企业三减创新和标签识读三类工作。', '将其用于研发议题与教育设计；单品数值、声称与安全判断另查产品标准及证据。'],
             limits='指导员服务规范不提供方便面减盐比例或单品营养功效许可。'),
}

HTML_SECTIONS = {
    8: [('三十三、', '三十四、')],
    9: [('三十一、', '三十二、'), ('三十五、', '三十六、'), ('三十八、', '三十九、')],
    10: [('12.', '13.'), ('14.', '15.'), ('16.', '17.'), ('20.', '21.')],
    11: [('8.', '9.')],
    12: [('3. 指导餐厅', '4. 协助食品'), ('4. 协助食品', '5. 在营养'), ('6. 提供食物选购指导', '（五）')],
}

MANUAL_QUOTES = {
    1: ['采用油炸工艺干燥的方便面，包括泡面、干吃面和煮面。',
        '采用除油炸以外的其他工艺（如微波、真空和热风等）干燥的方便面，包括泡面、干吃面和煮面。'],
    3: ['污染物限量应符合GB 2762中带馅（料）面米制品的规定。',
        '致病菌限量应符合GB 29921中方便面米制品的规定。'],
    4: ['食品添加剂的使用应符合GB 2760的规定。', '食品营养强化剂的使用应符合GB 14880的规定。'],
}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--work', type=Path, default=ROOT/'work/industry-policy-usage-20261108-r1')
    parser.add_argument('--out', type=Path, required=True)
    args = parser.parse_args()
    assert not args.out.exists(), 'OUTPUT_ALREADY_EXISTS'
    snapshot_path = ROOT/'work/industry-monitoring-20261107-r1/snapshot.json'
    snapshot_sha = sha(snapshot_path.read_bytes())
    assert snapshot_sha == '45b3d7f4cda3d70b4e3398da10fc8c21055d80a8c3ff2e33997ba79f06719408'
    snapshot = json.loads(snapshot_path.read_bytes())
    assert snapshot['generation'] == GENERATION
    docs = {d['id']: d for d in snapshot['entities']['documents']}
    inventory = json.loads((args.work/'pdf-inventory.json').read_bytes())
    registry = ROOT/'artifacts/ksf-industry-monitor-20261107-r11/package/data/monitoring-registry'
    db = registry/'monitoring.sqlite'
    db_sha = sha(db.read_bytes())
    args.out.mkdir(parents=True)
    sources, pages, html_texts = {}, {}, {}
    document_rows = []
    for row in inventory['rows']:
        ident = row['documentId']
        pdf = Path(row['pdfPath'])
        assert sha(pdf.read_bytes()) == row['sha256'] == docs[ident]['expected_sha256']
        rel = 'sources/'+ident+'.pdf'
        (args.out/'sources').mkdir(exist_ok=True)
        shutil.copyfile(pdf, args.out/rel)
        pages[ident] = json.loads(Path(row['textFile']).read_bytes())
        assert len(pages[ident]) == row['pages']
        for p in pages[ident]:
            assert sha(p['text'].encode()) == p['textSha256']
        sources[ident] = {'file': rel, 'sha256': row['sha256'], 'sourceUrl': docs[ident]['source_url']}
        render = args.work/'renders'/(ident+'-p1.png')
        dest = args.out/'previews'/render.name
        dest.parent.mkdir(exist_ok=True)
        shutil.copyfile(render, dest)
        document_rows.append({'documentId': ident, 'title': docs[ident]['title'], **sources[ident],
            'pdfPages': row['pages'], 'coverVisualVerified': True,
            'coverRender': 'previews/'+render.name, 'coverRenderSha256': sha(render.read_bytes()),
            'issuedDateOnCover': docs[ident]['issued_date'], 'effectiveDateOnCover': docs[ident]['effective_date'],
            'notYetEffectiveOnReviewDate': docs[ident]['effective_date'] > '2026-11-08',
            'declaredStatusInOldSnapshot': docs[ident]['declared_status'],
            'declaredStatusAsOf': docs[ident]['status_as_of'], 'currentLegalStatusRecheckedThisRun': False,
            'textLayerPresent': any(p['text'].strip() for p in pages[ident]),
            'substantivePagesReviewedThisRun': [3, 4, 5] if ident == 'GB17400-2015' else [3, 4, 10] if ident == 'GB28050-2011' else [3,4] if ident=='GB14880-2012' else [4,5,249] if ident=='GB2760-2024' else [5] if ident=='GB7718-2025' else []})
    with sqlite3.connect(db.resolve().as_uri()+'?mode=ro', uri=True) as con:
        pubs = [json.loads(r[0]) for r in con.execute('SELECT payload FROM policy_publications')]
    for pub in pubs:
        ident = pub['documentId']
        raw = (registry/'assets'/(pub['archiveWrapperSha256']+'.json')).read_bytes()
        assert sha(raw) == docs[ident]['expected_sha256'] == pub['archiveWrapperSha256']
        wrapper = json.loads(raw)
        assert wrapper['url'] == docs[ident]['source_url']
        body = wrapper['html'].encode()
        assert sha(body) == pub['pageSha256']
        text = text_of(body)
        assert norm(docs[ident]['title']) in norm(text)
        html_texts[ident] = text
        rel = 'sources/'+ident+'.html'
        (args.out/rel).write_bytes(body)
        (args.out/'sources'/(ident+'.wrapper.json')).write_bytes(raw)
        sources[ident] = {'file': rel, 'sha256': sha(body), 'sourceUrl': wrapper['url'],
                          'wrapperSha256': sha(raw), 'observedAtOriginal': wrapper['collectedAt']}
        visible = re.search(r'发布时间[:：]\s*(\d{4}-\d{2}-\d{2})', text)
        visible_day = visible[1] if visible else None
        document_rows.append({'documentId': ident, 'title': docs[ident]['title'], **sources[ident],
            'publicationIdentityMatched': True, 'metadataPublishedAtRaw': pub['publishedAtRaw'],
            'visiblePublishedDate': visible_day,
            'publicationDateConflict': visible_day is not None and visible_day != pub['publishedAtRaw'][:10],
            'currentLegalStatusRecheckedThisRun': False, 'declaredStatusInOldSnapshot': docs[ident]['declared_status']})
    for ident, doc in docs.items():
        if ident not in sources:
            document_rows.append({'documentId': ident, 'title': doc['title'], 'sourceUrl': doc['source_url'],
                'expectedSha256': doc['expected_sha256'], 'originalAvailableInThisReview': False,
                'archiveTextAvailable': bool(doc.get('full_text')), 'declaredStatusInOldSnapshot': doc['declared_status'],
                'currentLegalStatusRecheckedThisRun': False})
    clauses = []
    for old in sorted(snapshot['entities']['clauses'], key=lambda c: c['source']['line_number']):
        n, ident = old['source']['line_number'], old['document_id']
        r = {'clauseId': old['id'], 'sourceLine': n, 'documentId': ident, 'clause': old['clause'],
             'archivedExcerpt': old['excerpt'], 'archivedExcerptType': old['excerpt_type'],
             'archivedApplicationNote': old['application_note'], 'sourceRecordId': old['source_record_id'],
             'sourceUrl': docs[ident]['source_url'], 'declaredDocumentStatus': docs[ident]['declared_status'],
             'currentLegalStatusRecheckedThisRun': False, 'productComplianceDetermined': False,
             'interpretationIsRegulation': False, 'sourceReferences': []}
        if n <= 7:
            page = int(old['pdf_page'])
            render = args.work/'renders'/(ident+'-p'+str(page)+'.png')
            assert render.is_file()
            dest = args.out/'previews'/render.name
            shutil.copyfile(render, dest)
            ref = {**sources[ident], 'pdfPage': page,
                   'printedPage': str(page-2 if ident == 'GB17400-2015' else page-1),
                   'renderFile': 'previews/'+render.name, 'renderSha256': sha(render.read_bytes()),
                   'visuallyVerified': True}
            if n in MANUAL_QUOTES:
                ref.update(quoteMethod='manual_transcription_from_visual_page_layout_normalized',
                           quotes=MANUAL_QUOTES[n])
            elif n == 2:
                ref.update(quoteMethod='manual_table_cell_transcription', cells=[
                    {'object': '油炸面饼', 'metric': '水分', 'max': '10.0', 'unit': 'g/100g', 'testMethod': 'GB5009.3'},
                    {'object': '非油炸面饼', 'metric': '水分', 'max': '14.0', 'unit': 'g/100g', 'testMethod': 'GB5009.3'},
                    {'object': '油炸面饼', 'metric': '酸价（以脂肪计）', 'max': '1.8', 'unit': 'mg KOH/g', 'testMethod': 'GB5009.229'},
                    {'object': '油炸面饼', 'metric': '过氧化值（以脂肪计）', 'max': '0.25', 'unit': 'g/100g', 'testMethod': 'GB5009.227'}])
            else:
                ref.update(quoteMethod='pdf_text_layer_with_visual_table_check',
                           pageText=pages[ident][page-1]['text'], textSha256=pages[ident][page-1]['textSha256'])
            r['sourceReferences'].append(ref)
            r['verification'] = 'PDF_PAGE_VERIFIED'
        elif n <= 12:
            text = html_texts[ident]
            for begin, end in HTML_SECTIONS[n]:
                start = text.index(begin)
                stop = text.index(end, start+len(begin))
                r['sourceReferences'].append({**sources[ident], 'headingStart': begin, 'headingEndExclusive': end,
                    'textStart': start, 'textEndExclusive': stop, 'quote': text[start:stop],
                    'extraction': 'HTMLParser visible text, whitespace collapsed; character offsets in extracted text'})
            r['verification'] = 'ARCHIVED_OFFICIAL_HTML_VERIFIED'
        else:
            r.update(verification='PENDING_ORIGINAL', topic='资料待核',
                applies='原件及条款适用条件未复核', inputs=['与库内期望SHA一致的原件', '附件及其实施日期、适用范围'],
                steps=['先找回原件并核SHA、标题和定位。', '公告目录回查对应标准正文；三新食品回查批准附件、食品类别、限量、适用人群和产品类型。', '取得原件后逐条补引用；现有摘要保留为查找线索。'],
                limits='该条不能直接给出配方、用量、标签或合规结论。')
            if 13 <= n <= 17:
                r['topic'] = '数字标签资料待核'
                r['steps'] = ['找回数字标签公告原件并核发布信息和适用条件。', '分别核营销信息边界、可读性、变更追溯和实体标签保留义务。']
            elif 19 <= n <= 23:
                r['topic'] = '新标准目录线索'
            elif n in (25,29):
                r['topic'] = '历史征集或征求意见线索'
        if n in NOTES:
            r.update(NOTES[n])
        if n in (4,8,9):
            support = ([('GB2760-2024',5),('GB2760-2024',249),('GB14880-2012',4)] if n==4 else [('GB7718-2025',5)])
            r['supportingSourceReferences']=[]
            for docid,page in support:
                text=pages[docid][page-1]['text']
                render=args.work/'renders'/(docid+'-p'+str(page)+'.png')
                assert render.is_file()
                shutil.copyfile(render,args.out/'previews'/render.name)
                r['supportingSourceReferences'].append({**sources[docid],'pdfPage':page,
                    'printedPage':str(page-3 if docid=='GB2760-2024' else page-2),
                    'pageText':text,'textSha256':sha(text.encode()),'visuallyVerified':True,
                    'renderFile':'previews/'+render.name,'renderSha256':sha(render.read_bytes())})
        r['reviewNote'] = ('补回蔬菜水果章节语境，原摘要不改写为逐字引文。' if n==11 else
                           '保留表注与限制性条件，PDF页码与印刷页码分别记录。' if n in (5,6,7) else
                           '正文核验不等于已核今天法律状态或商品符合性。')
        clauses.append(r)
    assert len(clauses) == 31 and len({r['clauseId'] for r in clauses}) == 31
    counts = dict(Counter(r['verification'] for r in clauses))
    assert counts == {'PDF_PAGE_VERIFIED': 7, 'ARCHIVED_OFFICIAL_HTML_VERIFIED': 5, 'PENDING_ORIGINAL': 19}
    workflows = []
    for name, nums in [('process-quality',[1,2,3]),('additives-fortification',[4]),('sodium-claims',[5,7]),
                       ('fiber',[6]),('whole-grain',[10]),('nutrition-context',[11,12]),
                       ('label-transition',[8,9]),('digital-label-and-new-ingredients',list(range(13,32)))]:
        selected = [r for r in clauses if r['sourceLine'] in nums]
        workflows.append({'id':name, 'clauseIds':[r['clauseId'] for r in selected],
            'sourceVerificationComplete':all(r['verification']!='PENDING_ORIGINAL' for r in selected),
            'methodIsRegulation':False, 'requiredDecision':'先核输入、适用日期及原文条件；资料不足保持待核。'})
    result = {'schema':'ksf.policy-usage-review.v1', 'generation':GENERATION, 'reviewDate':'2026-11-08',
        'createdAt':datetime.now(timezone.utc).isoformat(), 'snapshotSha256':snapshot_sha,
        'sourceScope':'local frozen originals; no new official or upstream fetch',
        'summary':{'documents':25, 'pdfOriginalsHashMatched':9, 'pdfCoversVisuallyChecked':9,
                   'officialArchivedHtmlOriginalsMatched':4, 'documentsMissingOriginal':12,
                   'clauses':31, 'clauseVerification':counts,
                   'htmlPublicationDateConflicts':sum(r.get('publicationDateConflict',False) for r in document_rows)},
        'documents':document_rows, 'clauses':clauses, 'workflows':workflows,
        'upstreamCalls':0, 'paidCalls':0, 'researchDatabaseWrites':0, 'productionChanges':0,
        'currentLegalStatusesVerifiedThisRun':0, 'all31ClausesVerified':False}
    save(args.out/'policy-usage.json',result)
    assert sha(db.read_bytes()) == db_sha and sha(snapshot_path.read_bytes()) == snapshot_sha
    save(args.out/'policy-review-verification.json',{'ok':True, 'generation':GENERATION,
        'frozenLedgerSha256':db_sha, 'frozenLedgerUnchanged':True,'snapshotUnchanged':True,
        'sourceDocumentsVerified':13, 'clausesVerified':12, 'clausesPending':19,
        'scannedPdfTextIsNotInvented':True, 'paidCalls':0, 'researchDatabaseWrites':0})
    print(json.dumps(result['summary'],ensure_ascii=False))


if __name__ == '__main__':
    main()