kangshifu-board/scripts/finalize-industry-policy-confirmation.py

"""Produce a local confirmation letter and empty return form without sending it."""
from datetime import datetime, timezone
import hashlib
import json
from pathlib import Path
import re
import zipfile

from docx import Document
from docx.enum.table import WD_CELL_VERTICAL_ALIGNMENT
from docx.enum.text import WD_ALIGN_PARAGRAPH
from docx.oxml import OxmlElement
from docx.oxml.ns import qn
from docx.shared import Cm, Pt, RGBColor

ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT/'artifacts/ksf-industry-policy-confirmation-letter-20261108-r2/package'
WORK = ROOT/'work/industry-policy-confirmation-letter-20261108-r2'
SOURCE = Path('D:/下载/WXWork/1688857450794210/Cache/File/2026-11/ksf-industry-policy-confirmation-letter-20261108-r1.zip')
PRIOR = ROOT/'artifacts/ksf-industry-policy-ranking-review-20261108-r1/package'


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


def font(run, size=10.5, bold=False):
    run.font.name = 'Microsoft YaHei'
    run.font.size = Pt(size)
    run.font.bold = bold
    run.font.color.rgb = RGBColor(0, 0, 0)
    run._element.get_or_add_rPr().get_or_add_rFonts().set(qn('w:eastAsia'), 'Microsoft YaHei')
    return run


def para(doc, text, size=10.5, bold=False, keep=False, style=None):
    p = doc.add_paragraph(style=style)
    font(p.add_run(text), size, bold)
    p.paragraph_format.space_after = Pt(5)
    p.paragraph_format.line_spacing = 1.18
    p.paragraph_format.keep_with_next = keep
    return p


def heading(doc, text):
    p = para(doc, text, 13, True, True, 'Heading 1')
    p.paragraph_format.space_before = Pt(12)
    p.paragraph_format.space_after = Pt(7)
    return p


def save_json(path, value):
    path.write_text(json.dumps(value, ensure_ascii=False, indent=2)+'\n', encoding='utf-8')


