diff --git a/services/card_analysis.py b/services/card_analysis.py index 2e9e69b..51dcc3f 100644 --- a/services/card_analysis.py +++ b/services/card_analysis.py @@ -15,7 +15,7 @@ def _detect_card_roles(oracle_text: str, type_line: str, keywords) -> list: kw_raw = [] kw = " ".join(str(k).lower() for k in kw_raw) - if ("add {" in ot or "add one mana" in ot + if ("add {" in ot or "add one mana" in ot or ("search your library for a" in ot and "land" in ot) or "treasure token" in ot or "add mana" in ot): roles.append("Ramp") @@ -29,7 +29,7 @@ def _detect_card_roles(oracle_text: str, type_line: str, keywords) -> list: or re.search(r"deals? x damage to (target|any target)", ot) or "target creature gets -" in ot or "fights target" in ot): roles.append("Removal") - if ("destroy all" in ot or "exile all" in ot + if ("destroy all" in ot or "exile all" in ot or re.search(r"all creatures get -\d+/-\d+", ot) or re.search(r"each creature gets -\d+/-\d+", ot) or ("deals" in ot and "to each creature" in ot)): @@ -47,7 +47,7 @@ def _detect_card_roles(oracle_text: str, type_line: str, keywords) -> list: if ("target creature gains" in ot and ("hexproof" in ot or "indestructible" in ot or "shroud" in ot))\ or ("protection" in kw): roles.append("Protection") - if ("sacrifice" in ot and ("whenever" in ot or "you may" in ot)): + if ("sacrifice" in ot and ("whenever" in ot or "you may" in ot)): roles.append("Sacrifice") if ("return" in ot and ("from your graveyard" in ot and "onto the battlefield" in ot) or ("put" in ot and "from a graveyard" in ot and "onto the battlefield" in ot)): @@ -61,7 +61,7 @@ def _detect_card_roles(oracle_text: str, type_line: str, keywords) -> list: or "each opponent loses" in ot and "life" in ot or "deals damage to each opponent" in ot): roles.append("Burn") - if ("can't cast" in ot or "can't attack" in ot or "can't activate" in ot + if ("can't cast" in ot or "can't attack" in ot or "can't activate" in ot or ("enters the battlefield tapped" in ot and "opponents" in ot) or "each player can't" in ot or "players can't" in ot or ("cost {" in ot and "more to cast" in ot)): diff --git a/services/deck_service.py b/services/deck_service.py index e01d855..7aeccbc 100644 --- a/services/deck_service.py +++ b/services/deck_service.py @@ -34,7 +34,7 @@ def _classify_card_type(type_line: str) -> str: return "Other" - def _get_deck_or_404(deck_id: int): +def _get_deck_or_404(deck_id: int): conn = _get_db_conn() row = conn.execute("SELECT * FROM decks WHERE id = ?", (deck_id,)).fetchone() if not row: @@ -47,7 +47,7 @@ def _classify_card_type(type_line: str) -> str: return d - def _compute_deck_analysis(deck_id: int) -> dict: +def _compute_deck_analysis(deck_id: int) -> dict: conn = _get_db_conn() rows = conn.execute(""" SELECT dc.id, dc.scryfall_id, dc.card_name, dc.quantity, dc.is_commander, dc.role_tag, @@ -59,7 +59,7 @@ def _classify_card_type(type_line: str) -> str: ) ce ON ce.scryfall_id = dc.scryfall_id WHERE dc.deck_id = ? """, (deck_id,)).fetchall() - counts_by_type = {t: 0 for t in _TYPE_PRIORITY} + counts_by_type = {t: 0 for t in _TYPE_PRIORITY} counts_by_type["Other"] = 0 mana_curve = {"0": 0, "1": 0, "2": 0, "3": 0, "4": 0, "5": 0, "6+": 0} color_pips = {"W": 0, "U": 0, "B": 0, "R": 0, "G": 0, "C": 0} @@ -96,21 +96,21 @@ def _classify_card_type(type_line: str) -> str: } - def _check_ratio_limit(deck_id: int, card_type: str, count_to_add: int = 1) -> bool: +def _check_ratio_limit(deck_id: int, card_type: str, count_to_add: int = 1) -> bool: analysis = _compute_deck_analysis(deck_id) current = analysis["counts_by_type"].get(card_type, 0) target_max = _TYPE_TARGETS.get(card_type, [0, 9999])[1] return (current + count_to_add) <= target_max - def _to_edhrec_slug(name: str) -> str: +def _to_edhrec_slug(name: str) -> str: slug = name.lower() slug = re.sub(r"[',.]" , "", slug) slug = re.sub(r"[^a-z0-9]+", "-", slug) return slug.strip("-") - def parse_dck_file(deck_path: str) -> dict: +def parse_dck_file(deck_path: str) -> dict: """Parse a Forge .dck file into {name, commanders, mainboard, colorIdentity, totalCards}.""" path = Path(deck_path) if not path.exists(): @@ -153,7 +153,7 @@ def _classify_card_type(type_line: str) -> str: return {"name": name, "commander": commander_name, "commanders": commanders, "mainboard": mainboard, "colorIdentity": [], "totalCards": total} - def _save_profile_to_dck(profile: dict) -> Path: +def _save_profile_to_dck(profile: dict) -> Path: lines = ["[metadata]", f"Name={profile.get('name', 'Imported Deck')}", "", "[Commander]"] for name, qty in profile.get("commanders", {}).items(): lines.append(f"{qty} {name}") @@ -174,7 +174,7 @@ def _classify_card_type(type_line: str) -> str: return out_path - def _load_deck_cards_by_name(deck_name: str) -> list: +def _load_deck_cards_by_name(deck_name: str) -> list: """Load cards from a deck by name, returning list of card dicts.""" conn = _get_db_conn() deck_row = conn.execute("SELECT id FROM decks WHERE name = ?", (deck_name,)).fetchone() diff --git a/services/forge_runner.py b/services/forge_runner.py index 916acea..f64d91a 100644 --- a/services/forge_runner.py +++ b/services/forge_runner.py @@ -38,7 +38,7 @@ def get_java17() -> str: return _JAVA17_PATH - def build_java_command( +def build_java_command( decks: list, num_games: int, threads: int, seed: Optional[int], clock: int, output_path: str, use_learned_policy: bool = False, policy_server: str = "http://localhost:8080", policy_style: str = "midrange", @@ -70,7 +70,7 @@ def get_java17() -> str: return cmd - def _run_process_blocking(state: BatchState, cmd: list): +def _run_process_blocking(state: BatchState, cmd: list): """Run Forge subprocess, parse progress, generate deck reports.""" log.info(f"Running: {' '.join(cmd)}") state.started_at = datetime.datetime.now().isoformat() @@ -91,7 +91,7 @@ def get_java17() -> str: state.error = str(e) log.error(f"Failed to start subprocess: {e}") return - last_activity = [time.monotonic()] + last_activity = [time.monotonic()] stall_limit = 300 # 5 min def watchdog(): while proc.poll() is None: @@ -127,7 +127,7 @@ def get_java17() -> str: state.ended_at = datetime.datetime.now().isoformat() - async def run_batch_subprocess( +async def run_batch_subprocess( state: BatchState, decks: list, num_games: int, threads: int, seed: Optional[int] = None, clock: int = 6000, output_path: str = "results", use_learned_policy: bool = False, **kwargs, diff --git a/services/scryfall.py b/services/scryfall.py index e905ece..5c82f2c 100644 --- a/services/scryfall.py +++ b/services/scryfall.py @@ -11,7 +11,7 @@ from urllib.request import Request, urlopen # Lazy import to avoid circular dependency def _detect_card_roles(oracle_text, type_line, keywords): - from services.card_analysis import _detect_card_roles as _dcr + from services.card_analysis import _detect_card_roles as _dcr return _dcr(oracle_text, type_line, keywords) @@ -31,7 +31,7 @@ class _ScryfallCache: self._misses = 0 self._init_db() - def _init_db(self): + def _init_db(self): self._conn = sqlite3.connect(str(self._db_path), check_same_thread=False) self._conn.execute("PRAGMA journal_mode=WAL") self._conn.execute(""" @@ -43,11 +43,11 @@ class _ScryfallCache: """) self._conn.commit() - @staticmethod + @staticmethod def _make_key(name: str, set_code: str, collector_number: str) -> str: return f"{name.strip().lower()}|{(set_code or '').strip().lower()}|{(collector_number or '').strip()}" - def get(self, name: str, set_code: str = "", collector_number: str = "") -> Optional[dict]: + def get(self, name: str, set_code: str = "", collector_number: str = "") -> Optional[dict]: key = self._make_key(name, set_code, collector_number) with self._lock: row = self._conn.execute( @@ -59,7 +59,7 @@ class _ScryfallCache: self._misses += 1 return None - def put(self, name: str, set_code: str, collector_number: str, data: dict): + def put(self, name: str, set_code: str, collector_number: str, data: dict): key = self._make_key(name, set_code, collector_number) blob = json.dumps(data) with self._lock: @@ -69,12 +69,12 @@ class _ScryfallCache: ) self._conn.commit() - def stats(self) -> dict: + def stats(self) -> dict: with self._lock: count = self._conn.execute("SELECT COUNT(*) FROM scryfall_cache").fetchone()[0] return {"cached": count, "hits": self._hits, "misses": self._misses} - def clear(self) -> int: + def clear(self) -> int: with self._lock: count = self._conn.execute("SELECT COUNT(*) FROM scryfall_cache").fetchone()[0] self._conn.execute("DELETE FROM scryfall_cache") @@ -95,7 +95,7 @@ class _ScryfallCache: return deleted - _scryfall_cache = _ScryfallCache() +_scryfall_cache = _ScryfallCache() _scryfall_lock = threading.Lock() _scryfall_last_call = 0.0 @@ -110,7 +110,7 @@ def _scryfall_rate_limit(): _scryfall_last_call = time.monotonic() - def _fetch_scryfall_api(name: str, set_code: str = "", collector_number: str = "") -> dict: +def _fetch_scryfall_api(name: str, set_code: str = "", collector_number: str = "") -> dict: from urllib.parse import quote cached = _scryfall_cache.get(name, set_code, collector_number) if cached is not None: @@ -133,7 +133,7 @@ def _scryfall_rate_limit(): card_data = json.loads(resp.read().decode("utf-8")) except Exception as e: last_error = str(e) - if not card_data or card_data.get("object") == "error": + if not card_data or card_data.get("object") == "error": err_detail = card_data.get("details", "unknown error") if card_data else last_error return {"_error": f"Scryfall returned error for '{name}': {err_detail}"} _scryfall_cache.put(name, set_code, collector_number, card_data) @@ -143,7 +143,7 @@ def _scryfall_rate_limit(): return card_data - def _enrich_from_scryfall(name: str, set_code: str = "", collector_number: str = "") -> dict: +def _enrich_from_scryfall(name: str, set_code: str = "", collector_number: str = "") -> dict: card_data = _fetch_scryfall_api(name, set_code, collector_number) if not card_data or "_error" in card_data: return card_data or {} @@ -165,7 +165,7 @@ def _scryfall_rate_limit(): oracle_text = card_data.get("oracle_text", "") if not oracle_text and card_data.get("card_faces"): oracle_text = "\n//\n".join(f.get("oracle_text", "") for f in card_data["card_faces"] if f.get("oracle_text")) - power = card_data.get("power", "") + power = card_data.get("power", "") toughness = card_data.get("toughness", "") if not power and card_data.get("card_faces"): power = card_data["card_faces"][0].get("power", "") @@ -199,7 +199,7 @@ def _scryfall_rate_limit(): } - def _scryfall_fuzzy_lookup(name: str) -> Optional[dict]: +def _scryfall_fuzzy_lookup(name: str) -> Optional[dict]: """Fuzzy-search Scryfall for a card name. Returns raw Scryfall JSON dict, or None on failure.""" from urllib.parse import quote _scryfall_rate_limit()