kangshifu-board/scripts/publish-workshop-baseline-upgrade.py

"""Publish the strict known-baseline platform upgrade; target Skill remains the already-published r10."""
import argparse
import hashlib
import io
import json
from pathlib import Path
import re
import shutil
import subprocess
import urllib.request
import zipfile

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


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


def git(*args, cwd=STAGING, data=None):
    result = subprocess.run(['git', *args], cwd=cwd, input=data, capture_output=True, timeout=90)
    need(result.returncode == 0, 'GIT_OPERATION_FAILED')
    return result.stdout.decode('utf-8').strip()


def secret_scan(files):
    # Compare actual private values in memory only; never include them or their fingerprints in output.
    private = Path('F:/文档/正式环境连接配置.txt').read_text('utf-8-sig')
    match = re.search(r'^\|\s*masterKey\s*\|\s*`?([^|`\s]+)', private, re.M)
    need(match is not None, 'LOCAL_SECRET_CHECK_CONFIG_MISSING')
    secrets = [match.group(1).encode('utf-8')]
    env = Path('F:/文档/康师傅数字人受限连接.env').read_text('utf-8-sig')
    for key, value in re.findall(r'^([A-Z0-9_]+)=(.+)$', env, re.M):
        if re.search('TOKEN|SECRET|PASSWORD|MASTER_KEY', key):
            value = value.strip().strip('\"\'')
            if len(value) >= 16:
                secrets.append(value.encode('utf-8'))
    checked = 0
    for name, raw in files.items():
        bodies = [(name, raw)]
        if name.endswith('.zip'):
            with zipfile.ZipFile(io.BytesIO(raw)) as z:
                bodies.extend((name + ':' + n, z.read(n)) for n in z.namelist())
        for label, body in bodies:
            need(not any(value in body for value in secrets), 'SECRET_VALUE_DETECTED')
            need(b'-----BEGIN PRIVATE KEY-----' not in body and b'-----BEGIN RSA PRIVATE KEY-----' not in body,
                 'PRIVATE_KEY_DETECTED')
            checked += 1
    return {'ok': True, 'filesAndArchiveMembersChecked': checked, 'credentialValuesIncluded': False}


def prepare():
    need(not STAGING.exists(), 'NEW_PUBLICATION_DIRECTORY_REQUIRED')
    distribution = json.loads((OUT / 'publish/distribution.json').read_bytes())
    archive = distribution['archive']['path']
    names = ['README.md', 'distribution.json', archive]
    files = {n: (OUT / 'publish' / n).read_bytes() for n in names}
    need(sha(files[archive]) == distribution['archive']['sha256'], 'ARCHIVE_CHANGED')
    for p in (OUT / 'package').rglob('*'):
        if p.is_file():
            files['source/' + p.relative_to(OUT / 'package').as_posix()] = p.read_bytes()
    files['.gitattributes'] = b'* -text\n'
    scan = secret_scan(files)
    (OUT / 'public-secret-check.json').write_bytes(encode(scan))
    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)
    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: strict r6 baseline and missing-document support for unchanged workshop Skill r10')
    need(set(git('ls-files').splitlines()) == set(files), 'UNEXPECTED_PUBLISHED_FILES')
    for name, raw in files.items():
        oid = git('rev-parse', 'HEAD:' + name)
        result = subprocess.run(['git', 'cat-file', 'blob', oid], cwd=STAGING, capture_output=True, check=True)
        need(result.stdout == raw, 'GIT_BLOB_CHANGED')
    commit = git('rev-parse', 'HEAD')
    plan = {'ok': True, 'commit': commit, 'branch': BRANCH, 'files': {n: sha(raw) for n, raw 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']
    need(git('rev-parse', 'HEAD') == commit and not git('status', '--porcelain'), 'PUBLICATION_STAGING_CHANGED')
    remote = git('ls-remote', 'origin', 'refs/heads/' + BRANCH)
    need(not remote or remote.split()[0] == commit, 'REMOTE_BRANCH_ALREADY_DIFFERS')
    git('push', 'origin', 'HEAD:refs/heads/' + BRANCH)
    need(git('ls-remote', 'origin', 'refs/heads/' + BRANCH).split()[0] == commit, 'PUSH_NOT_CONFIRMED')
    checks = []
    # Anonymous reads, without platform credentials, verify all public artifacts and sources.
    for name, expected in plan['files'].items():
        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()
            need(response.status == 200 and sha(raw) == expected, 'ONLINE_FILE_HASH_MISMATCH')
            checks.append({'path': name, 'url': url, 'status': response.status, 'bytes': len(raw),
                           'sha256': sha(raw), 'matchesLocal': True})
    guide = PUBLIC + '/raw/' + commit + '/README.md'
    prompt = ('请完成原工作坊Agent的r10升级。固定执行单：\n' + guide + '\n'
              '核同目录distribution.json及ZIP后，按执行单完成plan/apply/verify。'
              '只接受已知完整历史基线；若不匹配，停止并回传inventory的公开文件指纹，不绕过保护。'
              '保留身份、授权、连接及7个运行文件，不安装VOC、不写业务记录、不重跑迁移。\n')
    (OUT / '执行正文.txt').write_text(prompt, encoding='utf-8')
    receipt = {'ok': True, 'published': True, 'releaseId': 'ksf-workshop-r10-baseline-upgrade-20261110-r1',
               'repository': PUBLIC, 'branch': BRANCH, 'commit': commit, 'guideUrl': guide,
               'resources': checks, 'allPublicFilesReadBack': True, 'mainBranchModified': False,
               'workshopSkillVersion': 'r10', 'supportedUpgradeFrom': ['r6', 'r7', 'r8', 'r9'],
               'productionDatabaseCutoverPreviouslyConfirmed': True, 'productionMigrationExecutedByPublisher': False,
               'targetHostInstallationVerified': False, 'businessDataModified': False}
    (OUT / 'online-release-receipt.json').write_bytes(encode(receipt))
    print(json.dumps({'ok': True, 'published': True, 'commit': commit, 'guideUrl': guide,
                      'publicFilesVerified': len(checks), 'productionMigrationExecuted': False}))


if __name__ == '__main__':
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument('command', choices=('prepare', 'publish'))
    args = p.parse_args()
    prepare() if args.command == 'prepare' else publish()