kangshifu-board/scripts/package-voc-database.py

"""Prepare a database-only supplement for the already installed Nov 6 query skills."""
import argparse
import hashlib
import importlib.util
import json
from pathlib import Path
import shutil
import sys
import tempfile
import zipfile

ROOT = Path(__file__).resolve().parents[1]
WORK = ROOT / 'work/voc-database-delivery-20261106'
OUT = ROOT / 'artifacts/研究数据库补装-20261106'
NAME = 'ksf-voc-database-20261106-r1'
RELEASE = OUT / NAME
DIST = Path('F:/文档/康师傅skill')
SKILL = ROOT / 'skills/ksf-voc-library'
EXAMPLES = ROOT / 'artifacts/数据库精准检索规则-20261106/ksf-agent-query-rules-20261106-r1/examples'


def digest(path):
    with Path(path).open('rb') as stream:
        return hashlib.file_digest(stream, 'sha256').hexdigest()


def write(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')


def manifest(folder):
    entries = {p.relative_to(folder).as_posix(): digest(p) for p in sorted(folder.rglob('*')) if p.is_file() and p.name != 'FILES.sha256'}
    (folder / 'FILES.sha256').write_text(''.join(sha + '  ' + name + '\n' for name, sha in entries.items()), encoding='utf-8', newline='\n')


def load_module(name, path):
    spec = importlib.util.spec_from_file_location(name, path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def prepare():
    if RELEASE.exists():
        raise ValueError('FROZEN_RELEASE_ALREADY_EXISTS')
    sys.path.insert(0, str(ROOT / 'server/voc-library'))
    import voc_db
    registry = ROOT / 'work/library-preview/registry'
    pointer = json.loads((registry / 'current.json').read_text(encoding='utf-8'))
    source = voc_db.registry_database(registry)
    if digest(source) != pointer['databaseSha256']:
        raise ValueError('LOCAL_BOARD_POINTER_HASH_MISMATCH')
    with voc_db.readonly(source) as con:
        meta = dict(con.execute('SELECT key,value FROM meta'))
        entity_counts = voc_db.table_counts(con)
        counts = {**entity_counts, 'retrieval_documents': con.execute('SELECT count(*) FROM retrieval_documents').fetchone()[0]}
    database = dict(schema=meta['schema'], generation=meta['generation'], bytes=source.stat().st_size, sha256=digest(source),
        originalArchiveSha256=meta['archive_sha256'], entityCounts=entity_counts, counts=counts)
    destination = RELEASE / 'library/library.sqlite'
    destination.parent.mkdir(parents=True)
    shutil.copyfile(source, destination)
    if digest(destination) != pointer['databaseSha256']:
        raise ValueError('COPIED_DATABASE_HASH_MISMATCH')
    required_skill = digest(SKILL / 'FILES.sha256')
    if required_skill != '4da410bb882dc73e277c740d13407918d9c9d5da2196ac9211f6ff7c009692fb':
        raise ValueError('INSTALLED_QUERY_RELEASE_NO_LONGER_MATCHES')
    write(RELEASE / 'DATABASE.json', dict(release=NAME, database=database,
        requiredResearchSkillManifestSha256=required_skill, source='Codex local board current database snapshot',
        matchesLocalBoardDatabaseBytes=True, storageMode='static-snapshot', automaticUpdatesConfigured=False,
        productionInstalled=False, productionBoardAligned=False, realMessageVerified=False,
        skillIncluded=False, skillInstallationRequired=False, defaultDatabasePath='/opt/data/ksf-voc-library/library.sqlite'))
    for name in ('install-voc-database.py', 'verify-installed-voc-database.py'):
        shutil.copyfile(ROOT / 'scripts' / name, RELEASE / name)
    shutil.copytree(EXAMPLES, RELEASE / 'examples')
    shutil.copytree(ROOT / 'docs/voc-database-delivery-20261106', RELEASE / 'docs')
    manifest(RELEASE)
    print(json.dumps(dict(prepared=str(RELEASE), database=database), ensure_ascii=False, indent=2))


def validate():
    WORK.mkdir(parents=True, exist_ok=True)
    installer = load_module('seed_install', RELEASE / 'install-voc-database.py')
    verifier = load_module('seed_verify', RELEASE / 'verify-installed-voc-database.py')
    with tempfile.TemporaryDirectory(prefix='simulate-', dir=WORK) as temporary:
        root = Path(temporary)
        target = root / 'ksf-voc-library'
        plan = installer.install(RELEASE, target)
        if target.exists() or not plan['readOnly']:
            raise ValueError('DRY_PLAN_WROTE_TARGET')
        applied = installer.install(RELEASE, target, True)
        landed = target / 'library.sqlite'
        before = (digest(landed), landed.stat().st_mtime_ns)
        repeated = installer.install(RELEASE, target, True)
        if repeated['applied'] or (digest(landed), landed.stat().st_mtime_ns) != before:
            raise ValueError('REPEAT_CHANGED_DATABASE')
        verification = verifier.verify(RELEASE, SKILL, landed)
        if (digest(landed), landed.stat().st_mtime_ns) != before:
            raise ValueError('READBACK_CHANGED_DATABASE')
        result = dict(ok=True, package=NAME, installPlan=plan, installSimulation=applied,
            exactRepeatVerified=True, installedReadback=verification,
            skillFilesChanged=False, productionInstalled=False, realMessageVerified=False)
        write(WORK / 'local-validation.json', result)
        print(json.dumps(dict(ok=True, generation=verification['generation'], scenarios=len(verification['scenarios']),
            sourceChecks=len(verification['citations']), installedDefaultPathSimulated=True, exactRepeatVerified=True), ensure_ascii=False, indent=2))


def finalize():
    validation = json.loads((WORK / 'local-validation.json').read_text(encoding='utf-8'))
    if not validation['ok']:
        raise ValueError('VALIDATION_REQUIRED')
    package_meta = json.loads((RELEASE / 'DATABASE.json').read_text(encoding='utf-8'))
    if digest(RELEASE / 'library/library.sqlite') != validation['installedReadback']['databaseSha256'] or package_meta['database']['generation'] != validation['installedReadback']['generation']:
        raise ValueError('DATABASE_CHANGED_AFTER_VALIDATION')
    archive = OUT / (NAME + '.zip')
    if archive.exists():
        raise ValueError('ZIP_ALREADY_FROZEN')
    shutil.copyfile(WORK / 'local-validation.json', RELEASE / 'local-validation.json')
    manifest(RELEASE)
    with zipfile.ZipFile(archive, 'x', zipfile.ZIP_DEFLATED) as z:
        for path in sorted(RELEASE.rglob('*')):
            if path.is_file():
                z.write(path, path.relative_to(RELEASE).as_posix())
    with zipfile.ZipFile(archive) as z:
        entries = z.read('FILES.sha256').decode('utf-8').splitlines()
        for line in entries:
            expected, name = line.split('  ', 1)
            if hashlib.sha256(z.read(name)).hexdigest() != expected:
                raise ValueError('ZIP_READBACK_MISMATCH')
    receipt = dict(release=NAME, filename=archive.name, bytes=archive.stat().st_size, sha256=digest(archive),
        entries=len(entries)+1, rootManifestVerified=str(len(entries)) + '/' + str(len(entries)),
        database=json.loads((RELEASE / 'DATABASE.json').read_text(encoding='utf-8'))['database'],
        localValidated=True, databaseIncluded=True, skillIncluded=False, productionInstalled=False,
        realMessageVerified=False, productionBoardAligned=False)
    write(OUT / 'delivery-receipt.json', receipt)
    shutil.copyfile(archive, DIST / archive.name)
    shutil.copyfile(OUT / 'delivery-receipt.json', DIST / '研究数据库补包校验-20261106.json')
    note = '''
## 研究库补装 · 2026-11-06

已收到运维与王刚001回报：本目录11月6日两份Skill已安装且实际加载；研究数据库此前未安装，查询停在DATABASE_NOT_INSTALLED。对此不需要重装或回退Skill。

请只转交 **ksf-voc-database-20261106-r1.zip** 与其包内docs/发给运维001-补装研究数据库.md；校验见研究数据库补包校验-20261106.json。该包只含数据库本体、安装/验收工具和样例，不含Skill，装到既定默认路径，无须新凭据或改网关。

补包取自Codex本地看板当前快照；尚未安装到远端，尚未对齐远端生产看板或配置后续自动同步。数据库装好再由王刚001继续原五步真实消息验收。原r2整包中的旧Skill安装器不要再执行。
'''
    readme = (DIST / 'README.md').read_text(encoding='utf-8')
    readme = readme.replace('不含数据库，沿用已安装v2研究库。缺库时需先用原完整r2资料包建立授权库；不能只复制Skill就声称能查到资料。',
        '查询规则包不含数据库。缺库时用本目录新增的数据库专用补包落地，不再执行原r2整包中的旧Skill安装器。')
    readme = readme.replace('本机准备和验证完成，未在王刚001生产实例安装/加载/真实消息验收。请将当前完整ZIP与包内docs/发给运维001.md一起转运维；安装后将docs/发给王刚001.md交本人真实消息验证。',
        '已收到运维001报告两份Skill安装完成，王刚001确认真实加载；数据库此前未安装，真实数据查询尚未验收。请按下方数据库补装说明继续，不重复安装Skill。')
    (DIST / 'README.md').write_text(readme + note, encoding='utf-8')
    version_file = DIST / '版本与校验.json'
    version = json.loads(version_file.read_text(encoding='utf-8'))
    version['deploymentReport20261106'] = dict(source='user forwarded Operations001 and WangGang001 reports',
        skillsInstalledReported=True, skillsLoadedReported=True, databaseInstalledReported=False,
        realMessageQueryVerified=False, interpretation='historical report before this database supplement; package preparation is not remote installation')
    version['databaseSupplement'] = dict(zip=receipt, installInstructions=str(ROOT / 'docs/voc-database-delivery-20261106/发给运维001-补装研究数据库.md'))
    write(version_file, version)
    handoff = Path('F:/文档/新数字人交接')
    (handoff / '11-研究数据库补装-20261106.md').write_text('''# 研究数据库补装 · 2026-11-06

11月6日查询规则包仅含Skill，研究数据库须独立落地。两份Skill已由运维报告安装并经王刚001实际加载，查询因DATABASE_NOT_INSTALLED未开始。

当前补装包：F:/文档/康师傅skill/ksf-voc-database-20261106-r1.zip；核验文件：同目录研究数据库补包校验-20261106.json。

只补数据库到默认路径，不重装旧r2 Skill，不改身份、会话、网关或生产看板。按包内运维单安装/回读后，再由用户将王刚001续验单转入真实企微消息。数据库源于本地看板同字节快照，远端共享存储和后续自动同步尚未接通。
''', encoding='utf-8')
    print(json.dumps({key: value for key, value in receipt.items() if key != 'database'}, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('phase', choices=['prepare', 'validate', 'finalize'])
    args = parser.parse_args()
    globals()[args.phase]()