Moxfield import

moxfield.py

@@ -0,0 +1,112 @@
+import asyncio
+import re
+from urllib.parse import urlparse
+
+import httpx
+
+API_BASE = "https://api2.moxfield.com/v3/decks/all"
+
+# A plain "python-httpx/…" UA gets blocked by Moxfield's edge; a normal
+# browser UA does not. There's no server-rendered deck data in the page
+# HTML to fall back to (it's a client-rendered SPA), so this API call is
+# the only viable path — verified by hand against api2.moxfield.com.
+HEADERS = {
+    "User-Agent": (
+        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+        "(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36"
+    ),
+    "Accept": "application/json",
+}
+
+# Boards that represent physical cards belonging to the deck. Sideboard and
+# maybeboard are staging areas, not part of the actual decklist.
+DECK_BOARDS = [
+    "mainboard", "commanders", "companions", "signatureSpells",
+    "attractions", "contraptions", "planes", "schemes", "stickers",
+]
+
+_ID_RE = re.compile(r"^[A-Za-z0-9_-]{6,}$")
+
+
+def parse_deck_id(text: str) -> str:
+    text = text.strip()
+    if not text:
+        raise ValueError("empty Moxfield URL")
+    if "moxfield.com" not in text.lower() and _ID_RE.match(text):
+        return text
+    parts = [p for p in urlparse(text).path.split("/") if p]
+    if len(parts) >= 2 and parts[0] == "decks":
+        return parts[1]
+    raise ValueError("could not parse a deck id from that Moxfield URL")
+
+
+async def fetch_deck(client: httpx.AsyncClient, deck_id: str) -> dict:
+    url = f"{API_BASE}/{deck_id}"
+    r = None
+    for attempt in range(4):
+        r = await client.get(url, headers=HEADERS)
+        if r.status_code == 200:
+            return r.json()
+        if r.status_code == 404:
+            raise ValueError("deck not found on Moxfield (private, deleted, or wrong id?)")
+        if r.status_code == 403:
+            raise ValueError("Moxfield blocked the request (403) — try again in a bit")
+        if r.status_code in (429, 500, 502, 503, 504):
+            wait = float(r.headers.get("Retry-After", 2 ** attempt))
+            await asyncio.sleep(wait)
+            continue
+        r.raise_for_status()
+    r.raise_for_status()
+    raise ValueError("failed to fetch deck from Moxfield")
+
+
+def _is_token(card: dict) -> bool:
+    return card.get("layout") == "token" or (card.get("type_line") or "").startswith("Token")
+
+
+def extract(deck_json: dict, include_tokens: bool) -> dict:
+    """Flatten a Moxfield deck payload into (name, commanders, entries).
+
+    Each entry carries a scryfall_id when Moxfield has one on file, which lets
+    the fetch pipeline pull the exact printing (and both faces of a DFC/MDFC)
+    straight from Scryfall's /cards/{id} instead of a fuzzy name search.
+    """
+    boards = deck_json.get("boards", {})
+    commanders: list[str] = []
+    entries: list[dict] = []
+
+    for board_name in DECK_BOARDS:
+        board = boards.get(board_name) or {}
+        for row in board.get("cards", {}).values():
+            card = row.get("card") or {}
+            name = card.get("name") or ""
+            scry_id = card.get("scryfall_id")
+            qty = max(1, int(row.get("quantity") or 1))
+            if board_name == "commanders" and name:
+                commanders.append(name)
+            for _ in range(qty):
+                entries.append({"name": name, "scry_id": scry_id, "is_token": False})
+
+    if include_tokens:
+        seen_ids = {e["scry_id"] for e in entries if e["scry_id"]}
+        token_cards: dict[str, dict] = {}
+
+        for row in (boards.get("tokens") or {}).get("cards", {}).values():
+            c = row.get("card") or {}
+            if c.get("scryfall_id"):
+                token_cards[c["scryfall_id"]] = c
+
+        for c in (deck_json.get("tokenMappings") or {}).values():
+            if c.get("scryfall_id") and _is_token(c):
+                token_cards[c["scryfall_id"]] = c
+
+        for scry_id, c in token_cards.items():
+            if scry_id in seen_ids:
+                continue
+            entries.append({"name": c.get("name") or "", "scry_id": scry_id, "is_token": True})
+
+    return {
+        "name": deck_json.get("name") or "Moxfield Deck",
+        "commanders": commanders,
+        "entries": entries,
+    }