kangshifu-board/scripts/package-product-images-layout.py

"""Freeze the integrated product UI against the current observed public baseline."""
import argparse
import concurrent.futures
import difflib
import hashlib
import json
from pathlib import Path
import re
import subprocess
import urllib.error
import urllib.request
import zipfile

ROOT = Path(__file__).resolve().parents[1]
WORK = ROOT / 'work/product-images-layout-20261106-r2'
PACKAGE = WORK / 'package'
DOCS = ROOT / 'docs/industry-product-images-20261106/release-r2'
OUT = ROOT / 'artifacts/商品图片与详情排版-20261106-r2'
RELEASE = 'ksf-product-images-layout-20261106-r2'
FILES = [
    'src/core/library-comparison.js', 'src/services/library-images.js', 'src/pages/library-product-image.js',
    'src/core/library.js', 'src/services/library-service.js', 'src/services/library-comparison.js',
    'src/pages/library-product-compare.js', 'src/pages/library-styles.js',
    'src/pages/library-product-detail.js', 'src/pages/view-library.js',
]


def sha(raw):
    return hashlib.sha256(raw).hexdigest()


def norm(raw):
    return sha(raw.replace(b'\r\n', b'\n'))


def save(name, data):
    p = PACKAGE / name
    p.parent.mkdir(parents=True, exist_ok=True)
    p.write_bytes(data.encode() if isinstance(data, str) else data)


def js(name, data):
    save(name, json.dumps(data, ensure_ascii=False, indent=2) + '\n')


def prepare(refresh=False):
    assert not PACKAGE.exists() or refresh, 'PACKAGE_ALREADY_EXISTS'
    assert not (OUT / (RELEASE + '.zip')).exists(), 'IMMUTABLE_ARCHIVE_ALREADY_EXISTS'
    baselines = {r['path']: r for r in json.loads((WORK / 'production-baseline.json').read_text(encoding='utf-8'))}
    assert baselines['src/pages/library-product-detail.js']['sha256'] == 'f77135b2dff426a4ac30d668271831d7e20816e5d0ff1b7ead41219e4e2051ba'
    old_manifest = json.loads((ROOT / 'docs/industry-product-images-20261106/frontend-release.json').read_text(encoding='utf-8'))
    for row in old_manifest['files']:
        assert baselines[row['path']]['sha256'] == row['observedProductionSha256'], 'PUBLIC_BASELINE_CHANGED:' + row['path']
    entries = []
    diffs = []
    for name in FILES:
        before = (WORK / 'production-before' / name).read_bytes()
        observed = baselines[name]
        assert sha(before) == observed['sha256']
        raw = (ROOT / name).read_bytes()
        save('payload/' + name, raw)
        subprocess.run(['node', '--check', str(PACKAGE / 'payload' / name)], check=True, capture_output=True)
        entries.append({'path': name, 'sha256': sha(raw), 'normalizedSha256': norm(raw),
                        'priorNormalizedSha256': [] if observed['absent'] else [norm(before)],
                        'observedProductionSha256': sha(before), 'allowMissing': observed['absent']})
        diffs.extend(difflib.unified_diff([] if observed['absent'] else before.decode('utf-8').splitlines(True),
                     raw.decode('utf-8').splitlines(True), fromfile='production/' + name, tofile='payload/' + name))
    detail = (PACKAGE / 'payload/src/pages/library-product-detail.js').read_text(encoding='utf-8')
    public_detail = (WORK / 'production-before/src/pages/library-product-detail.js').read_text(encoding='utf-8')
    getter = lambda text: text[text.index('  get reviewQuality() {'):text.index('  _count(entity) {')]
    assert getter(detail) == getter(public_detail)
    for part in ["if (request?.entity === 'reviews') request.quality = this.reviewQuality;", '.value=${this.reviewQuality}',
                 '已收录评价仍有信息待确认，先展示原文；这些记录尚未计入分析概况。']:
        assert part in detail and part in public_detail
    assert '<h3>商品信息</h3>' not in detail
    manifest = {'version': 'ksf.product-images-layout.publish.v1', 'release': RELEASE,
                'origin': 'https://kangshifu.fmode.cn', 'homepageSha256': baselines['index.html']['sha256'],
                'files': entries, 'reviewDefaultDisplayHotfixPreserved': True,
                'databaseImagesAlreadyStored': 4674, 'productionInstalled': False}
    js('deploy/product-ui-files.json', manifest)
    publisher = (ROOT / 'scripts/publish-opportunity-ui.mjs').read_text(encoding='utf-8')
    start = publisher.index('export const FILES=Object.freeze([')
    end = publisher.index(']);', start) + 3
    publisher = publisher[:start] + 'export const FILES=Object.freeze(' + json.dumps(FILES) + ');' + publisher[end:]
    publisher = publisher.replace('ksf.opportunity-ui.publish.v1', manifest['version']).replace('ksf.opportunity-ui.20261103.v1', RELEASE)
    publisher = publisher.replace('filesChecked:6', 'filesChecked:FILES.length').replace('ksf-opportunity-ui', 'ksf-product-ui')
    publisher = publisher.replace("sourceDir:root,manifestPath:path.join(root,'deploy/opportunity-ui-files-20261103.json')",
                                  "sourceDir:path.join(root,'payload'),manifestPath:path.join(root,'deploy/product-ui-files.json')")
    save('scripts/publish-product-ui.mjs', publisher)
    subprocess.run(['node', '--check', str(PACKAGE / 'scripts/publish-product-ui.mjs')], check=True, capture_output=True)
    save('evidence/product-ui.patch', ''.join(diffs))
    save('evidence/database-before.json', (WORK / 'database-before.json').read_bytes())
    deps = []
    for name in FILES:
        text = (PACKAGE / 'payload' / name).read_text(encoding='utf-8')
        for match in re.finditer(r'''(?:from|import)\s*\(?\s*['"](\.[^'"]+)['"]''', text):
            resolved = (Path(name).parent / match[1]).as_posix()
            absolute = (ROOT / resolved).resolve()
            assert absolute.is_relative_to(ROOT) and absolute.is_file()
            deps.append(absolute.relative_to(ROOT).as_posix())
    js('RELEASE.json', {'release': RELEASE, 'staticFiles': len(FILES),
                       'updatedExistingFiles': sum(not e['allowMissing'] for e in entries),
                       'newFiles': sum(e['allowMissing'] for e in entries),
                       'reviewDefaultDisplayHotfixPreserved': True, 'relativeImports': sorted(set(deps)),
                       'databaseIncluded': False, 'credentialsIncluded': False, 'productionInstalled': False})
    DOCS.mkdir(parents=True, exist_ok=True)
    (DOCS / 'frontend-release.json').write_bytes((PACKAGE / 'deploy/product-ui-files.json').read_bytes())
    print(json.dumps({'ok': True, 'staticFiles': len(FILES), 'newFiles': sum(e['allowMissing'] for e in entries),
                      'reviewDefaultDisplayHotfixPreserved': True, 'productionWrites': 0}, ensure_ascii=False))


