kangshifu-board/scripts/release-unified-runtime-hotfix.py

"""Seal and publish the runtime repair separately from migration r3 and workshop r9."""
import argparse
import ast
import difflib
import hashlib
import importlib.util
import io
import json
from pathlib import Path
import re
import subprocess
import urllib.request
import zipfile

ROOT=Path(__file__).resolve().parents[1]
NAME='ksf-unified-runtime-hotfix-20261109-r1'
OUT=ROOT/'docs'/NAME
TEST=ROOT/'work/unified-runtime-hotfix-20261109'
STAGING=ROOT/'work'/('publish-'+NAME)
BASE=ROOT/'docs/workshop-onboarding-20261109-r4/publish/operations/ksf-unified-database-migration-20261109-r3.zip'
BASE_SHA='6bf273509ea6f016a5b3e6c38d077bbce158e677bd701e189c80e016788d93c2'
BRANCH='codex/unified-runtime-hotfix-20261109-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 git(*args,cwd=STAGING):
    p=subprocess.run(['git',*args],cwd=cwd,capture_output=True,timeout=120)
    need(p.returncode==0,'GIT_OPERATION_FAILED')
    return p.stdout.decode('utf-8').strip()


def prepare():
    need(not OUT.exists() and not STAGING.exists(),'NEW_RELEASE_DIRECTORY_REQUIRED')
    need(sha(BASE.read_bytes())==BASE_SHA,'SEALED_R3_CHANGED')
    validation=json.loads((TEST/'validation.json').read_bytes())
    need(validation['ok'] and validation['realPostgreSQL'],'LOCAL_PG_VALIDATION_REQUIRED')
    for name in ('targeted-regression','real-pg'):
        raw=(TEST/(name+'.log')).read_text('utf-8')
        need(re.search(r'\nOK\s*$',raw),'LOCAL_TEST_FAILURE')
    real=json.loads((TEST/'real-final2/real-archive-validation.json').read_bytes())
    receiver=json.loads((TEST/'real-final2/real-receiver-validation.json').read_bytes())
    need(real['ok'] and receiver['ok'],'REAL_ARCHIVE_VALIDATION_REQUIRED')
    with zipfile.ZipFile(BASE) as z:
        baseline={n:z.read(n) for n in z.namelist()}
    payload={n:raw for n,raw in baseline.items() if n.startswith(('server/','scripts/','components/research-runtime/'))
             and n!='scripts/migrate-unified-database.py'}
    patches=[];changes=[]
    for module in ('retrieval.py','update_library.py'):
        n='server/voc-library/'+module;raw=(ROOT/n).read_bytes()
        need(raw!=baseline[n],'EMPTY_REPAIR')
        payload[n]=raw
        payload['components/research-runtime/scripts/'+module]=raw
        changes.append({'path':n,'sealedR3Sha256':sha(baseline[n]),'afterSha256':sha(raw)})
        patches.extend(difflib.unified_diff(baseline[n].decode('utf-8').splitlines(True),raw.decode('utf-8').splitlines(True),
                                          fromfile='sealed-r3/'+n,tofile=NAME+'/'+n))
    component='components/research-runtime/'
    payload[component+'FILES.sha256']=''.join(sha(raw)+'  '+n[len(component):]+'\n'
        for n,raw in sorted(payload.items()) if n.startswith(component) and n!=component+'FILES.sha256').encode()
    for n,raw in payload.items():
        if n.endswith('.py'):ast.parse(raw.decode('utf-8-sig'),filename=n,feature_version=(3,9))
    evidence={'ok':True,'targetedRegressionTests':66,'realPostgreSQLTests':15,
              'genuineArchiveValidation':real,'genuineReceiverValidation':receiver,
              'productionWrites':0,'productionAcceptancePending':True}
    payload['validation.json']=encode(evidence)
    for n in ('targeted-regression.log','real-pg.log'):payload['evidence/'+n]=(TEST/n).read_bytes()
    payload['patches/runtime.diff']=''.join(patches).encode('utf-8')
    readme=(ROOT/'docs/unified-runtime-hotfix-instructions-20261109.md').read_bytes()
    payload['README.md']=readme
    release={'release':NAME,'sealedMigrationBase':BASE_SHA,'database':'FKieRbIoGt_master','role':'FKieRbIoGt_admin',
             'appId':'FKieRbIoGt_main','changedServerModules':changes,'serverModulesOtherThanTheseUnchanged':True,
             'mainSkillIncluded':False,'manualIncluded':False,'migrationApplyToolIncluded':False,
             'credentialsIncluded':False,'rawResearchDataIncluded':False,'productionDeployed':False}
    payload['release.json']=encode(release)
    payload['FILES.sha256']=''.join(sha(raw)+'  '+n+'\n' for n,raw in sorted(payload.items())).encode()
    stream=io.BytesIO()
    with zipfile.ZipFile(stream,'w',zipfile.ZIP_DEFLATED) as z:
        for n,raw in sorted(payload.items()):z.writestr(n,raw)
    archive=stream.getvalue()
    with zipfile.ZipFile(io.BytesIO(archive)) as z:
        need(z.testzip() is None,'ARCHIVE_CORRUPT')
        for line in z.read('FILES.sha256').decode().splitlines():
            h,n=line.split('  ',1);need(sha(z.read(n))==h,'ARCHIVE_HASH_MISMATCH')
    manifest={'release':NAME,'archive':NAME+'.zip','bytes':len(archive),'sha256':sha(archive),
              'filesVerified':len(payload)-1,'validation':{'targetedRegressionTests':66,'realPostgreSQLTests':15,
              'realArchiveRebuildPassed':True,'realReceiverPreflightPassed':True,'productionWrites':0},
              'productionDeploymentPending':True,'mainSkillAndManualDeferred':True}
    files={'README.md':readme,'release.json':encode(manifest),NAME+'.zip':archive,
           'validation.json':encode(evidence),'patches/runtime.diff':payload['patches/runtime.diff'],'.gitattributes':b'* -text\n'}
    spec=importlib.util.spec_from_file_location('publication_helpers',ROOT/'scripts/publish-workshop-onboarding-r4.py')
    helpers=importlib.util.module_from_spec(spec);spec.loader.exec_module(helpers)
    scan=helpers.secret_scan(files)
    OUT.mkdir(parents=True);STAGING.mkdir(parents=True)
    for n,raw in files.items():
        for base in (STAGING,OUT/'publish'):
            p=base/n;p.parent.mkdir(parents=True,exist_ok=True);p.write_bytes(raw)
    git('init','--initial-branch='+BRANCH)
    for k in ('user.name','user.email'):git('config',k,git('config','--get',k,cwd=ROOT))
    git('config','core.autocrlf','false');git('remote','add','origin',REMOTE)
    git('add','--',*sorted(files));git('commit','-m','fix: valid default facets and source locator reuse on archive rebuild')
    need(set(git('ls-files').splitlines())==set(files),'UNEXPECTED_RELEASE_FILES')
    for n,raw in files.items():
        body=subprocess.run(['git','show','HEAD:'+n],cwd=STAGING,capture_output=True,check=True).stdout
        need(body==raw,'PUBLICATION_BYTES_CHANGED')
    plan={'commit':git('rev-parse','HEAD'),'branch':BRANCH,'files':{n:sha(raw) for n,raw in files.items()},
          'secretScan':scan,'release':manifest,'published':False}
    (OUT/'publication-plan.json').write_bytes(encode(plan))
    print(json.dumps({'ok':True,'prepared':True,'commit':plan['commit'],'archive':manifest}))


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'),'RELEASE_STAGING_CHANGED')
    prior=git('ls-remote','origin','refs/heads/'+BRANCH)
    need(not prior or prior.split()[0]==commit,'REMOTE_BRANCH_DIFFERS')
    git('push','origin','HEAD:refs/heads/'+BRANCH)
    need(git('ls-remote','origin','refs/heads/'+BRANCH).split()[0]==commit,'PUSH_UNCONFIRMED')
    checks=[]
    for n,h in plan['files'].items():
        url=PUBLIC+'/raw/'+commit+'/'+n
        with urllib.request.urlopen(urllib.request.Request(url,headers={'User-Agent':'ksf-runtime-release-verifier/1'}),timeout=50) as response:
            raw=response.read();need(response.status==200 and sha(raw)==h,'PUBLIC_READBACK_MISMATCH')
            checks.append({'path':n,'url':url,'httpStatus':response.status,'bytes':len(raw),'sha256':h})
    receipt={'ok':True,'published':True,'commit':commit,'branch':BRANCH,'mainBranchModified':False,
             'resources':checks,'release':plan['release'],'productionDeployed':False}
    (OUT/'online-release-receipt.json').write_bytes(encode(receipt))
    print(json.dumps({'ok':True,'commit':commit,'publicFilesVerified':len(checks),'readmeUrl':PUBLIC+'/raw/'+commit+'/README.md'}))


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