113 lines
4.1 KiB
Python
113 lines
4.1 KiB
Python
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,
|
|
}
|