def check_baselines():
    baselines = json.loads((WORK / 'production-baseline.json').read_text(encoding='utf-8'))
    def check(row):
        req = urllib.request.Request('https://kangshifu.fmode.cn/' + row['path'], headers={'Cache-Control': 'no-cache'})
        try:
            with urllib.request.urlopen(req, timeout=30) as r:
                actual, mime = r.read(), r.headers.get('Content-Type', '')
        except urllib.error.HTTPError as error:
            assert row['status'] == error.code
            actual, mime = error.read(), error.headers.get('Content-Type', '')
        assert sha(actual) == row['sha256'] and mime == row['contentType'], 'PUBLIC_BASELINE_CHANGED:' + row['path']
        return {'path': row['path'], 'baselineMatches': True}
    rows = list(concurrent.futures.ThreadPoolExecutor(max_workers=4).map(check, baselines))
    return {'ok': True, 'files': rows, 'productionWrites': 0}


def package():
    for name in ['browser-verification.json', 'publisher-verification.json', 'routine-verification.json']:
        raw = (WORK / name).read_bytes()
        assert json.loads(raw)['ok'], 'VERIFICATION_FAILED:' + name
        save('evidence/' + name, raw)
        (DOCS / name).write_bytes(raw)
    save('evidence/module-graph-local.json', (WORK / 'module-graph-local.json').read_bytes())
    assert json.loads((WORK / 'browser-verification.json').read_text(encoding='utf-8'))['productionInstalled'] is False
    js('evidence/public-baselines-final.json', check_baselines())
    save('交接说明.md', (DOCS / 'README.md').read_bytes())
    for p in sorted((WORK / 'screenshots').glob('*.png')):
        save('evidence/screenshots/' + p.name, p.read_bytes())
        dest = DOCS / 'screenshots' / p.name
        dest.parent.mkdir(parents=True, exist_ok=True)
        dest.write_bytes(p.read_bytes())
    for name in ['package-product-images-layout.py', 'verify-product-ui-release.py']:
        save('verification/' + name, (ROOT / 'scripts' / name).read_bytes())
    members = sorted(p for p in PACKAGE.rglob('*') if p.is_file() and p.name != 'FILES.sha256')
    assert not any(p.suffix.lower() in {'.sqlite', '.db', '.key', '.pem', '.pyc'} for p in members)
    save('FILES.sha256', ''.join(sha(p.read_bytes()) + '  ' + p.relative_to(PACKAGE).as_posix() + '\n' for p in members))
    OUT.mkdir(parents=True, exist_ok=True)
    archive = OUT / (RELEASE + '.zip')
    with zipfile.ZipFile(archive, 'x', zipfile.ZIP_DEFLATED, compresslevel=9) as z:
        for p in [*members, PACKAGE / 'FILES.sha256']:
            z.write(p, p.relative_to(PACKAGE).as_posix())
    with zipfile.ZipFile(archive) as z:
        assert z.testzip() is None
        for line in z.read('FILES.sha256').decode().splitlines():
            expected, name = line.split('  ', 1)
            assert sha(z.read(name)) == expected
    receipt = {'ok': True, 'path': str(archive), 'sha256': sha(archive.read_bytes()), 'bytes': archive.stat().st_size,
               'filesVerified': len(members), 'staticFiles': len(FILES), 'productionInstalled': False,
               'productionWrites': 0, 'supersedes': 'ksf-product-images-20261106-r1.zip'}
    (DOCS / 'delivery-receipt.json').write_text(json.dumps(receipt, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
    print(json.dumps(receipt, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--package', action='store_true')
    parser.add_argument('--refresh', action='store_true', help='Refresh unarchived local staging after a source correction')
    args = parser.parse_args()
    assert not (args.package and args.refresh)
    package() if args.package else prepare(args.refresh)