kangshifu-board/scripts/publish-workshop-manual.py

"""Publish the sealed static manual and deployment kit; this does not deploy the production website."""
import argparse
from concurrent.futures import ThreadPoolExecutor
import hashlib
import importlib.util
import json
from pathlib import Path
import subprocess
import urllib.request

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


def git(*args, cwd=STAGING):
    result = subprocess.run(['git', *args], cwd=cwd, capture_output=True, timeout=90)
    if result.returncode:
        raise RuntimeError('GIT_OPERATION_FAILED')
    return result.stdout.decode('utf-8').strip()


def prepare():
    assert not STAGING.exists(), 'NEW_PUBLICATION_DIRECTORY_REQUIRED'
    meta = json.loads((OUT / 'publish/manual-release.json').read_bytes())
    archive = meta['archive']['path']
    files = {p.relative_to(OUT / 'package').as_posix(): p.read_bytes()
             for p in (OUT / 'package').rglob('*') if p.is_file()}
    for name in [archive, 'manual-release.json']:
        files[name] = (OUT / 'publish' / name).read_bytes()
    assert sha(files[archive]) == meta['archive']['sha256']
    files['.gitattributes'] = b'* -text\n'
    spec = importlib.util.spec_from_file_location('public_release_checks', ROOT / 'scripts/publish-workshop-onboarding-r5.py')
    check = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(check)
    scan = check.secret_scan(files)
    (OUT / 'public-secret-check.json').write_bytes(encode(scan))
    STAGING.mkdir(parents=True)
    for name, b in files.items():
        p = STAGING / name
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_bytes(b)
    git('init', '--initial-branch=' + BRANCH)
    git('config', 'user.name', git('config', '--get', 'user.name', cwd=ROOT))
    git('config', 'user.email', git('config', '--get', 'user.email', cwd=ROOT))
    git('config', 'core.autocrlf', 'false')
    git('remote', 'add', 'origin', REMOTE)
    git('add', '--', *sorted(files))
    git('commit', '-m', 'release: compact three-half-day manual paired with workshop Skill r10')
    assert set(git('ls-files').splitlines()) == set(files), 'UNEXPECTED_PUBLISHED_FILES'
    for name, b in files.items():
        oid = git('rev-parse', 'HEAD:' + name)
        result = subprocess.run(['git', 'cat-file', 'blob', oid], cwd=STAGING, capture_output=True, check=True)
        assert result.stdout == b, 'GIT_BLOB_CHANGED'
    commit = git('rev-parse', 'HEAD')
    plan = {'ok': True, 'commit': commit, 'branch': BRANCH, 'files': {n: sha(b) for n, b in files.items()},
            'published': False, 'mainBranchModified': False, 'secretCheck': scan}
    (OUT / 'git-plan.json').write_bytes(encode(plan))
    print(json.dumps({'ok': True, 'prepared': True, 'commit': commit, 'files': len(files)}))


def publish():
    plan = json.loads((OUT / 'git-plan.json').read_bytes())
    commit = plan['commit']
    assert git('rev-parse', 'HEAD') == commit and not git('status', '--porcelain'), 'PUBLICATION_STAGING_CHANGED'
    before = git('ls-remote', 'origin', 'refs/heads/' + BRANCH)
    assert not before or before.split()[0] == commit, 'REMOTE_BRANCH_ALREADY_DIFFERS'
    git('push', 'origin', 'HEAD:refs/heads/' + BRANCH)
    assert git('ls-remote', 'origin', 'refs/heads/' + BRANCH).split()[0] == commit

    def verify(item):
        name, expected = item
        url = PUBLIC + '/raw/' + commit + '/' + name
        request = urllib.request.Request(url, headers={'User-Agent': 'ksf-release-verifier/1'})
        with urllib.request.urlopen(request, timeout=40) as response:
            raw = response.read()
            assert response.status == 200 and sha(raw) == expected, 'ONLINE_FILE_HASH_MISMATCH'
            return {'path': name, 'url': url, 'status': response.status, 'bytes': len(raw),
                    'sha256': sha(raw), 'matchesLocal': True}

    with ThreadPoolExecutor(max_workers=4) as pool:
        resources = list(pool.map(verify, plan['files'].items()))
    receipt = {'ok': True, 'releaseId': 'ksf-workshop-manual-20261110-r1', 'publicResourcesPublished': True,
               'repository': PUBLIC, 'branch': BRANCH, 'commit': commit,
               'deploymentGuide': PUBLIC + '/raw/' + commit + '/DEPLOY.md',
               'distributionUrl': PUBLIC + '/raw/' + commit + '/manual-release.json',
               'resources': resources, 'allPublicFilesReadBack': True,
               'productionWebsiteDeployed': False, 'productionAgentSkillUpdated': False,
               'employeeWorkshopTrialCompleted': False, 'businessDataModified': False,
               'mainBranchModified': False}
    (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('command', choices=['prepare', 'publish'])
    args = parser.parse_args()
    prepare() if args.command == 'prepare' else publish()