def main():
    assert not OUT.exists(), 'OUTPUT_ALREADY_EXISTS'
    raw = SOURCE.read_bytes()
    assert len(raw) == 43330 and sha(raw) == 'a0ffb740b17e4bd85baf3db82f1d726f8c5a054a52c5738eaee4bb7ce675fc89'
    with zipfile.ZipFile(SOURCE) as z:
        entries = {name: digest for digest, name in
                   (line.split('  ', 1) for line in z.read('FILES.sha256').decode().splitlines() if line)}
        assert len(entries) == 3 and set(z.namelist()) == set(entries) | {'FILES.sha256'}
        assert len(z.namelist()) == len(set(z.namelist()))
        for name, digest in entries.items():
            assert '/' not in name and '\\' not in name and sha(z.read(name)) == digest
        incoming_md = z.read('原件补充与榜单来源确认单.md')
        source_docx = z.read('原件补充与榜单来源确认单.docx')
    original = (PRIOR/'原件补充与榜单来源确认单.md').read_bytes()
    norm = lambda b: b.decode('utf-8-sig').replace('\r\n', '\n')
    assert norm(original) == norm(incoming_md)
    with zipfile.ZipFile(WORK/'received/原件补充与榜单来源确认单.docx') as z:
        docx_text = z.read('word/document.xml').decode('utf-8')
    assert 'nhc-' not in docx_text
    review = json.loads((PRIOR/'policy-gap-review.json').read_bytes())
    policies = sorted(review['missing'], key=lambda r: (-r['blockedClauses'], r['documentId']))
    assert len(policies) == 12 and sum(r['blockedClauses'] for r in policies) == 19
    questions = re.findall(r'^([1-5])\. (.+)$', norm(original), re.M)
    assert [n for n, _ in questions] == list('12345')
    intro = '请政策资料方补充下列12份政策正文及相关附件，请VOC账号方或其他合法数据来源方回复商品榜来源5个问题。本单用于确认原件、授权范围和后续验证条件；先提供已有资料及书面说明，收费采集继续停止。'
    policy_intro = '已有9份PDF及4份官方HTML共13份已核原件无需重复提供。请补齐以下文件，确认实际所附的批准清单、适用范围、用量限制、标准文本或征求意见稿；确无附件时请保留原页面链接结构，不按标题推定附件内容。'
    priority = '数字标签公告与2026年第6号标准公告各对应5条待核，可优先补齐；这是原件核查顺序，不是机会排序。GB2762-2025标准PDF已持有，其发布公告与相关缺失附件仍需补充。'
    evidence = '每份请附来源URL、取得时间（含时区）、标题和文号、正文及附件原始文件、文件字节数和SHA-256。普通浏览器可正常访问时可保存网页；遇到挑战保留失败记录。'
    screenshots = '仅能提供截图时，请同时附原始URL和取得时间（含时区）；截图按辅助证据单列，不计入完整正文原件数。浏览器另存页按正文完整性、来源身份及附件保留情况核验，可列为网页存档候选，不因保存方式直接降级或自动核验通过。所有新件先标待核，不自动计入13份已核原件。'
    versions = '新文件与库内expected SHA不同时，先区分HTML正文、归档包装、PDF和版本，再核发布身份、正文与附件；不覆盖旧常量，不仅凭同标题认定同文件。补件到位后逐条回读19条待核记录，核实适用条件、具体用法和边界。'
    ranking_intro = '需要方便面或面类真实商品榜。优先确认一个可接通的平台与榜单（京东、天猫或其他合法授权平台），分别说明Top100和Top400可达范围。请回复以下5问，并提供已有样例或字段合同，不发起收费测试。'
    unsupported = '如回复不支持，请提供注明日期、提供方和具体能力范围的书面结论，留作缺项依据；未公开授权能力无法确认时请写未知。后续来源或能力变化另记新观察，不将一次答复扩大为永久或全平台结论。'
    boundary = '公开OpenAPI只证明当前文档未声明该服务，不能代表全部商业授权能力。搜索排序、店铺排序、详情徽标及社媒热搜不能替代完整商品榜。来源、权限、完整范围及报价明确前，不生成榜单预算或发出采集请求。首期通过后再确认第二期同口径验证。'
    form_note = '以下均为待回填空表，不表示资料已经收到或能力已经确认。政策状态填写已收待核、未收或不适用；榜单问题填写已回复待核、未回复或不适用。不适用须说明依据；提供方结论可为支持、不支持或未知。收到文件不等于核验通过，书面能力答复也不等于已取得榜单。'
    file_note = '文件或答复栏请填文件名或书面答复位置；每个文件分别登记实际字节数、完整64位SHA-256及取得时间（含时区）。同项多个文件可另续行，正文与附件分开登记。没有文件时不编造字节数或SHA；写不适用并记录书面答复日期和依据。'
    next_note = '政策原件及附件核验后再关闭相应待核条款。榜单先验收一项真实来源的完整首期，再依同口径安排第二期。目录、价格、详情、评价补采及持续监听仍按逐商品计划和明确预算推进；本确认单不代表全量主线完成。'

    md = ['# 原件补充与榜单来源确认单', '', '版本 v2　2026年11月8日', '', intro, '',
          '## 政策原件补充', '', policy_intro, '',
          '| 编号 | 文件ID | 完整标题与来源 | 待核条款数 |', '| --- | --- | --- | ---: |']
    for i, r in enumerate(policies, 1):
        md.append(f"| P{i:02} | {r['documentId']} | [{r['title']}]({r['sourceUrl']}) | {r['blockedClauses']} |")
    md += ['', priority, '', evidence, '', screenshots, '', versions, '', '## 商品榜来源确认', '', ranking_intro, '']
    md += [f'{n}. {text}' for n, text in questions]
    md += ['', unsupported, '', boundary, '', '## 收件后验证', '', next_note, '', '## 逐项回执', '', form_note, '', file_note, '',
           '| 关联项 | 收件或答复状态 | 文件名或书面答复 | 字节数 | SHA-256 | 取得时间含时区 |',
           '| --- | --- | --- | ---: | --- | --- |']
    receipt_rows = []
    for i, r in enumerate(policies, 1):
        key = f'P{i:02}'
        receipt_rows.append(dict(itemId=key, kind='policy_original', documentId=r['documentId'],
                                 title=r['title'], sourceUrl=r['sourceUrl'], receiptStatus='PENDING_REPLY',
                                 verificationStatus='NOT_REVIEWED', evidenceFiles=[], notApplicableReason=None))
    for n, text in questions:
        receipt_rows.append(dict(itemId=f'Q{int(n):02}', kind='ranking_question', question=text,
                                 receiptStatus='PENDING_REPLY', providerConclusion=None,
                                 writtenReply=None, repliedAt=None, evidenceFiles=[],
                                 verificationStatus='NOT_REVIEWED', notApplicableReason=None))
    md += [f"| {r['itemId']} | 待回填 |  |  |  |  |" for r in receipt_rows]
    md += ['', '附件文件字段：fileName、itemId、sourceUrl、bytes、sha256、obtainedAt（含时区）、evidenceType、verificationStatus。多个附件可另附文件清单。', '']

    doc = Document()
    section = doc.sections[0]
    section.page_width, section.page_height = Cm(21), Cm(29.7)
    section.top_margin, section.bottom_margin = Cm(1.8), Cm(1.8)
    section.left_margin, section.right_margin = Cm(1.8), Cm(1.8)
    section.header_distance, section.footer_distance = Cm(0.8), Cm(0.8)
    normal = doc.styles['Normal']
    normal.font.name, normal.font.size = 'Microsoft YaHei', Pt(10.5)
    normal.element.get_or_add_rPr().get_or_add_rFonts().set(qn('w:eastAsia'), 'Microsoft YaHei')
    for name in ('Title', 'Heading 1', 'Heading 2'):
        doc.styles[name].font.color.rgb = RGBColor(0, 0, 0)
    for style in doc.styles:
        for border in style.element.xpath('.//w:pBdr'):
            border.getparent().remove(border)
    para(doc, '原件补充与榜单来源确认单', 19, True, True, 'Title')
    para(doc, '版本 v2　2026年11月8日', 9)
    para(doc, intro)
    heading(doc, '政策原件补充')
    para(doc, policy_intro)
    for i, r in enumerate(policies, 1):
        para(doc, f"P{i:02}　{r['title']}", 10.5, True, True)
        p = para(doc, f"文件ID　{r['documentId']}　待核条款 {r['blockedClauses']} 条", 8.5, keep=True)
        p.paragraph_format.space_after = Pt(2)
        p = para(doc, r['sourceUrl'], 8.5)
        p.paragraph_format.space_after = Pt(7)
    for text in (priority, evidence, screenshots, versions):
        para(doc, text)
    heading(doc, '商品榜来源确认')
    para(doc, ranking_intro)
    for n, text in questions:
        para(doc, n+'　'+text)
    for text in (unsupported, boundary):
        para(doc, text)
    heading(doc, '收件后验证')
    para(doc, next_note, 10)
    doc.add_page_break()
    heading(doc, '逐项回执')
    para(doc, form_note, 10)
    para(doc, file_note, 10)
    table = doc.add_table(rows=1, cols=6)
    table.autofit = False
    widths = [Cm(v) for v in (1.15, 2.1, 4.4, 1.5, 4.6, 3.65)]
    for column, width in zip(table.columns, widths):
        column.width = width
    headers = ['关联项', '收件或\n答复状态', '文件名或\n书面答复', '字节数', 'SHA-256', '取得时间\n含时区']
    for cell, width, text in zip(table.rows[0].cells, widths, headers):
        cell.width = width
        font(cell.paragraphs[0].add_run(text), 9, True)
        cell.paragraphs[0].alignment = WD_ALIGN_PARAGRAPH.CENTER
        shade = OxmlElement('w:shd'); shade.set(qn('w:fill'), 'E7EDF3')
        cell._tc.get_or_add_tcPr().append(shade)
    repeat = OxmlElement('w:tblHeader')
    table.rows[0]._tr.get_or_add_trPr().append(repeat)
    for r in receipt_rows:
        cells = table.add_row().cells
        for cell, width in zip(cells, widths):
            cell.width = width
        font(cells[0].paragraphs[0].add_run(r['itemId']), 9)
        font(cells[1].paragraphs[0].add_run('待回填'), 9)
    for row in table.rows:
        cant_split = OxmlElement('w:cantSplit')
        row._tr.get_or_add_trPr().append(cant_split)
        for i, cell in enumerate(row.cells):
            cell.vertical_alignment = WD_CELL_VERTICAL_ALIGNMENT.CENTER
            props = cell._tc.get_or_add_tcPr()
            margins = OxmlElement('w:tcMar')
            for side, value in (('top', 70), ('bottom', 70), ('left', 85), ('right', 85)):
                edge = OxmlElement('w:'+side); edge.set(qn('w:w'), str(value)); edge.set(qn('w:type'), 'dxa'); margins.append(edge)
            props.append(margins)
            borders = OxmlElement('w:tcBorders')
            for side in ('top', 'left', 'bottom', 'right'):
                edge = OxmlElement('w:'+side)
                for key, value in (('val', 'single'), ('sz', '4'), ('color', 'D9D9D9')):
                    edge.set(qn('w:'+key), value)
                borders.append(edge)
            props.append(borders)
            for p in cell.paragraphs:
                if not p.runs:
                    font(p.add_run(''), 9)
                p.paragraph_format.space_after = Pt(0)
                p.paragraph_format.space_before = Pt(0)
                p.paragraph_format.line_spacing = 1.0
                snap = OxmlElement('w:snapToGrid'); snap.set(qn('w:val'), 'false')
                p._p.get_or_add_pPr().append(snap)
                if i in (0, 1, 3, 5):
                    p.alignment = WD_ALIGN_PARAGRAPH.CENTER
    para(doc, 'P01至P12对应上文政策文件，Q01至Q05对应上文商品榜5问。多个文件可续行；附件逐件记录。回执模板JSON中已保留完整文件ID及问题文本。', 9)
    footer = section.footer.paragraphs[0]
    footer.alignment = WD_ALIGN_PARAGRAPH.RIGHT
    font(footer.add_run('v2　|　'), 8)
    field = OxmlElement('w:fldSimple'); field.set(qn('w:instr'), 'PAGE')
    footer._p.append(field)
    doc.core_properties.title = '原件补充与榜单来源确认单'
    doc.core_properties.subject = '12份政策原件和商品榜来源5问及逐项回执'
    doc.core_properties.author = ''
    OUT.mkdir(parents=True)
    (OUT/'原件补充与榜单来源确认单-v2.md').write_text('\n'.join(md), encoding='utf-8')
    doc.save(OUT/'原件补充与榜单来源确认单-v2.docx')
    save_json(OUT/'逐项回执模板-v2.json', dict(schema='ksf.industry.source-confirmation-return-template.v2',
        templateOnly=True, newlyReceivedOriginals=0, newlyReceivedRankingSnapshots=0,
        items=receipt_rows, evidenceFileFields=['fileName', 'itemId', 'sourceUrl', 'bytes', 'sha256',
                                              'obtainedAt', 'evidenceType', 'verificationStatus'],
        fileFieldRules='bytes为实际非负字节数，sha256为完整64位值，obtainedAt带时区；无文件保持空，不编造。'))
    save_json(WORK/'authoring-verification.json', dict(ok=True, inputSha256=sha(raw), inputManifest=3,
        markdownExactByteMatch=original == incoming_md, markdownNormalizedTextMatch=True,
        inputDocxFullDocumentIds=0, outputDocxFullDocumentIds=12, policyEntries=12,
        pendingClauses=19, rankingQuestions=5, emptyReceiptItems=17,
        additions=['截图辅助证据与完整网页存档候选分列', '不支持答复包含日期与范围', '逐项回执与文件证据字段', '补齐Word版完整文件ID'],
        createdAt=datetime.now(timezone.utc).isoformat(), paidCalls=0,
        externalMessagesSent=0, researchDatabaseWrites=0, productionChanges=0))
    print(json.dumps(dict(ok=True, output=str(OUT), policyEntries=12, rankingQuestions=5, receiptItems=17), ensure_ascii=False))


if __name__ == '__main__':
    main()