kangshifu-board/scripts/prepare-noodle-innovation.py

"""Prepare source-linked research additions; all writes are local until the intake script applies."""
from pathlib import Path
from datetime import datetime, timezone
import hashlib
import json

ROOT = Path(__file__).resolve().parents[1]
WORK = ROOT / 'work/noodle-innovation-20261108-r1'
DOCS = ROOT / 'docs/noodle-innovation-20261108'
PREFIX = 'NOODLE-INNOVATION-20261108-R1'


def load(p):
    return json.loads(p.read_bytes())


def main():
    DOCS.mkdir(parents=True, exist_ok=True)
    frozen = WORK / 'intake-plan.json'
    if frozen.exists():
        plan = load(frozen)
        print(json.dumps({'alreadyPrepared': True, 'preservedFrozenPlan': str(frozen),
                          'sources': len(plan['groups']),
                          'records': sum(len(g['records']) for g in plan['groups'])}, ensure_ascii=False))
        return
    groups = []
    standards = load(WORK / 'pdf/index.json')
    policy = load(ROOT / 'artifacts/ksf-industry-policy-usage-20261108-r1/package/policy-usage.json')
    metrics = load(WORK / 'scope-audit.json')

    def group(key, title, text, kind='dataset', mode='manual', url=None, payload=None, observed=None):
        source = dict(title=title, rawText=text, sourceKind=kind, dataMode=mode,
                      platform='source-linked-research', collectedAt=observed or datetime.now(timezone.utc).isoformat())
        if url:
            source['sourceUrl'] = url
        if payload:
            source['payload'] = payload
        g = dict(key=key, source=source, contentHash=hashlib.sha256(text.encode()).hexdigest(), records=[])
        groups.append(g)
        return g

    def add(g, cls, key, **fields):
        g['records'].append(dict(className=cls, itemKey=key, externalKey=PREFIX + ':' + key,
                                 data=dict(dataMode=g['source']['dataMode'], reviewState='pending', recordState='active', **fields)))

    # An analyst taxonomy is explicitly marked manual; it never reclassifies the VOC products.
    definition = ('研究范围：预包装速食面及相邻方便米粉/粉丝；包括冲泡、短时煮制、微波等制备方式。'
                  '堂食、现制、生鲜面、原料挂面不进入成品范围。开袋即食只是制备方式之一，不等同全部速食。'
                  '这是工作坊研究分类，不是国家标准分类，也不把标题卖点当生产证据。'
                  '按食用方式、干燥工艺、制备方式三个独立维度建图，不将拌面与非油炸设为同一层排他类别。')
    tax = group('taxonomy', '速食面研究品类图谱与范围说明', definition,
                payload={'claimKind': 'analyst_working_taxonomy', 'libraryGeneration': metrics['generation'],
                         'officialDefinition': 'GB17400-2015 PDF第3页，第1、2章，仅用作边界参考'})
    cats = [
        ('scope', '预包装速食面研究范围', '范围', '不是开袋即食的同义词；只描述本次研究范围'),
        ('eating', '食用形态维度', '维度', '食用形态与干燥工艺分开；具体产品可为组合包装'),
        ('soup', '汤面', '食用形态', '按食用说明确认保留汤汁食用；牛肉口味不自动证明汤面'),
        ('mixed', '拌面', '食用形态', '按沥水拌食或即拌说明确认；不把拌面酱作为成品面'),
        ('fried', '炒面', '食用形态', '方便炒面候选；需核是否速食成品，不含堂食现炒'),
        ('snack', '干吃面', '食用形态', '按开袋干吃食用说明确认，与干燥原料挂面分开'),
        ('noodle-sheets', '方便面皮与凉皮', '相邻产品', '须确认已熟制/速食预包装；生面皮不纳入'),
        ('rice-vermicelli', '方便米粉与粉丝', '相邻产品', '单列基质，不与小麦面混算；GB17400范围参考'),
        ('drying', '干燥工艺维度', '维度', '由生产文件确认，不根据拌面名称或营养宣称推定'),
        ('oil', '油炸干燥', '干燥工艺', 'GB17400-2015第2.1.1条参考；不是口味分类'),
        ('nonoil', '非油炸干燥', '干燥工艺', 'GB17400-2015第2.1.2条参考；非油炸不等于低脂'),
        ('preparation', '食前制备维度', '维度', '以完整食用说明为证据，可与多种食用形态关联'),
        ('rehydrate', '冲泡复水', '食前制备', '保留水温、用水量、时间，当前未知则留空'),
        ('boil', '短时煮制', '食前制备', '速煮与普通原料挂面分开，核食用说明'),
        ('microwave', '微波加热', '食前制备', '需核包装与设备适用条件，不自动等同保鲜湿面'),
        ('ready', '开袋即食', '食前制备', '只用于标签明确无需烹调的产品，不能覆盖所有速食面'),
    ]
    for code, name, axis, note in cats:
        add(tax, 'KfCategory', 'category-' + code, code='NI-' + code.upper(), name=name,
            taxonomyVersion=PREFIX, axis=axis, scopeNote=note + '。' + definition)
    rel = group('taxonomy-relations', '速食面研究分类关系及解释', definition, payload={'claimKind': 'analyst_working_taxonomy'})
    for target in ['eating', 'drying', 'preparation']:
        add(rel, 'KfCategoryRelation', 'relation-scope-' + target, fromCategoryRef={'$category': 'scope'},
            toCategoryRef={'$category': target}, relationType='has_research_axis', taxonomyVersion=PREFIX,
            rationale='研究范围包含独立分类维度，不代表某商品已核实属于该类别', claimKind='research_model')
    for parent, children in [('eating', ['soup', 'mixed', 'fried', 'snack', 'noodle-sheets', 'rice-vermicelli']),
                             ('drying', ['oil', 'nonoil']), ('preparation', ['rehydrate', 'boil', 'microwave', 'ready'])]:
        for child in children:
            add(rel, 'KfCategoryRelation', 'relation-' + parent + '-' + child,
                fromCategoryRef={'$category': parent}, toCategoryRef={'$category': child},
                relationType='research_axis_option', taxonomyVersion=PREFIX,
                rationale='该维度的可选研究项；商品归属需查对应原始证据', claimKind='research_model')

    for doc in standards:
        code = doc['documentId']
        text = (WORK / 'pdf' / (code + '.txt')).read_text(encoding='utf-8')
        g = group(code, code + ' ' + doc['title'] + '：原件提取与核验边界', text, 'standard', 'live', doc['sourceUrl'],
                  payload={'originalSha256': doc['sha256'], 'originalBytesArchived': doc['file'],
                           'extractedTextSha256': doc['extractedTextSha256'], 'pdfPages': doc['pdfPages'],
                           'textExtractionMayDistortTables': True, 'fullTextAllClausesVerified': False,
                           'currentLegalStatusReverified': False, 'statusAsOfPreviouslyDeclared': doc['declaredStatusAsOf'],
                           'originalPackage': 'ksf-industry-policy-usage-20261108-r1',
                           'alreadyReviewedSubstantivePages': doc['substantivePagesReviewedThisRun']})
        verified = [c for c in policy['clauses'] if c['documentId'] == code and c.get('verification') != 'PENDING_ORIGINAL']
        add(g, 'KfStandard', 'standard-' + code, standardCode=code.split('-')[0], edition=code.split('-')[1],
            title=doc['title'], status='future_by_cover_not_currently_enforced' if doc['notYetEffectiveOnReviewDate'] else 'archived_original_current_status_not_reverified',
            effectiveDate=doc['effectiveDateOnCover'] + 'T00:00:00+08:00',
            scope='原件封面、目录与已核页供研究检索；当前适用版本、产品类别及完整条款仍需核验。' +
                  ('2027-03-16实施，不能视作2026年已强制执行。' if doc['notYetEffectiveOnReviewDate'] else ''),
            standardType='national_food_safety_standard', issuer='国家卫生健康主管部门及相关发布部门（以原件封面为准）',
            clauses=[{'clauseId': c['clauseId'], 'verification': c['verification'], 'review': c} for c in verified])
        if code == 'GB17400-2015':
            for key, name in [('wheat', '小麦粉'), ('grain', '其他谷物粉'), ('starch', '食用淀粉')]:
                add(g, 'KfMaterial', 'material-' + key, name=name, materialType='food_ingredient_candidate',
                    function='方便面面饼主要原料候选；不是特定竞品的实际配方',
                    constraints={'sourceLocator': 'GB17400-2015 PDF第3页，第2.1、2.4条',
                                 'foodGradeSpecificationPending': True, 'formulaRatioUnknown': True,
                                 'supplierAndQuoteUnknown': True, 'allergenStatus': '需按实际原料核验，小麦原料涉及麸质'},
                    specification='待补食品级规格、粒度、蛋白/淀粉特性、批次COA及供应商信息')
            for key, name, quote in [('oil', '油炸方便面干燥路线（定义级）', '采用油炸工艺干燥的方便面'),
                                     ('nonoil', '非油炸方便面干燥路线（定义级）', '采用除油炸以外的其他工艺（如微波、真空和热风等）干燥的方便面')]:
                add(g, 'KfProcess', 'process-' + key, name=name, processType='drying_route', processKind='standard_definition',
                    description=quote + '。仅记录分类定义，不是康师傅或竞品工厂的真实产线文件。',
                    steps=[{'step': '面饼干燥', 'basis': quote, 'sourceLocator': 'PDF第3页，第2.1.' + ('1' if key == 'oil' else '2')}],
                    parameters=[], factoryContext={'factoryVerified': False, 'haccpAndCcpPending': True,
                    'missing': ['和面/熟化/压延/切丝/蒸煮等实际工序', '温度', '时间', '水分/水活度', '产能', '损耗', '设备', '验证与检验记录']})
        if code == 'GB2760-2024':
            for name in ['碳酸钠', '碳酸钾', '黄原胶', '瓜尔胶', '三聚磷酸钠', '谷氨酸钠']:
                pages = doc['termPageHits'][name]
                add(g, 'KfMaterial', 'additive-' + name, name=name, materialType='food_additive_candidate',
                    function='添加剂检索候选；功能和使用范围必须回到本标准该物质完整表项核验',
                    constraints={'standardCode': code, 'nameFoundOnPdfPages': pages, 'permissionForInstantNoodlesVerified': False,
                                 'dosageLimitVerified': False, 'maximumUseAmount': None,
                                 'requiredChecks': ['面饼与调料包分别确定食品类别', '使用范围与最大用量/残留量', '表A.2及注释', '带入原则', '物质质量规格']})
        if code == 'GB14880-2012':
            for name, alias in [('维生素B1', '硫胺素'), ('维生素B2', '核黄素')]:
                add(g, 'KfMaterial', 'fortifier-' + name, name=name, aliases=[alias], materialType='nutrient_fortifier_candidate',
                    function='营养强化剂候选；不是已批准的方便面配方，也不作为保健补充剂上市资料',
                    constraints={'standardCode': code, 'compoundSourcePdfPages': doc['termPageHits'][alias],
                                 'permissionForChosenCategoryVerified': False, 'dosageLimitVerified': False,
                                 'requiredChecks': ['附录A食品类别及允许强化量', '附录B化合物来源', '营养声称', '工艺稳定性及检测']})

    for key in ['angel-ye', 'angel-ye-processing', 'angel-ye-salt']:
        meta = load(WORK / 'web' / (key + '.meta.json'))
        text = (WORK / 'web' / (key + '.txt')).read_text(encoding='utf-8')
        g = group(key, '安琪公开技术资料：' + meta['title'][0], text, 'web', 'live', meta['url'],
                  payload={'originalHtmlSha256': meta['sha256'], 'extractedTextSha256': meta['textSha256'],
                           'evidenceKind': 'manufacturer_self_description_and_advertising', 'independentVerification': False,
                           'priceOrSupplierApprovalVerified': False}, observed=meta['collectedAt'])
        if key == 'angel-ye':
            add(g, 'KfMaterial', 'material-ye', name='酵母抽提物（YE）', materialType='food_ingredient_manufacturer_candidate',
                function='生产商公开描述：增鲜增味、平衡异味，列有方便面应用；效果须通过目标配方试验',
                constraints={'sourceClaimNotIndependentProof': True, 'regulatoryClassificationRequiresActualSpec': True,
                             'saltReductionNotEstablishedByTaste': True, 'formulaRatioUnknown': True})
            add(g, 'KfSupplier', 'supplier-angel', name='安琪酵母（公开资料供应线索）', supplierType='manufacturer_lead',
                region='湖北宜昌（网页公司地址）', supplyScope='酵母抽提物及食品调味配料；具体供应实体/型号/产地待核',
                approvalState='not_assessed', qualification={'evidence': '官网产品自述', 'businessLicenseVerified': False,
                'foodProductionLicenseVerified': False, 'coAReceived': False, 'quotationReceived': False})
        if key == 'angel-ye-processing':
            for name in ['FA92', 'FA93']:
                add(g, 'KfMaterial', 'material-ye-' + name.lower(), name='安琪抗吸潮酵母抽提物 ' + name,
                    materialType='manufacturer_product_candidate', specification=name + '；网页列有方便面调料包应用',
                    function='生产商声称抗吸潮；FA92增鲜、FA93增厚。尚未验证目标料包效果',
                    storageConditions='供应商网页建议阴凉通风干燥，启封未用完密封；实际采购按当批规格书',
                    constraints={'claimKind': 'manufacturer_claim', 'coARequired': True, 'foodSafetyLimitsNotAdoptedFromAdvertisement': True,
                                 'supplierQuotePending': True}, sensoryProfile={'supplierClaimOnly': True, 'independentSensoryTest': False})
            add(g, 'KfProcess', 'process-ye', name='抗吸潮YE制备路线（供应商公开概述）', processType='ingredient_processing',
                processKind='manufacturer_public_description', description='供应商网页描述高蛋白酵母原料经定向酶解、后处理、喷雾干燥制成抗吸潮抽提物。',
                steps=[{'sequence': i + 1, 'step': x} for i, x in enumerate(['高蛋白酵母原料', '定向酶解', '后处理', '喷雾干燥'])],
                parameters=[], factoryContext={'sourceClaimOnly': True, 'factoryParametersVerified': False,
                    'missing': ['酶种/活力', 'pH', '时间', '温度', '灭酶条件', '喷雾参数', '含水量', '成品检验']})
        if key == 'angel-ye-salt':
            # Retain surrounding use cases. Do not transfer soy sauce dose ranges to noodles.
            g['source']['payload']['criticalBoundary'] = '页面方案对象主要是酱油、食醋和调味酱；推荐比例不是方便面合法用量，也不证明钠含量下降。'

    locator = load(WORK / 'product-locator-index.json')
    raw = json.dumps({'coverage': metrics, 'locators': locator,
                      'reviewContextExamples': load(WORK / 'review-context-candidates.json')}, ensure_ascii=False, separators=(',', ':'))
    group('library-locators', '康师傅与竞品速食面：2607件库内商品定位及覆盖审计', raw, 'dataset', 'live',
          'https://kangshifu.fmode.cn/library?type=products&quality=audit',
          payload={'libraryGeneration': metrics['generation'], 'snapshotExportedAt': metrics['snapshotExportedAt'],
                   'newUpstreamCollection': False, 'scope': 'existing_records_only', 'marketShareOrNegativeRate': False,
                   'purchaseCostNotRetailPrice': True})
    plan = {'schema': 'ksf.noodle-innovation-intake-plan.v1', 'createdAt': datetime.now(timezone.utc).isoformat(),
            'libraryGeneration': metrics['generation'], 'schemaChange': False, 'updateExistingRecords': False,
            'reviewState': 'pending', 'groups': groups}
    (WORK / 'intake-plan.json').write_text(json.dumps(plan, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
    print(json.dumps({'sources': len(groups), 'records': sum(len(g['records']) for g in groups),
                      'classes': {c: sum(r['className'] == c for g in groups for r in g['records']) for c in
                      ['KfCategory', 'KfCategoryRelation', 'KfStandard', 'KfMaterial', 'KfProcess', 'KfSupplier']},
                      'bytes': (WORK / 'intake-plan.json').stat().st_size}, ensure_ascii=False))


if __name__ == '__main__':
    main()