vosonandCursor
a257f96d94
Export and OSS upload require wide Heroes-page cards only; bump cache stamp to wide2 so clients drop polluted squares. Co-authored-by: Cursor <cursoragent@cursor.com>
237 lines
7.5 KiB
Python
237 lines
7.5 KiB
Python
"""Sync Climperor web site static images to Aliyun OSS (keyzoo inject).
|
||
|
||
Builds the same portrait/item/ability/... tree as export_relations_site and
|
||
uploads under oss://<bucket>/{attr,item,portrait,...}/.
|
||
|
||
Env (from keyzoo asset_exec on digitevents/voson-RAM):
|
||
KEYZOO_ASSET_META_ACCESSKEY_ID
|
||
KEYZOO_ASSET_SECRET_ACCESSKEY_SECRET
|
||
|
||
Usage:
|
||
python _oss_static_assets.py upload [--force]
|
||
python _oss_static_assets.py verify
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import mimetypes
|
||
import os
|
||
import shutil
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
import urllib.request
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||
|
||
import oss2
|
||
|
||
from shared.paths import ROOT
|
||
|
||
from export_relations_site import DEFAULT_OSS_BASE, populate_static_assets
|
||
from serve_relations import build_payload
|
||
|
||
ENDPOINT = "https://oss-cn-shanghai.aliyuncs.com"
|
||
REGION = "cn-shanghai"
|
||
ASSET_DIRS = (
|
||
"attr",
|
||
"role-icon",
|
||
"rank",
|
||
"item",
|
||
"item-cat",
|
||
"ability",
|
||
"ui-icon",
|
||
"portrait",
|
||
"streamer-avatar",
|
||
"streamer-video",
|
||
)
|
||
|
||
|
||
def _creds() -> tuple[str, str]:
|
||
ak = os.environ.get("KEYZOO_ASSET_META_ACCESSKEY_ID") or os.environ.get(
|
||
"OSS_ACCESS_KEY_ID"
|
||
)
|
||
sk = os.environ.get("KEYZOO_ASSET_SECRET_ACCESSKEY_SECRET") or os.environ.get(
|
||
"OSS_ACCESS_KEY_SECRET"
|
||
)
|
||
if not ak or not sk:
|
||
raise SystemExit(
|
||
"missing AccessKey: inject via keyzoo or set OSS_ACCESS_KEY_ID / "
|
||
"OSS_ACCESS_KEY_SECRET"
|
||
)
|
||
return ak, sk
|
||
|
||
|
||
def _bucket(name: str) -> oss2.Bucket:
|
||
return oss2.Bucket(oss2.Auth(*_creds()), ENDPOINT, name)
|
||
|
||
|
||
def _resolve_bucket(explicit: str | None) -> str:
|
||
if explicit:
|
||
return explicit
|
||
state = ROOT / ".oss_ability_videos_bucket"
|
||
if state.is_file():
|
||
return state.read_text(encoding="utf-8").strip()
|
||
return "climperor"
|
||
|
||
|
||
def _build_staging() -> tuple[Path, dict[str, int]]:
|
||
payload = build_payload()
|
||
staging = Path(tempfile.mkdtemp(prefix="climperor-static-"))
|
||
counts = populate_static_assets(staging, payload)
|
||
return staging, counts
|
||
|
||
|
||
def _png_size(path: Path) -> tuple[int, int] | None:
|
||
"""Return (width, height) for a PNG, or None if unreadable."""
|
||
try:
|
||
raw = path.read_bytes()
|
||
except OSError:
|
||
return None
|
||
if len(raw) < 24 or raw[:8] != b"\x89PNG\r\n\x1a\n":
|
||
return None
|
||
# IHDR: length(4) + type(4) + width(4) + height(4)
|
||
if raw[12:16] != b"IHDR":
|
||
return None
|
||
w = int.from_bytes(raw[16:20], "big")
|
||
h = int.from_bytes(raw[20:24], "big")
|
||
return w, h
|
||
|
||
|
||
def _assert_wide_portraits(staging: Path) -> None:
|
||
"""Refuse top-bar match crops (square ~96×96) before any OSS put.
|
||
|
||
Official cards are usually 256×144; Steam occasionally ships half-res
|
||
128×72 landscape (still fine for the grid). Match templates are square.
|
||
"""
|
||
portrait_dir = staging / "portrait"
|
||
if not portrait_dir.is_dir():
|
||
raise SystemExit("staging missing portrait/ — run fetch_hero_portraits.py")
|
||
bad: list[str] = []
|
||
for path in sorted(portrait_dir.glob("*.png")):
|
||
size = _png_size(path)
|
||
if size is None:
|
||
bad.append(f"{path.name}: unreadable")
|
||
continue
|
||
w, h = size
|
||
# Landscape Heroes cards: aspect ≈ 16:9. Match CDN faces are square.
|
||
if h <= 0 or w / h < 1.4:
|
||
bad.append(f"{path.name}: {w}x{h}")
|
||
if bad:
|
||
sample = "; ".join(bad[:6])
|
||
more = f" (+{len(bad) - 6} more)" if len(bad) > 6 else ""
|
||
raise SystemExit(
|
||
f"refusing to upload {len(bad)} non-wide portrait(s) "
|
||
f"(need landscape aspect ≥1.4, not match-template squares): "
|
||
f"{sample}{more}. Run: python web/fetch_hero_portraits.py"
|
||
)
|
||
|
||
|
||
def _iter_files(root: Path) -> list[Path]:
|
||
files: list[Path] = []
|
||
for sub in ASSET_DIRS:
|
||
d = root / sub
|
||
if not d.is_dir():
|
||
continue
|
||
files.extend(sorted(p for p in d.rglob("*") if p.is_file()))
|
||
return files
|
||
|
||
|
||
def upload(bucket_name: str, *, force: bool = False) -> None:
|
||
staging, counts = _build_staging()
|
||
try:
|
||
_assert_wide_portraits(staging)
|
||
files = _iter_files(staging)
|
||
total_bytes = sum(f.stat().st_size for f in files)
|
||
print(f"staging {staging} ({len(files)} files, {total_bytes / 1e6:.1f} MB)")
|
||
for name, n in counts.items():
|
||
print(f" {name}/: {n}")
|
||
|
||
bucket = _bucket(bucket_name)
|
||
remote_sizes: dict[str, int] = {}
|
||
if not force:
|
||
print("listing remote static objects ...")
|
||
for sub in ASSET_DIRS:
|
||
for obj in oss2.ObjectIterator(bucket, prefix=f"{sub}/"):
|
||
remote_sizes[obj.key] = int(obj.size)
|
||
print(f"remote keys indexed: {len(remote_sizes)}")
|
||
|
||
uploaded = skipped = failed = 0
|
||
t0 = time.time()
|
||
for i, path in enumerate(files, 1):
|
||
key = path.relative_to(staging).as_posix()
|
||
size = path.stat().st_size
|
||
if not force and remote_sizes.get(key) == size:
|
||
skipped += 1
|
||
continue
|
||
headers: dict[str, str] = {}
|
||
ctype, _ = mimetypes.guess_type(str(path))
|
||
if ctype:
|
||
headers["Content-Type"] = ctype
|
||
try:
|
||
bucket.put_object_from_file(key, str(path), headers=headers)
|
||
uploaded += 1
|
||
if i % 100 == 0 or i == len(files) or uploaded <= 5:
|
||
print(f"[{i}/{len(files)}] ok {key} ({size} bytes)")
|
||
except Exception as exc: # noqa: BLE001
|
||
failed += 1
|
||
print(f"[{i}/{len(files)}] FAIL {key}: {exc}")
|
||
|
||
elapsed = time.time() - t0
|
||
print(
|
||
f"done: uploaded={uploaded} skipped={skipped} failed={failed} "
|
||
f"elapsed={elapsed:.0f}s"
|
||
)
|
||
if failed:
|
||
raise SystemExit(1)
|
||
finally:
|
||
shutil.rmtree(staging, ignore_errors=True)
|
||
|
||
|
||
def verify(bucket_name: str, samples: int = 8) -> None:
|
||
staging, _ = _build_staging()
|
||
try:
|
||
files = _iter_files(staging)
|
||
if not files:
|
||
raise SystemExit("no files in staging — run fetch scripts first")
|
||
base = f"https://{bucket_name}.oss-{REGION}.aliyuncs.com"
|
||
step = max(1, len(files) // samples)
|
||
picks = [files[i] for i in range(0, len(files), step)][:samples]
|
||
ok = 0
|
||
for path in picks:
|
||
key = path.relative_to(staging).as_posix()
|
||
url = f"{base}/{key}"
|
||
try:
|
||
req = urllib.request.Request(url, method="HEAD")
|
||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||
print(f" {resp.status} {resp.headers.get('Content-Type', '')} {url}")
|
||
if resp.status == 200:
|
||
ok += 1
|
||
except Exception as exc: # noqa: BLE001
|
||
print(f" FAIL {url}: {exc}")
|
||
print(f"spot-check ok: {ok}/{len(picks)}")
|
||
if ok < len(picks):
|
||
raise SystemExit(1)
|
||
finally:
|
||
shutil.rmtree(staging, ignore_errors=True)
|
||
|
||
|
||
def main() -> None:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("cmd", choices=("upload", "verify"))
|
||
ap.add_argument("--bucket", default=None)
|
||
ap.add_argument("--force", action="store_true")
|
||
args = ap.parse_args()
|
||
name = _resolve_bucket(args.bucket)
|
||
print(f"bucket: {name} public_base: {DEFAULT_OSS_BASE}")
|
||
if args.cmd == "upload":
|
||
upload(name, force=args.force)
|
||
else:
|
||
verify(name)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|