kangshifu-board/scripts/release-workshop-board.py

"""Seal and publish only the verified nine-file workshop board release."""
import argparse
from concurrent.futures import ThreadPoolExecutor
import hashlib
import importlib.util
import io
import json
from pathlib import Path
import subprocess
import urllib.error
import urllib.request
import zipfile

ROOT = Path(__file__).resolve().parents[1]
NAME = 'ksf-workshop-board-20261110-r1'
OUT = ROOT / 'docs' / NAME
PACK = OUT / 'package'
STAGING = ROOT / 'work' / (NAME + '-publication')
BRANCH = 'codex/workshop-board-20261110-r1'
REMOTE = 'https://git.fmode.cn/18307996893/kangshifu-board.git'
PUBLIC = REMOTE[:-4]
sha = lambda raw: hashlib.sha256(raw).hexdigest()
encode = lambda obj: (json.dumps(obj, ensure_ascii=False, indent=2) + '\n').encode('utf-8')


def need(ok, code):
    if not ok:
        raise ValueError(code)


def run(args, cwd=ROOT):
    result = subprocess.run(args, cwd=cwd, capture_output=True, timeout=120)
    need(result.returncode == 0, 'COMMAND_FAILED:' + args[0])
    return result.stdout


def git(*args, cwd=STAGING):
    return run(['git', *args], cwd).decode('utf-8').strip()


def request(url):
    req = urllib.request.Request(url, headers={'User-Agent': 'ksf-board-release/1', 'Cache-Control': 'no-cache'})
    return urllib.request.urlopen(req, timeout=40)


