Files

194 lines
6.8 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""CDN 发布与本地联调工具。
目录约定(与 ProjectDesign / Unity BuildAndStage 一致)::
Tools/cdn_root/wechat/version.json
Tools/cdn_root/wechat/1.0.0/DefaultPackage/<yooasset files>
用法::
# 本地 HTTP(HostPlayMode 联调)
python Tools/cdn_publish.py serve --port 8787
# 校验目录
python Tools/cdn_publish.py status
# 上传到远端(需配置)
python Tools/cdn_publish.py upload --mode copy --dest //server/cdn/www
python Tools/cdn_publish.py upload --mode rsync --dest user@host:/var/cdn/www
# 或自定义命令({src} {dest}):
set CDN_UPLOAD_CMD=scp -r {src}/. {dest}/
python Tools/cdn_publish.py upload --mode cmd --dest user@host:/var/www
"""
from __future__ import annotations
import argparse
import json
import os
import shutil
import subprocess
import sys
from functools import partial
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
REPO = Path(__file__).resolve().parents[1]
CDN_ROOT = REPO / "Tools" / "cdn_root"
CHANNEL = "wechat"
DEFAULT_PORT = 8787
class QuietHandler(SimpleHTTPRequestHandler):
def log_message(self, fmt: str, *args) -> None:
sys.stderr.write("[%s] %s\n" % (self.log_date_time_string(), fmt % args))
def end_headers(self) -> None:
# 方便微信/编辑器跨域拉 version.json;生产 CDN 由 Caddy 配
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Cache-Control", "no-cache")
super().end_headers()
def cmd_status(_: argparse.Namespace) -> int:
version = CDN_ROOT / CHANNEL / "version.json"
meta = CDN_ROOT / CHANNEL / "last_publish.json"
print(f"cdn_root={CDN_ROOT}")
print(f"version_json_exists={version.is_file()} path={version}")
if version.is_file():
print(version.read_text(encoding="utf-8"))
print(f"meta_exists={meta.is_file()} path={meta}")
if meta.is_file():
print(meta.read_text(encoding="utf-8"))
pkg_root = CDN_ROOT / CHANNEL
if pkg_root.is_dir():
for p in sorted(pkg_root.rglob("*")):
if p.is_file() and p.name != "version.json":
rel = p.relative_to(CDN_ROOT)
print(f" file {rel} ({p.stat().st_size} bytes)")
return 0
def cmd_serve(args: argparse.Namespace) -> int:
if not CDN_ROOT.is_dir():
print(f"ERROR: missing {CDN_ROOT}, run Unity menu AIGame/YooAsset/Build And Stage first", file=sys.stderr)
return 1
os.chdir(CDN_ROOT)
handler = partial(QuietHandler, directory=str(CDN_ROOT))
server = ThreadingHTTPServer(("0.0.0.0", args.port), handler)
print(f"serving {CDN_ROOT} at http://127.0.0.1:{args.port}/")
print(f"version url: http://127.0.0.1:{args.port}/{CHANNEL}/version.json")
try:
server.serve_forever()
except KeyboardInterrupt:
print("\nstopped")
return 0
def cmd_upload(args: argparse.Namespace) -> int:
src = CDN_ROOT
if not src.is_dir():
print(f"ERROR: missing {src}", file=sys.stderr)
return 1
dest = args.dest
if not dest:
print("ERROR: --dest required", file=sys.stderr)
return 1
# 上传前把 version.prod.json 覆盖为对外 version.json(若存在)
prod = src / CHANNEL / "version.prod.json"
live = src / CHANNEL / "version.json"
if prod.is_file():
shutil.copy2(prod, live)
print(f"promoted {prod.name} -> version.json for upload")
mode = args.mode
if mode == "copy":
dest_path = Path(dest)
dest_path.mkdir(parents=True, exist_ok=True)
# 同步 wechat 通道目录
target = dest_path / CHANNEL
if target.exists():
shutil.rmtree(target)
shutil.copytree(src / CHANNEL, target, ignore=shutil.ignore_patterns("version.prod.json", "last_publish.json"))
print(f"copied {src / CHANNEL} -> {target}")
return 0
if mode == "rsync":
cmd = [
"rsync", "-avz", "--delete",
"--exclude", "version.prod.json",
"--exclude", "last_publish.json",
f"{src / CHANNEL}/",
f"{dest.rstrip('/')}/{CHANNEL}/",
]
print(" ".join(cmd))
return subprocess.call(cmd)
if mode == "cmd":
template = os.environ.get("CDN_UPLOAD_CMD", "").strip()
if not template:
print("ERROR: set CDN_UPLOAD_CMD with {src} and {dest}", file=sys.stderr)
return 1
cmd = template.format(src=str(src / CHANNEL), dest=dest)
print(cmd)
return subprocess.call(cmd, shell=True)
print(f"ERROR: unknown mode {mode}", file=sys.stderr)
return 1
def cmd_write_version(args: argparse.Namespace) -> int:
"""手动改写 version.json(不重新打 AB)。"""
path = CDN_ROOT / CHANNEL / "version.json"
path.parent.mkdir(parents=True, exist_ok=True)
data = {
"appVersionMin": args.app_version,
"appVersionMax": args.app_version,
"resVersion": args.res_version,
"packVersion": args.pack_version,
"cdnRoot": args.cdn_root,
"packages": ["DefaultPackage", "Art"],
"forceUpdate": bool(args.force_update),
"notice": args.notice or "",
}
path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
print(f"wrote {path}")
print(json.dumps(data, ensure_ascii=False, indent=2))
return 0
def main() -> int:
parser = argparse.ArgumentParser(description="AIGame CDN publish helpers")
sub = parser.add_subparsers(dest="cmd", required=True)
p_status = sub.add_parser("status", help="Show staged CDN files")
p_status.set_defaults(func=cmd_status)
p_serve = sub.add_parser("serve", help="Serve Tools/cdn_root over HTTP")
p_serve.add_argument("--port", type=int, default=DEFAULT_PORT)
p_serve.set_defaults(func=cmd_serve)
p_upload = sub.add_parser("upload", help="Upload wechat channel to remote CDN root")
p_upload.add_argument("--mode", choices=("copy", "rsync", "cmd"), default="copy")
p_upload.add_argument("--dest", required=True, help="Remote or local destination root (www root)")
p_upload.set_defaults(func=cmd_upload)
p_ver = sub.add_parser("write-version", help="Rewrite version.json")
p_ver.add_argument("--app-version", default="1.0.0")
p_ver.add_argument("--res-version", required=True, help="MD5 of the Art index bytes")
p_ver.add_argument("--pack-version", default="", help="MD5 of the DefaultPackage index bytes")
p_ver.add_argument("--cdn-root", default="https://www.xworld.ren/game/Friends/webgl/yoo/DefaultPackage")
p_ver.add_argument("--force-update", action="store_true")
p_ver.add_argument("--notice", default="")
p_ver.set_defaults(func=cmd_write_version)
args = parser.parse_args()
return int(args.func(args) or 0)
if __name__ == "__main__":
raise SystemExit(main())