kangshifu-board/scripts/package-new-workshop-agent.py

"""Package the new workshop Agent handoff documents; perform no deployment."""
from pathlib import Path
import hashlib
import json
import os
import re
import tarfile
import zipfile

ROOT = Path(__file__).resolve().parents[1]
DOCS = Path(os.environ.get('KSF_NEW_AGENT_HANDOFF_DIR', 'F:/文档/新数字人交接'))
DEST = ROOT / 'work/delivery-by-recipient-20261102'
DOCUMENT_NAMES = [
    'README.md',
    '00-你需要手动做什么.txt',
    '01-发给运维001的接入说明.md',
    '02-新数字人角色说明.txt',
    '03-新数字人工作手册.md',
    '04-现有成果与接手范围.md',
    '05-验收与回报清单.md',
    '06-安装材料索引.json',
]


def digest(data):
    return hashlib.sha256(data).hexdigest()


def require(condition, message):
    if not condition:
        raise ValueError(message)


def manifest_entries(data):
    result = {}
    for line in data.decode('utf-8-sig').splitlines():
        if not line.strip():
            continue
        match = re.fullmatch(r'([a-f0-9]{64})  (.+)', line)
        require(match is not None, 'Malformed manifest')
        sha, name = match.groups()
        require(name not in result, 'Duplicate manifest entry')
        require(not name.startswith('/') and '..' not in name.split('/'), 'Unsafe member')
        result[name] = sha
    return result


def read_package(path, expected_sha, expected_bytes, manifest_name, expected_count, prefix=''):
    data = path.read_bytes()
    require(len(data) == expected_bytes and digest(data) == expected_sha, f'Package mismatch: {path.name}')
    if path.suffix == '.tgz':
        with tarfile.open(path) as archive:
            members = archive.getmembers()
            require(all(not m.issym() and not m.islnk() for m in members), 'Unexpected archive link')
            files = {m.name: archive.extractfile(m).read() for m in members if m.isfile()}
    else:
        with zipfile.ZipFile(path) as archive:
            require(archive.testzip() is None, 'Corrupt input ZIP')
            require(len(archive.namelist()) == len(set(archive.namelist())), 'Duplicate ZIP members')
            files = {name: archive.read(name) for name in archive.namelist() if not name.endswith('/')}
    entries = manifest_entries(files[manifest_name])
    require(len(entries) == expected_count, 'Unexpected manifest size')
    for name, sha in entries.items():
        require(digest(files[prefix + name]) == sha, f'Member mismatch: {name}')
    return files, {
        'filename': path.name,
        'bytes': len(data),
        'sha256': digest(data),
        'knownLocalPath': path.as_posix(),
        'manifestPath': manifest_name,
        'manifestVerified': f'{len(entries)}/{expected_count}',
        'includedInHandoffZip': False,
    }