def seal():
    need(not (OUT / 'publish').exists(), 'RELEASE_ALREADY_SEALED')
    manifest = json.loads((PACK / 'deploy/industry-ui-files.json').read_bytes())
    need(len(manifest['files']) == 9 and sum(f['allowMissing'] for f in manifest['files']) == 2, 'BAD_SCOPE')
    for f in manifest['files']:
        need(sha((PACK / 'payload' / f['path']).read_bytes()) == f['sha256'], 'PAYLOAD_CHANGED')
    def check_baseline(row):
        try:
            with request('https://kangshifu.fmode.cn/' + row['path']) as response:
                raw = response.read()
                if row.get('allowMissing'):
                    need('text/html' in response.headers.get('Content-Type', ''), 'NEW_MODULE_ALREADY_EXISTS')
                else:
                    digest = sha(raw.replace(b'\r\n', b'\n'))
                    expected = row.get('priorNormalizedSha256', [row['sha256']])
                    need(digest in expected, 'PRODUCTION_BASELINE_CHANGED:' + row['path'])
        except urllib.error.HTTPError as error:
            need(error.code == 404 and row.get('allowMissing'), 'PRODUCTION_BASELINE_UNAVAILABLE')
        return {'path': row['path'], 'observedBaselineStillValid': True}
    with ThreadPoolExecutor(max_workers=6) as pool:
        freshness = list(pool.map(check_baseline, [{'path': 'index.html', 'sha256': manifest['homepageSha256']}, *manifest['files']]))
    tests = run(['node', '--test', 'test/publisher.test.js', 'test/workshop-outputs.test.js'], PACK)
    need(b'pass 13' in tests and b'fail 0' in tests, 'RELEASE_TESTS_FAILED')
    (PACK / 'evidence/release-tests.log').write_bytes(tests)
    run(['C:/Program Files/Git/bin/bash.exe', '-n', 'scripts/deploy-site.sh'], PACK)
    scripts = [*list((PACK / 'payload').rglob('*.js')), *list((PACK / 'scripts').glob('*.mjs'))]
    for p in scripts:
        run(['node', '--check', str(p)])
    browser = json.loads((PACK / 'evidence/release-browser.json').read_bytes())
    api = json.loads((PACK / 'evidence/public-api-acceptance.json').read_bytes())
    need(browser['checksPassed'] == 15 and not browser['errors'] and browser['businessWrites'] == 0, 'BROWSER_CHECK_REQUIRED')
    need(api['publicReadbackPassed'] and api['originalPublicRecordsUnchanged'] and api['sharedGroupStateUnchanged'], 'API_CHECK_REQUIRED')
    validation = {'ok': True, 'release': NAME, 'releaseTestsPassed': 13, 'browserCompatibilityChecksPassed': 15,
                  'browserErrors': 0, 'browserMethod': browser['method'], 'staticArchitectureCheckPassed': True,
                  'payloadAndPublisherSyntaxPassed': len(scripts), 'deploymentShellSyntaxPassed': True,
                  'backendMinimalPatchTestsPassed': 3, 'backendAlreadyDeployedAndReadBack': True,
                  'originalPublicRecordsAndGroupsUnchanged': True, 'prepublicationBaselineRechecked': freshness,
                  'businessWrites': 0, 'mainSkillIncluded': False, 'manualIncluded': False,
                  'productionStaticFilesDeployed': False, 'realEmployeeOrHostTrial': False,
                  'linuxSitePermissionsAcceptance': 'run on production by deploy-site.sh'}
    (PACK / 'validation.json').write_bytes(encode(validation))
    files = {p.relative_to(PACK).as_posix(): p.read_bytes() for p in PACK.rglob('*') if p.is_file()}
    need(not any(n.startswith(('skills/', 'work/', 'private/')) for n in files), 'UNEXPECTED_RELEASE_CONTENT')
    files['FILES.sha256'] = ''.join(sha(raw) + '  ' + name + '\n' for name, raw in sorted(files.items())).encode('utf-8')
    (PACK / 'FILES.sha256').write_bytes(files['FILES.sha256'])
    stream = io.BytesIO()
    with zipfile.ZipFile(stream, 'w', zipfile.ZIP_DEFLATED) as archive:
        for name, raw in sorted(files.items()):
            info = zipfile.ZipInfo(name, date_time=(2026, 11, 10, 0, 0, 0))
            info.external_attr = 0o100644 << 16
            archive.writestr(info, raw, compress_type=zipfile.ZIP_DEFLATED)
    raw_zip = stream.getvalue()
    with zipfile.ZipFile(io.BytesIO(raw_zip)) as archive:
        need(archive.testzip() is None, 'ARCHIVE_CORRUPT')
        for line in archive.read('FILES.sha256').decode().splitlines():
            digest, name = line.split('  ', 1)
            need(sha(archive.read(name)) == digest, 'ARCHIVE_HASH_MISMATCH')
    release = {'schema': 'ksf.workshop-board.release.v1', 'release': NAME,
               'archive': {'path': NAME + '.zip', 'bytes': len(raw_zip), 'sha256': sha(raw_zip)},
               'filesManifestEntries': len(files) - 1, 'frontendFiles': manifest['files'],
               'homepageSha256': manifest['homepageSha256'], 'deploymentScript': 'scripts/deploy-site.sh',
               'targetDirectory': '/opt/kangshifu-web', 'backendFieldAlreadyDeployed': 'reportSeriesKey',
               'mainSkillIncluded': False, 'manualIncluded': False, 'businessWrites': 0,
               'productionStaticDeploymentPending': True,
               'validation': {'testsPassed': 13, 'browserChecksPassed': 15, 'browserErrors': 0}}
    publish = {'release.json': encode(release), NAME + '.zip': raw_zip, 'DEPLOY.md': files['DEPLOY.md'],
               'validation.json': files['validation.json'], 'evidence/ui.patch': files['evidence/ui.patch'],
               'evidence/public-api-acceptance.json': files['evidence/public-api-acceptance.json'],
               'evidence/release-browser.json': files['evidence/release-browser.json'], '.gitattributes': b'* -text\n'}
    spec = importlib.util.spec_from_file_location('secret_checks', ROOT / 'scripts/publish-workshop-onboarding-r5.py')
    helper = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(helper)
    scan = helper.secret_scan(publish)
    (OUT / 'public-secret-check.json').write_bytes(encode(scan))
    for name, raw in publish.items():
        p = OUT / 'publish' / name
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_bytes(raw)
    print(json.dumps({'ok': True, 'sealed': True, **release['archive'], 'filesManifestEntries': release['filesManifestEntries'], 'secretCheck': scan}))


