"""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]()