Spaces:
Running on Zero
Running on Zero
| """Read the dex for the Browse tab, and fetch it when it is not local. | |
| The dex lives in a private GitHub repo. The Space that renders it is public - | |
| that is the price of gating access with Hugging Face OAuth on a free | |
| organization - so the dex must never be committed into the Space. Instead the | |
| Space pulls a tarball at startup with a read-only token held in Space secrets, | |
| and what visitors see comes over the wire rather than out of the repo. | |
| Locally there is no fetching: DEX_ROOT points at the working copy. | |
| """ | |
| import io | |
| import json | |
| import logging | |
| import os | |
| import shutil | |
| import tarfile | |
| import threading | |
| import time | |
| import urllib.request | |
| log = logging.getLogger(__name__) | |
| DEFAULT_LOCAL = os.path.expanduser("~/fakemon-dex") | |
| CACHE = os.environ.get("DEX_CACHE", "/tmp/fakemon-dex-snapshot") | |
| REPO = os.environ.get("DEX_REPO", "tikitakatoe56-rgb/fakemon-dex") | |
| REF = os.environ.get("DEX_REF", "main") | |
| TTL = int(os.environ.get("DEX_TTL", "900")) # re-fetch at most every 15 min | |
| # One fetch at a time: two of them share CACHE + ".new", so a second fetch | |
| # starting mid-swap used to extract into a directory the first one was about | |
| # to rename away, and one of the two then failed on a source that had moved. | |
| _FETCHING = threading.Lock() | |
| def _swap(tmp): | |
| """Put the freshly extracted tmp directory in place of the live cache. | |
| The old snapshot is moved aside rather than deleted first so that a | |
| failure here leaves a working dex behind, and the move-aside target is | |
| cleared first: a rename onto a non-empty .old raises ENOTEMPTY, so one | |
| interrupted swap would otherwise wedge every fetch from then on - the | |
| Browse tab silently serving the same snapshot for the life of the | |
| container, which is exactly what a reload button must not do. | |
| """ | |
| old = CACHE + ".old" | |
| shutil.rmtree(old, ignore_errors=True) | |
| if not os.path.isdir(CACHE): | |
| os.rename(tmp, CACHE) | |
| return CACHE | |
| os.rename(CACHE, old) | |
| try: | |
| os.rename(tmp, CACHE) | |
| except OSError: | |
| os.rename(old, CACHE) | |
| raise | |
| shutil.rmtree(old, ignore_errors=True) | |
| return CACHE | |
| def _fetch(token): | |
| """Download the private repo as a tarball and keep only dex/. | |
| A tarball is one request for the whole dex; the contents API would be one | |
| request per file, and an entry is twenty files. | |
| """ | |
| url = f"https://api.github.com/repos/{REPO}/tarball/{REF}" | |
| req = urllib.request.Request(url, headers={ | |
| "Authorization": f"Bearer {token}", | |
| "Accept": "application/vnd.github+json", | |
| "User-Agent": "fakemon-workshop"}) | |
| with urllib.request.urlopen(req, timeout=60) as r: | |
| blob = r.read() | |
| tmp = CACHE + ".new" | |
| # Emptied, not reused: an interrupted fetch leaves files here, and merging | |
| # them into the next snapshot would resurrect entries the dex has deleted. | |
| shutil.rmtree(tmp, ignore_errors=True) | |
| os.makedirs(tmp) | |
| kept = 0 | |
| with tarfile.open(fileobj=io.BytesIO(blob), mode="r:gz") as tar: | |
| for member in tar.getmembers(): | |
| # Strip GitHub's <owner>-<repo>-<sha>/ prefix, keep dex/ only, and | |
| # refuse anything that would escape the target directory. | |
| parts = member.name.split("/", 1) | |
| if len(parts) != 2 or not parts[1].startswith("dex/"): | |
| continue | |
| if not member.isfile() or os.path.isabs(parts[1]) \ | |
| or ".." in parts[1].split("/"): | |
| continue | |
| dest = os.path.join(tmp, parts[1]) | |
| os.makedirs(os.path.dirname(dest), exist_ok=True) | |
| src = tar.extractfile(member) | |
| if src is None: | |
| continue | |
| with open(dest, "wb") as f: | |
| f.write(src.read()) | |
| kept += 1 | |
| if not kept: | |
| # A tarball with no dex/ in it means the repo, the ref or the token | |
| # changed under us - not that the dex is empty. Swapping it in would | |
| # blank the Browse tab and, because the snapshot would then be fresh | |
| # by mtime, keep it blank for a TTL. Fail instead and keep what works. | |
| shutil.rmtree(tmp, ignore_errors=True) | |
| raise RuntimeError(f"no dex/ files in the {REPO}@{REF} tarball") | |
| return _swap(tmp) | |
| def _mtime(): | |
| try: | |
| return os.path.getmtime(CACHE) | |
| except OSError: | |
| return 0.0 | |
| def root(force=False): | |
| """Where dex/ can be read from right now, or None if it cannot be. | |
| force=True is what the Browse tab's reload button asks for: skip the TTL | |
| and go and look. Page loads leave it False, so an ordinary visit still | |
| costs nothing beyond a stat of the snapshot. | |
| """ | |
| local = os.environ.get("DEX_ROOT", DEFAULT_LOCAL) | |
| if os.path.isdir(os.path.join(local, "dex")): | |
| return local | |
| token = os.environ.get("DEX_TOKEN") | |
| if not token: | |
| log.warning("dex: no DEX_TOKEN, nothing to read the dex from") | |
| return None | |
| started = _mtime() | |
| if started and time.time() - started < TTL and not force: | |
| return CACHE | |
| with _FETCHING: | |
| if _mtime() > started: | |
| # Someone else fetched while this call waited for the lock; their | |
| # snapshot is the one this call was about to go and get. | |
| return CACHE | |
| try: | |
| return _fetch(token) | |
| except Exception: | |
| # A stale dex is better than an empty one: the Browse tab is the | |
| # part people look at when the generator is busy or out of quota. | |
| # Silent is what it must not be - a reload that quietly served the | |
| # old snapshot is indistinguishable from a reload that did nothing. | |
| stale = os.path.isdir(CACHE) | |
| log.exception("dex: fetch from %s@%s failed; serving %s", REPO, | |
| REF, "the previous snapshot" if stale else "nothing") | |
| return CACHE if stale else None | |
| def _load(path): | |
| try: | |
| with open(path) as f: | |
| return json.load(f) | |
| except (OSError, ValueError): | |
| return {} | |
| def _title(token, prefix): | |
| """ABILITY_SWIFT_SWIM -> Swift Swim. The dex stores game constants; | |
| a reader wants the name the game prints.""" | |
| return token.replace(prefix, "", 1).replace("_", " ").title() | |
| def _abilities(species): | |
| """The abilities worth naming, in slot order and without repeats. | |
| ABILITY_NONE is a filled slot in the C table, not an ability, and the | |
| second slot is NONE far more often than not. | |
| """ | |
| out = [] | |
| for a in species.get("abilities") or []: | |
| if a in ("ABILITY_NONE", "ABILITY_UNKNOWN"): | |
| continue | |
| name = _title(a, "ABILITY_") | |
| if name not in out: | |
| out.append(name) | |
| return out | |
| def _evolutions(species): | |
| """Where this stage goes next, as {target, method, param}. | |
| Read from species.json rather than family.json: the species file is what | |
| the game is built from, and it is per-stage, so a stage knows its own | |
| evolution without anyone having to work out where it sits in the family. | |
| """ | |
| # evolvesTo is the flat schema the workshop writes and the ROM is built | |
| # from; ui.accept() pops "evolutions" precisely so the two cannot drift | |
| # apart. Reading the popped key meant every stage reported no evolution, | |
| # and the dex drew every member of a line as a creature that stands alone. | |
| # The old key is still read so entries written before the change render. | |
| out = [] | |
| for e in (species.get("evolvesTo") or species.get("evolutions") or []): | |
| target = e.get("target") | |
| if target: | |
| out.append({"target": target, "method": e.get("method"), | |
| "param": e.get("param")}) | |
| return out | |
| def entries(dex_root): | |
| """Every stage in the dex, dex order first, as flat dicts. | |
| Stages are listed individually rather than folded into their family: a | |
| three-stage line is three creatures somebody designed, and collapsing them | |
| into one card is what made the evolution work invisible in the README. | |
| The evolution links travel with each stage, so a reader can be shown the | |
| line without the flat list having to become a tree. | |
| """ | |
| base = os.path.join(dex_root, "dex") | |
| if not os.path.isdir(base): | |
| return [] | |
| order = _load(os.path.join(base, "registry.json")).get("entries", {}) | |
| out = [] | |
| for name in sorted(os.listdir(base)): | |
| entry = os.path.join(base, name) | |
| if not os.path.isdir(entry): | |
| continue | |
| stage_root = os.path.join(entry, "stages") | |
| dirs = ([(st, os.path.join(stage_root, st)) | |
| for st in sorted(os.listdir(stage_root))] | |
| if os.path.isdir(stage_root) else [(None, entry)]) | |
| for st, sdir in dirs: | |
| front = os.path.join(sdir, "front.png") | |
| if not os.path.isfile(front): | |
| continue | |
| spec = _load(os.path.join(sdir, "spec.json")) | |
| species = _load(os.path.join(sdir, "species.json")) | |
| # The registry is keyed per creature, not per family: a three-stage | |
| # line holds three IDs, and the stage directory carries the name | |
| # they are filed under. Looking up the family directory gave every | |
| # stage of a line the same number. | |
| dex_id = order.get(st) if st else None | |
| if dex_id is None: | |
| dex_id = order.get(str(spec.get("name", "")).lower()) | |
| if dex_id is None and st is None: | |
| dex_id = order.get(name) | |
| stats = {k: species.get(v) for k, v in ( | |
| ("HP", "baseHP"), ("ATK", "baseAttack"), | |
| ("DEF", "baseDefense"), ("SPA", "baseSpAttack"), | |
| ("SPD", "baseSpDefense"), ("SPE", "baseSpeed"))} | |
| back = os.path.join(sdir, "back.png") | |
| out.append({ | |
| "key": name if st is None else f"{name}/{st}", | |
| # The family is what the spec says, not what the directory is | |
| # called. Every stage has its own directory now, so keying off | |
| # the directory gave each member of a line a family of one - | |
| # and _family() draws no evolution line for a family of one. | |
| "family": spec.get("family") or name, | |
| "id": dex_id, | |
| "name": spec.get("name") or species.get("speciesName") or name, | |
| "types": spec.get("types") or [ | |
| t.replace("TYPE_", "").lower() | |
| for t in species.get("types", [])], | |
| "trainer": spec.get("trainer") or spec.get("creator") | |
| or "unknown", | |
| "description": species.get("description", ""), | |
| "category": species.get("categoryName", ""), | |
| "height": species.get("height"), "weight": species.get("weight"), | |
| "stats": stats, | |
| "bst": sum(v for v in stats.values() if isinstance(v, int)), | |
| "abilities": _abilities(species), | |
| "evolutions": _evolutions(species), | |
| "moves": [[lvl, _title(mv, "MOVE_")] | |
| for lvl, mv in species.get("levelUpLearnset") or []], | |
| "front": front, | |
| # Back sprites arrive from a separate render and a few entries | |
| # predate them, so this is None rather than a path that 404s. | |
| "back": back if os.path.isfile(back) else None, | |
| }) | |
| out.sort(key=lambda e: (e["id"] is None, e["id"] or 0, e["key"])) | |
| return out | |