feat: new channel
This commit is contained in:
hashbro
@@ -1,123 +1,345 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Pack a per-channel weifile zip from the already-built staged tree.
|
||||
"""Pack a per-channel exclusive tree under public/channel/{ver}/.
|
||||
|
||||
Does not rebuild natives or details. Requires `tools/build.py --apply` first.
|
||||
Requires `tools/build.py --apply` first (shared staged weifile + public/details).
|
||||
|
||||
1. Copy staged weifile (storage/app/channel-builder-new/out/weifile)
|
||||
2. Replace CACACACA in index.js with the 8-char channel id
|
||||
3. Zip as {artifact-root}/channel-source-new/{CHANNELID}.zip
|
||||
1. Copy public/details + staged weifile
|
||||
2. Patch corepayload `2.2.66` -> channel ver (same length)
|
||||
3. Patch corepayload `/details/show.html` -> `/c/{ver}/show.htm` (18 bytes; netconfig)
|
||||
4. Rewrite show.html asset URLs to /channel/{ver}/details/...
|
||||
5. Patch secondary `/details/show.html` -> `/c/{ver}/show.htm` (18 bytes)
|
||||
6. Strip iptj beacon from index.js; inject t.js into weifile.html
|
||||
7. Write to {artifact-root}/channel/{ver}/
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
import shutil
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from zipfile import ZIP_DEFLATED, ZipFile
|
||||
|
||||
import build as xxbb_build
|
||||
from _details_pack import extract_member, make_passworded_7z
|
||||
from _secondary_pack import decrypt_secondary_minjs, encrypt_secondary_minjs
|
||||
|
||||
RESULT_MARKER = xxbb_build.RESULT_MARKER
|
||||
INDEX_CHANNEL_PLACEHOLDER = xxbb_build.INDEX_CHANNEL_PLACEHOLDER
|
||||
WEIFILE_ROOT = xxbb_build.WEIFILE_ROOT
|
||||
DETAILS_ROOT = xxbb_build.DETAILS_ROOT
|
||||
LAB_SEEDS_NAME = xxbb_build.LAB_SEEDS_NAME
|
||||
CORE_WIRE_NAME = xxbb_build.CORE_WIRE_NAME
|
||||
CORE_MEMBER_NAME = xxbb_build.CORE_MEMBER_NAME
|
||||
SHOW_WIRE_NAME = xxbb_build.SHOW_WIRE_NAME
|
||||
SHOW_MEMBER_NAME = xxbb_build.SHOW_MEMBER_NAME
|
||||
ORIGINAL_VER = "2.2.66"
|
||||
SHOW_PATH_OLD = b"/details/show.html"
|
||||
# equal length (18): /c/X.Y.ZZ/show.htm
|
||||
SHOW_PATH_TMPL = "/c/{ver}/show.htm"
|
||||
CHANNEL_VER_RE = re.compile(r"^[0-9A-Za-z]\.[0-9A-Za-z]\.[0-9A-Za-z]{2}$")
|
||||
IPTJ_TAIL_RE = re.compile(
|
||||
r",function\(\)\{\[67,72,77,75,54,73,71,48,56,70,52,50,52,57,54,67,50,50\]"
|
||||
r".*?\},2e3\)\}\(\);",
|
||||
re.S,
|
||||
)
|
||||
|
||||
|
||||
def normalize_channel_ver(value: str) -> str:
|
||||
ver = (value or "").strip().upper()
|
||||
if not CHANNEL_VER_RE.fullmatch(ver):
|
||||
raise SystemExit(
|
||||
f"--channel-ver must match X.Y.ZZ (6 chars, alnum+dots) like {ORIGINAL_VER!r} (got {value!r})"
|
||||
)
|
||||
return ver
|
||||
|
||||
|
||||
def staged_weifile_dir(state_root: Path) -> Path:
|
||||
return state_root / "out" / WEIFILE_ROOT
|
||||
|
||||
|
||||
def patch_channel_code(text: str, channel_code: str) -> str:
|
||||
if INDEX_CHANNEL_PLACEHOLDER not in text:
|
||||
def sha256_hex(data: bytes) -> str:
|
||||
return hashlib.sha256(data).hexdigest()
|
||||
|
||||
|
||||
def patch_ver_in_bytes(data: bytes, ver: str, *, label: str) -> bytes:
|
||||
old = ORIGINAL_VER.encode("ascii")
|
||||
new = ver.encode("ascii")
|
||||
if len(old) != len(new):
|
||||
raise SystemExit(f"{label}: ver length mismatch {len(old)} vs {len(new)}")
|
||||
count = data.count(old)
|
||||
if count < 1:
|
||||
raise SystemExit(f"{label}: missing {ORIGINAL_VER!r} to patch")
|
||||
return data.replace(old, new)
|
||||
|
||||
|
||||
def patch_show_path_in_bytes(data: bytes, ver: str, *, label: str) -> bytes:
|
||||
"""Equal-length rewrite of netconfig path /details/show.html -> /c/{ver}/show.htm."""
|
||||
new_path = SHOW_PATH_TMPL.format(ver=ver).encode("ascii")
|
||||
if len(new_path) != len(SHOW_PATH_OLD):
|
||||
raise SystemExit(f"{label}: show path length {len(new_path)} != {len(SHOW_PATH_OLD)}")
|
||||
count = data.count(SHOW_PATH_OLD)
|
||||
if count < 1:
|
||||
raise SystemExit(f"{label}: missing {SHOW_PATH_OLD!r}")
|
||||
return data.replace(SHOW_PATH_OLD, new_path)
|
||||
|
||||
|
||||
def patch_corepayload_wire(wire: bytes, ver: str) -> tuple[bytes, dict]:
|
||||
member, plain = extract_member(wire)
|
||||
if member != CORE_MEMBER_NAME:
|
||||
raise SystemExit(f"unexpected core member {member!r}")
|
||||
patched = patch_ver_in_bytes(plain, ver, label=CORE_MEMBER_NAME)
|
||||
show_hits = patched.count(SHOW_PATH_OLD)
|
||||
patched = patch_show_path_in_bytes(patched, ver, label=CORE_MEMBER_NAME)
|
||||
digest = sha256_hex(patched)
|
||||
new_wire = make_passworded_7z(CORE_MEMBER_NAME, patched)
|
||||
return new_wire, {
|
||||
"sha256": digest,
|
||||
"size": len(patched),
|
||||
"ver_hits": patched.count(ver.encode()),
|
||||
"show_path_hits": show_hits,
|
||||
}
|
||||
|
||||
|
||||
def rewrite_show_urls(config_bytes: bytes, *, ver: str, core_sha256: str, core_size: int) -> bytes:
|
||||
doc = json.loads(config_bytes.decode("utf-8"))
|
||||
if not isinstance(doc, dict) or not isinstance(doc.get("core"), dict):
|
||||
raise SystemExit("show data.bin: missing core object")
|
||||
prefix = f"/channel/{ver}/details/"
|
||||
|
||||
def rewrite_url(url: str) -> str:
|
||||
# http://host/details/foo.js -> http://host/channel/{ver}/details/foo.js
|
||||
return re.sub(r"(/details/)", f"/channel/{ver}/details/", url, count=1)
|
||||
|
||||
core = doc["core"]
|
||||
if isinstance(core.get("url"), str):
|
||||
core["url"] = rewrite_url(core["url"])
|
||||
core["sha256"] = core_sha256
|
||||
core["size"] = core_size
|
||||
for entry in doc.get("entries") or []:
|
||||
if isinstance(entry, dict) and isinstance(entry.get("url"), str):
|
||||
entry["url"] = rewrite_url(entry["url"])
|
||||
_ = prefix # documented intent
|
||||
return json.dumps(doc, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
|
||||
|
||||
|
||||
def patch_show_wire(wire: bytes, *, ver: str, core_sha256: str, core_size: int) -> bytes:
|
||||
member, plain = extract_member(wire)
|
||||
if member != SHOW_MEMBER_NAME:
|
||||
raise SystemExit(f"unexpected show member {member!r}")
|
||||
updated = rewrite_show_urls(plain, ver=ver, core_sha256=core_sha256, core_size=core_size)
|
||||
return make_passworded_7z(SHOW_MEMBER_NAME, updated)
|
||||
|
||||
|
||||
def patch_secondary_show_path(dylib: bytes, ver: str, *, label: str) -> bytes:
|
||||
# One occurrence per secondary slice is enough; replace first only.
|
||||
new_path = SHOW_PATH_TMPL.format(ver=ver).encode("ascii")
|
||||
if len(new_path) != len(SHOW_PATH_OLD):
|
||||
raise SystemExit(f"{label}: show path length {len(new_path)} != {len(SHOW_PATH_OLD)}")
|
||||
if SHOW_PATH_OLD not in dylib:
|
||||
raise SystemExit(f"{label}: missing {SHOW_PATH_OLD!r}")
|
||||
return dylib.replace(SHOW_PATH_OLD, new_path, 1)
|
||||
|
||||
|
||||
def patch_weifile_secondaries(weifile_dir: Path, ver: str) -> int:
|
||||
keys = xxbb_build.load_keys()
|
||||
patched = 0
|
||||
for stem, info in keys["stems"].items():
|
||||
# build.py stages patched secondaries as {stem}.min.js
|
||||
path = weifile_dir / f"{stem}.min.js"
|
||||
if not path.is_file():
|
||||
path = weifile_dir / f"{stem}.js"
|
||||
if not path.is_file():
|
||||
continue
|
||||
key = bytes.fromhex(info["key"])
|
||||
try:
|
||||
dylib = decrypt_secondary_minjs(path.read_bytes(), key)
|
||||
except Exception as exc:
|
||||
raise SystemExit(f"{path.name}: decrypt failed: {exc}") from exc
|
||||
if SHOW_PATH_OLD not in dylib:
|
||||
continue
|
||||
dylib = patch_secondary_show_path(dylib, ver, label=stem)
|
||||
path.write_bytes(encrypt_secondary_minjs(dylib, key))
|
||||
patched += 1
|
||||
if patched < 1:
|
||||
raise SystemExit(
|
||||
f"index.js: missing {INDEX_CHANNEL_PLACEHOLDER} placeholder; "
|
||||
"re-run tools/build.py --apply"
|
||||
"no secondary packs contained /details/show.html "
|
||||
f"(looked under {weifile_dir} for *.min.js)"
|
||||
)
|
||||
return text.replace(INDEX_CHANNEL_PLACEHOLDER, channel_code, 1)
|
||||
return patched
|
||||
|
||||
|
||||
def write_zip(weifile_dir: Path, zip_path: Path) -> Path:
|
||||
zip_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
if zip_path.exists():
|
||||
zip_path.unlink()
|
||||
skip = {"_bak", "__pycache__", ".DS_Store", "decoded", "mm", "stages"}
|
||||
with ZipFile(zip_path, "w", compression=ZIP_DEFLATED) as archive:
|
||||
for path in sorted(weifile_dir.rglob("*")):
|
||||
if not path.is_file():
|
||||
continue
|
||||
if path.name in skip or path.suffix == ".pyc":
|
||||
continue
|
||||
if any(part in skip for part in path.relative_to(weifile_dir).parts):
|
||||
continue
|
||||
rel = Path(WEIFILE_ROOT) / path.relative_to(weifile_dir)
|
||||
archive.write(path, rel.as_posix())
|
||||
return zip_path
|
||||
def strip_iptj_beacon(index_js: str) -> str:
|
||||
if not IPTJ_TAIL_RE.search(index_js):
|
||||
# already stripped or unexpected layout — allow missing during rebuilds
|
||||
if "channelCode" in index_js or "__iptj_sid" in index_js:
|
||||
raise SystemExit("index.js: iptj beacon present but pattern mismatch")
|
||||
return index_js
|
||||
return IPTJ_TAIL_RE.sub(";", index_js, count=1)
|
||||
|
||||
|
||||
def inject_tjs(weifile_html: str) -> str:
|
||||
if "t.js" in weifile_html:
|
||||
return weifile_html
|
||||
snippet = '<script src="/t.js" defer></script>\n'
|
||||
if "</head>" in weifile_html:
|
||||
return weifile_html.replace("</head>", snippet + "</head>", 1)
|
||||
return snippet + weifile_html
|
||||
|
||||
|
||||
LANDING_TEMPLATES = ("test", "blank")
|
||||
DEFAULT_LANDING_TEMPLATE = "test"
|
||||
LANDING_TEMPLATE_ROOT = xxbb_build.BUILDER_ROOT / "source" / "weifile" / "templates"
|
||||
|
||||
|
||||
def normalize_landing_template(value: str | None) -> str:
|
||||
template = (value or DEFAULT_LANDING_TEMPLATE).strip().lower()
|
||||
if template not in LANDING_TEMPLATES:
|
||||
raise SystemExit(
|
||||
f"--landing-template must be one of: {', '.join(LANDING_TEMPLATES)} (got {value!r})"
|
||||
)
|
||||
return template
|
||||
|
||||
|
||||
def apply_landing_template(weifile_dir: Path, template: str) -> Path:
|
||||
template = normalize_landing_template(template)
|
||||
src = LANDING_TEMPLATE_ROOT / f"{template}.html"
|
||||
if not src.is_file():
|
||||
raise SystemExit(f"missing landing template: {src}")
|
||||
dest = weifile_dir / "weifile.html"
|
||||
dest.write_text(inject_tjs(src.read_text(encoding="utf-8")), encoding="utf-8")
|
||||
return dest
|
||||
|
||||
|
||||
def copy_details_tree(src: Path, dest: Path) -> None:
|
||||
if dest.exists():
|
||||
shutil.rmtree(dest)
|
||||
skip = {"_bak", "__pycache__", ".DS_Store", "README.md"}
|
||||
shutil.copytree(
|
||||
src,
|
||||
dest,
|
||||
symlinks=False,
|
||||
ignore=lambda _d, names: {n for n in names if n in skip or n.endswith(".pyc")},
|
||||
)
|
||||
|
||||
|
||||
def pack_channel(
|
||||
*,
|
||||
channel_code: str,
|
||||
channel_ver: str,
|
||||
weifile_src: Path,
|
||||
zip_out: Path,
|
||||
details_src: Path,
|
||||
channel_out: Path,
|
||||
landing_template: str = DEFAULT_LANDING_TEMPLATE,
|
||||
) -> dict:
|
||||
code = xxbb_build.normalize_channel_code(channel_code)
|
||||
index_src = weifile_src / "index.js"
|
||||
if not index_src.is_file():
|
||||
raise SystemExit(
|
||||
f"staged weifile missing ({index_src}). "
|
||||
"Run: python tools/build.py --apply"
|
||||
)
|
||||
ver = normalize_channel_ver(channel_ver)
|
||||
landing_template = normalize_landing_template(landing_template)
|
||||
if not (weifile_src / "index.js").is_file():
|
||||
raise SystemExit(f"staged weifile missing ({weifile_src / 'index.js'}). Run build.py --apply")
|
||||
if not (details_src / CORE_WIRE_NAME).is_file():
|
||||
raise SystemExit(f"details missing ({details_src / CORE_WIRE_NAME}). Run build.py --apply")
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
dest = Path(tmp) / WEIFILE_ROOT
|
||||
xxbb_build.copy_tree(weifile_src, dest)
|
||||
index_path = dest / "index.js"
|
||||
root = Path(tmp) / "channel"
|
||||
details_dest = root / DETAILS_ROOT
|
||||
weifile_dest = root / WEIFILE_ROOT
|
||||
copy_details_tree(details_src, details_dest)
|
||||
xxbb_build.copy_tree(weifile_src, weifile_dest)
|
||||
# Drop template sources from the published tree if build copied them.
|
||||
templates_dir = weifile_dest / "templates"
|
||||
if templates_dir.is_dir():
|
||||
shutil.rmtree(templates_dir)
|
||||
|
||||
core_meta: dict
|
||||
core_path = details_dest / CORE_WIRE_NAME
|
||||
core_wire, core_meta = patch_corepayload_wire(core_path.read_bytes(), ver)
|
||||
core_path.write_bytes(core_wire)
|
||||
|
||||
show_path = details_dest / SHOW_WIRE_NAME
|
||||
show_path.write_bytes(
|
||||
patch_show_wire(
|
||||
show_path.read_bytes(),
|
||||
ver=ver,
|
||||
core_sha256=core_meta["sha256"],
|
||||
core_size=core_meta["size"],
|
||||
)
|
||||
)
|
||||
|
||||
secondary_hits = patch_weifile_secondaries(weifile_dest, ver)
|
||||
|
||||
index_path = weifile_dest / "index.js"
|
||||
index_path.write_text(
|
||||
patch_channel_code(index_path.read_text(encoding="utf-8"), code),
|
||||
strip_iptj_beacon(index_path.read_text(encoding="utf-8")),
|
||||
encoding="utf-8",
|
||||
)
|
||||
write_zip(dest, zip_out)
|
||||
apply_landing_template(weifile_dest, landing_template)
|
||||
|
||||
if channel_out.exists():
|
||||
shutil.rmtree(channel_out)
|
||||
channel_out.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copytree(root, channel_out)
|
||||
|
||||
return {
|
||||
"campaign": "xxbb",
|
||||
"builder_type": "new",
|
||||
"channel_code": code,
|
||||
"zip_path": str(zip_out),
|
||||
"weifile_src": str(weifile_src),
|
||||
"channel_ver": ver,
|
||||
"channel_id": ver,
|
||||
"channel_dir": str(channel_out),
|
||||
"landing_path": f"/channel/{ver}/weifile/weifile.html",
|
||||
"landing_template": landing_template,
|
||||
"details_path": f"/channel/{ver}/details/",
|
||||
"show_alias": SHOW_PATH_TMPL.format(ver=ver),
|
||||
"core_sha256": core_meta["sha256"],
|
||||
"core_ver_hits": core_meta["ver_hits"],
|
||||
"core_show_path_hits": core_meta["show_path_hits"],
|
||||
"secondary_show_patches": secondary_hits,
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Pack staged weifile into public/channel-source-new/{CHANNELID}.zip"
|
||||
description=f"Pack exclusive channel tree into public/channel/{{ver}}/ (ver like {ORIGINAL_VER})"
|
||||
)
|
||||
parser.add_argument("--channel-code", required=True, help="8-char channel id (e.g. FAFA9988)")
|
||||
parser.add_argument(
|
||||
"--state-root",
|
||||
"--channel-ver",
|
||||
required=True,
|
||||
help="channel id / ver patch string, X.Y.ZZ alnum (e.g. 2.2.66 or A.B.C1)",
|
||||
)
|
||||
parser.add_argument("--state-root", type=Path, default=None)
|
||||
parser.add_argument("--weifile-src", type=Path, default=None)
|
||||
parser.add_argument(
|
||||
"--details-src",
|
||||
type=Path,
|
||||
default=None,
|
||||
help=f"builder state (default: {xxbb_build.default_state_root()})",
|
||||
help="built details dir (default: <artifact-root>/details)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--weifile-src",
|
||||
type=Path,
|
||||
default=None,
|
||||
help="staged weifile directory (default: <state-root>/out/weifile)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--zip-out",
|
||||
"--channel-out",
|
||||
type=Path,
|
||||
required=True,
|
||||
help="output zip path",
|
||||
help="output directory e.g. public/channel/A.B.C1",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--landing-template",
|
||||
default=DEFAULT_LANDING_TEMPLATE,
|
||||
choices=LANDING_TEMPLATES,
|
||||
help="weifile.html template: test=loading countdown, blank=empty (default: test)",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
state_root = (args.state_root or xxbb_build.default_state_root()).resolve()
|
||||
artifact_root = xxbb_build.default_artifact_root().resolve()
|
||||
weifile_src = (args.weifile_src or staged_weifile_dir(state_root)).resolve()
|
||||
zip_out = args.zip_out.resolve()
|
||||
details_src = (args.details_src or (artifact_root / DETAILS_ROOT)).resolve()
|
||||
channel_out = args.channel_out.resolve()
|
||||
|
||||
result = pack_channel(channel_code=args.channel_code, weifile_src=weifile_src, zip_out=zip_out)
|
||||
result = pack_channel(
|
||||
channel_ver=args.channel_ver,
|
||||
weifile_src=weifile_src,
|
||||
details_src=details_src,
|
||||
channel_out=channel_out,
|
||||
landing_template=args.landing_template,
|
||||
)
|
||||
seeds_path = state_root / LAB_SEEDS_NAME
|
||||
if seeds_path.is_file():
|
||||
try:
|
||||
@@ -132,7 +354,7 @@ def main() -> int:
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
|
||||
print(f"packed {result['channel_code']} -> {zip_out}")
|
||||
print(f"packed {result['channel_ver']} -> {channel_out}")
|
||||
print(RESULT_MARKER + json.dumps(result, separators=(",", ":")))
|
||||
return 0
|
||||
|
||||
|
||||
Reference in New Issue
Block a user