kangshifu-board/scripts/package-dedup-acceptance.py

"""Build the standalone sample dedup acceptance delivery without private inputs."""
from pathlib import Path
import hashlib
import json
import zipfile

ROOT = Path(__file__).resolve().parents[1]
DEST = ROOT / 'work/delivery-by-recipient-20261102'
KEY = 'KSF-查重实测-20261102-01'
OPS = '''运维001：查重验收选A，专用执行器已准备好。只落地脚本和输入，不重启、不改Skill/Function、不代跑业务。

1. 在独立目录解包，核验FILES.sha256。对照本次交付ZIP指纹，不改清单。
2. 目标仍是R730 / agent-node-u17879018108。执行器安装到 /opt/data/ksf-executor-20261102/workshop-dedup-acceptance.mjs，root:root 0644；用目标Node做 node --check。不要覆盖workshop-full-chain.mjs。
3. 模板context-dedup.example.json的agentProfileKey=null故意不可运行。仅从已核实的context-K.json取agentProfileKey字段，或回读已核实当前Agent档案键；核对和现有registration.json一致，不新建身份。以hermes(10000)、HOME=/opt/data写 /opt/data/ksf-workshop-private/context-dedup-01.json，0600，字段仅version/taskKey/agentProfileKey。不输出完整上下文。
4. 以hermes建父目录 /opt/data/ksf-workshop-private/dedup，0700；检查目标 /opt/data/ksf-workshop-private/dedup/KSF-dedup-20261102-01 不存在。有任何痕迹先核查，不删除目录重跑，不另换任务键，不碰full-chain-K。
5. 落地后回报脚本SHA、语法检查、输入属主权限、状态目录不存在，交王刚001在真实消息任务中执行。你不在运维身份下执行入库。

目标侧执行会产生一份sample来源/原料和两批导入审计。重复实体复用不代表数据库完全不增长。详情见docs/workshop-dedup-acceptance-20261102.md。
'''
WANG = '''王刚001：本次执行专用查重验收，任务键固定 KSF-查重实测-20261102-01。待运维完成落地后，在本条真实消息授权的任务中，以hermes(10000)、HOME=/opt/data只执行下面这一条命令一次：

node /opt/data/ksf-executor-20261102/workshop-dedup-acceptance.mjs --input /opt/data/ksf-workshop-private/context-dedup-01.json --state-dir /opt/data/ksf-workshop-private/dedup/KSF-dedup-20261102-01 --execute

脚本固定样例内容，只换两轮requestId和externalKey；不创建工作坊会话或报告、不重跑K轮、不手工构造身份。首次导入预期created=1，第二次created=0/unchanged=1，且sourceId/materialId前后相同。unchanged就是本次要验证的复用。

成功回报：ok、operation、taskKey，两轮sourceId/materialId/jobId/created/unchanged，两次import invocationId及状态目录。不贴完整输入/日志/账号。成功operation应为sample-material-dedup-verified。

失败立即停，保留原键、目录和全部回执，不自动重试、不删记录、不换目录重跑；报最后一步、错误码及已返回编号。错误不等于零写入，超时也可能已入库。当前仅授权本样例验收，不宣称真实平台采集或正式报告通过。
'''

def main():
    DEST.mkdir(parents=True, exist_ok=True)
    files = {
        'scripts/workshop-dedup-acceptance.mjs': (ROOT / 'scripts/workshop-dedup-acceptance.mjs').read_bytes(),
        'docs/workshop-dedup-acceptance-20261102.md': (ROOT / 'docs/workshop-dedup-acceptance-20261102.md').read_bytes(),
        'context-dedup.example.json': (json.dumps({'version': 'ksf.dedup-acceptance.v1', 'taskKey': KEY, 'agentProfileKey': None}, ensure_ascii=False, indent=2) + '\n').encode(),
        '给运维001-查重验收落地.txt': OPS.encode(),
        '给王刚001-查重验收执行.txt': WANG.encode(),
    }
    digest = lambda b: hashlib.sha256(b).hexdigest()
    files['FILES.sha256'] = ''.join(f'{digest(b)}  {name}\n' for name,b in files.items()).encode()
    target = DEST / 'ksf-dedup-acceptance-20261102.zip'
    with zipfile.ZipFile(target, 'w', compression=zipfile.ZIP_DEFLATED) as z:
        for name,b in files.items():
            z.writestr(name,b)
    with zipfile.ZipFile(target) as z:
        assert z.testzip() is None
        assert len(z.namelist()) == 6
        for line in z.read('FILES.sha256').decode().splitlines():
            sha,name = line.split('  ',1)
            assert digest(z.read(name)) == sha
        assert json.loads(z.read('context-dedup.example.json'))['agentProfileKey'] is None
        assert z.read('scripts/workshop-dedup-acceptance.mjs') == files['scripts/workshop-dedup-acceptance.mjs']
    for name in ['给运维001-查重验收落地.txt', '给王刚001-查重验收执行.txt']:
        (DEST / name).write_bytes(files[name])
    receipt = {'file': target.name, 'bytes': target.stat().st_size, 'sha256': digest(target.read_bytes()), 'manifestChecked': '5/5', 'archiveIntegrity': True, 'executorSha256': digest(files['scripts/workshop-dedup-acceptance.mjs']), 'containsCredentials': False, 'targetExecuted': False}
    (DEST / 'ksf-dedup-acceptance-20261102.receipt.json').write_text(json.dumps(receipt, ensure_ascii=False, indent=2),encoding='utf-8')
    print(json.dumps(receipt, ensure_ascii=False, indent=2))

if __name__ == '__main__':
    main()