def prepare():
    need(not STAGING.exists(), 'NEW_STAGING_DIRECTORY_REQUIRED')
    files = {p.relative_to(OUT / 'publish').as_posix(): p.read_bytes() for p in (OUT / 'publish').rglob('*') if p.is_file()}
    need('release.json' in files and len(files) == 8, 'UNSEALED_RELEASE')
    STAGING.mkdir(parents=True)
    for name, raw in files.items():
        p = STAGING / name
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_bytes(raw)
    git('init', '--initial-branch=' + BRANCH)
    for name in ['user.name', 'user.email']:
        git('config', name, git('config', '--get', name, cwd=ROOT))
    git('config', 'core.autocrlf', 'false')
    git('remote', 'add', 'origin', REMOTE)
    git('add', '--', *sorted(files))
    git('commit', '-m', 'release: workshop group outputs, report navigation and version history')
    need(set(git('ls-files').splitlines()) == set(files), 'UNEXPECTED_PUBLICATION_FILES')
    for name, raw in files.items():
        need(run(['git', 'cat-file', 'blob', 'HEAD:' + name], STAGING) == raw, 'GIT_BLOB_CHANGED')
    plan = {'commit': git('rev-parse', 'HEAD'), 'branch': BRANCH,
            'files': {n: sha(raw) for n, raw in files.items()}, 'mainBranchModified': False}
    (OUT / 'publication-plan.json').write_bytes(encode(plan))
    print(json.dumps({'ok': True, 'prepared': True, 'commit': plan['commit'], 'files': len(files)}))


def publish():
    plan = json.loads((OUT / 'publication-plan.json').read_bytes())
    commit = plan['commit']
    need(git('rev-parse', 'HEAD') == commit and not git('status', '--porcelain'), 'STAGING_CHANGED')
    before = git('ls-remote', 'origin', 'refs/heads/' + BRANCH)
    need(not before or before.split()[0] == commit, 'REMOTE_BRANCH_CHANGED')
    git('push', 'origin', 'HEAD:refs/heads/' + BRANCH)
    need(git('ls-remote', 'origin', 'refs/heads/' + BRANCH).split()[0] == commit, 'PUSH_UNCONFIRMED')
    def verify(item):
        name, digest = item
        url = PUBLIC + '/raw/' + commit + '/' + name
        with request(url) as response:
            raw = response.read()
            need(response.status == 200 and sha(raw) == digest, 'PUBLIC_RESOURCE_HASH_MISMATCH')
            return {'path': name, 'url': url, 'bytes': len(raw), 'sha256': digest, 'http': response.status, 'matchesLocal': True}
    with ThreadPoolExecutor(max_workers=4) as pool:
        checked = list(pool.map(verify, plan['files'].items()))
    receipt = {'ok': True, 'release': NAME, 'commit': commit, 'branch': BRANCH,
               'manifestUrl': PUBLIC + '/raw/' + commit + '/release.json',
               'deploymentGuideUrl': PUBLIC + '/raw/' + commit + '/DEPLOY.md',
               'publicResourcesPublishedAndReadBack': True, 'resources': checked,
               'backendAlreadyDeployedAndVerified': True, 'productionStaticFilesDeployed': False,
               'mainSkillAndManualDeployed': False, 'mainBranchModified': False, 'businessWrites': 0}
    (OUT / 'online-release-receipt.json').write_bytes(encode(receipt))
    print(json.dumps({k: v for k, v in receipt.items() if k != 'resources'}, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('action', choices=['seal', 'prepare', 'publish'])
    args = parser.parse_args()
    {'seal': seal, 'prepare': prepare, 'publish': publish}[args.action]()