def main():
    v4, v4_info = read_package(
        Path('D:/下载/WXWork/1688857450794210/Cache/File/2026-11/ksf-adapter-patch-v4-20261102.tgz'),
        'fc1888c99160e8af28a69355e78c91d1f4f8b3f612c1cc0f0fb0d1f4f675f0a5', 65842,
        'adapter-patch-v4/MANIFEST.sha256', 21, 'adapter-patch-v4/',
    )
    dedup, dedup_info = read_package(
        DEST / 'ksf-workshop-dedup-20261102.zip',
        '96fc5c10557b7601b63767e293bcfc33fcea1058f21961949348c105cd75afde', 339939,
        'FILES.sha256', 34,
    )
    hotfix, hotfix_info = read_package(
        DEST / 'ksf-wecom-tool-result-hotfix-20261102.zip',
        'ea1cd7e2663b8d67f0ca12381c131bddd4edd51e45518fb55fb04c4d836538e8', 39426,
        'FILES.sha256', 14,
    )
    skill_prefix = 'skills/ksf-workshop-runtime/'
    skill_entries = manifest_entries(dedup[skill_prefix + 'FILES.sha256'])
    require(len(skill_entries) == 10, 'Skill manifest count changed')
    for name, sha in skill_entries.items():
        require(digest(dedup[skill_prefix + name]) == sha, 'Skill manifest mismatch')

    final_files = []
    for package_info, files, member, target in [
        (v4_info, v4, 'adapter-patch-v4/aibot_identity.py', '/opt/hermes/plugins/platforms/wecom/aibot_identity.py'),
        (v4_info, v4, 'adapter-patch-v4/patches/adapter.py.patched', '/opt/hermes/plugins/platforms/wecom/adapter.py'),
        (v4_info, v4, 'adapter-patch-v4/plugin/wecom-identity/plugin.yaml', '/opt/data/plugins/wecom-identity/plugin.yaml'),
        (v4_info, v4, 'adapter-patch-v4/plugin/wecom-identity/__init__.py', '/opt/data/plugins/wecom-identity/__init__.py'),
        (hotfix_info, hotfix, 'integrations/wecom-identity/identity_bridge.py', '/opt/hermes/plugins/platforms/wecom/identity_bridge.py'),
        (hotfix_info, hotfix, 'integrations/wecom-identity/workshop_registration.py', '/opt/hermes/plugins/platforms/wecom/workshop_registration.py'),
    ]:
        final_files.append({'package': package_info['filename'], 'member': member,
                            'targetForVerifiedLayout': target, 'sha256': digest(files[member])})

    index = {
        'version': 'ksf.new-workshop-agent-handoff.v1',
        'preparedDate': '2026-11-02',
        'purpose': '新实例接入材料索引；只核对原件与清单，不是首次开通器或部署脚本',
        'suggestedNameOnly': '康师傅工作坊助手001',
        'newAgentCreated': False,
        'newAgentRegistered': False,
        'newTargetChanged': False,
        'configuredNameTriggersRegistration': False,
        'platform': 'Hermes + WeCom aibot WebSocket',
        'verifiedLayout': {'hermesVersion': '0.21.3', 'gatewayUser': 'hermes', 'uid': 10000,
                           'hermesHome': '/opt/data', 'requiresNewTargetVerification': True},
        'packages': [v4_info, dedup_info, hotfix_info],
        'adapterRequiredBaselineSha256': digest(v4['adapter-patch-v4/baseline/adapter.py']),
        'finalPlatformFiles': final_files,
        'skill': {'package': dedup_info['filename'], 'prefix': skill_prefix,
                  'files': skill_entries, 'manifestVerified': '10/10',
                  'metadataManifestSha256': digest(dedup[skill_prefix + 'FILES.sha256'])},
        'toolNames': ['wecom_identity_current', 'ksf_register_current_employee'],
        'contractMarkers': ['ksf.agent-self.v1', 'ksf.wecom-aibot-sender.v1', 'ksf.ingest-dedup.v1'],
        'operatorNotes': [
            '原安装包不包含在本交接ZIP；运维可按名称和完整指纹复用已收到的原件。',
            '旧收据的未部署状态仅指打包时；旧测试实例已验，新实例仍待验。',
            '现有后台先回读，不运行旧身份部署覆盖当前查重合同。',
            'install-workshop-skill.mjs是升级器；新实例运行时/独立凭据须经受控首次开通。',
            '新bot账号与最终名称需一次受控Agent建档及registration.json绑定；名称本身不触发登记。',
            '不复制旧私钥、信封、credential-binding、Agent/员工状态、instance或context-K。',
            '全流程与查重脚本含旧固定任务键；新实例不得原样重放。',
            '本次核验是档案和指纹核验；不代表新实例功能验收已通过。',
        ],
    }
    (DOCS / '06-安装材料索引.json').write_text(json.dumps(index, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')

    files = {name: (DOCS / name).read_bytes() for name in DOCUMENT_NAMES}
    require(all(data for data in files.values()), 'Empty handoff document')
    for name in ['00-你需要手动做什么.txt', '01-发给运维001的接入说明.md',
                 '02-新数字人角色说明.txt', '03-新数字人工作手册.md']:
        text = files[name].decode('utf-8')
        require('ksf_register_current_employee' in text, 'Missing real tool name')
        require('ksfregistercurrent' not in text, 'Incorrect tool alias')
    for text in files.values():
        require(b'-----BEGIN PRIVATE KEY-----' not in text, 'Unexpected private key')
    files['FILES.sha256'] = ''.join(f'{digest(data)}  {name}\n' for name, data in files.items()).encode('utf-8')
    (DOCS / 'FILES.sha256').write_bytes(files['FILES.sha256'])
    archive_path = DOCS / 'ksf-new-workshop-agent-handoff-20261102.zip'
    with zipfile.ZipFile(archive_path, 'w', compression=zipfile.ZIP_DEFLATED) as archive:
        for name, data in files.items():
            info = zipfile.ZipInfo(name, date_time=(2026, 11, 2, 0, 0, 0))
            info.compress_type = zipfile.ZIP_DEFLATED
            info.external_attr = 0o100644 << 16
            archive.writestr(info, data)
    with zipfile.ZipFile(archive_path) as archive:
        require(archive.testzip() is None, 'Corrupt output ZIP')
        require(set(archive.namelist()) == set(files), 'Unexpected output member')
        entries = manifest_entries(archive.read('FILES.sha256'))
        require(len(entries) == 8, 'Unexpected handoff document count')
        for name, sha in entries.items():
            require(digest(archive.read(name)) == sha, 'Handoff manifest mismatch')

    receipt = {
        'file': archive_path.name, 'bytes': archive_path.stat().st_size,
        'sha256': digest(archive_path.read_bytes()), 'archiveIntegrity': True,
        'manifestVerified': '8/8', 'zipEntries': 9,
        'originalPackageManifestsVerified': ['21/21', '34/34', '14/14'],
        'skillManifestVerified': '10/10', 'contents': 'handoff-documents-only',
        'includesInstallationArchives': False, 'includesCredentials': False,
        'newAgentCreated': False, 'newAgentRegistered': False, 'newTargetChanged': False,
    }
    (DOCS / 'ksf-new-workshop-agent-handoff-20261102.receipt.json').write_text(
        json.dumps(receipt, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
    print(json.dumps(receipt, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    main()