KoalaTrapLord
06f007e605
The guard condition was inverted: `if not deck_row:` entered the query block when no deck was found, causing `deck_row[0]` to crash with 'NoneType' object is not subscriptable. Flipped to `if deck_row:` so the query runs only when the deck exists.
266 lines
9.7 KiB
Python
266 lines
9.7 KiB
Python
"""Deck service: .dck parsing, deck profile saving, deck loading."""
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from models.state import CFG
|
|
from services.database import _get_db_conn, _row_to_dict
|
|
|
|
log_collect = logging.getLogger("lab.collection")
|
|
|
|
|
|
_TYPE_PRIORITY = ["Land", "Instant", "Sorcery", "Artifact", "Enchantment", "Planeswalker", "Creature"]
|
|
_TYPE_TARGETS = {
|
|
"Land": [36, 38],
|
|
"Instant": [9, 11],
|
|
"Sorcery": [7, 9],
|
|
"Artifact": [9, 11],
|
|
"Creature": [20, 30],
|
|
"Enchantment": [5, 10],
|
|
"Planeswalker": [0, 5],
|
|
}
|
|
|
|
|
|
def _classify_card_type(type_line: str) -> str:
|
|
tl = type_line or ""
|
|
for t in _TYPE_PRIORITY:
|
|
if t in tl:
|
|
return t
|
|
return "Other"
|
|
|
|
|
|
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:
|
|
raise HTTPException(404, f"Deck {deck_id} not found")
|
|
d = dict(row)
|
|
try:
|
|
d["color_identity"] = json.loads(d.get("color_identity", "[]"))
|
|
except Exception:
|
|
d["color_identity"] = []
|
|
return d
|
|
|
|
|
|
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,
|
|
ce.type_line, ce.oracle_text, ce.keywords, ce.cmc, ce.color_identity
|
|
FROM deck_cards dc
|
|
LEFT JOIN (
|
|
SELECT scryfall_id, type_line, oracle_text, keywords, cmc, color_identity
|
|
FROM collection_entries GROUP BY scryfall_id
|
|
) 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["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}
|
|
total_cards = 0
|
|
roles_count = {}
|
|
for row in rows:
|
|
r = dict(row)
|
|
qty = r.get("quantity", 1)
|
|
total_cards += qty
|
|
card_type = _classify_card_type(r.get("type_line", ""))
|
|
counts_by_type[card_type] = counts_by_type.get(card_type, 0) + qty
|
|
cmc = int(r.get("cmc", 0) or 0)
|
|
if card_type != "Land":
|
|
key = str(min(cmc, 6)) if cmc < 6 else "6+"
|
|
mana_curve[key] = mana_curve.get(key, 0) + qty
|
|
ci_raw = r.get("color_identity", "[]")
|
|
if isinstance(ci_raw, str):
|
|
try:
|
|
ci_raw = json.loads(ci_raw)
|
|
except Exception:
|
|
ci_raw = []
|
|
for c in (ci_raw or []):
|
|
if c in color_pips:
|
|
color_pips[c] += qty
|
|
role = r.get("role_tag") or "Unknown"
|
|
roles_count[role] = roles_count.get(role, 0) + qty
|
|
return {
|
|
"total_cards": total_cards,
|
|
"counts_by_type": counts_by_type,
|
|
"mana_curve": mana_curve,
|
|
"color_pips": color_pips,
|
|
"total_cards": total_cards,
|
|
"roles": roles_count,
|
|
}
|
|
|
|
|
|
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:
|
|
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:
|
|
"""Parse a Forge .dck file into {name, commanders, mainboard, colorIdentity, totalCards}."""
|
|
path = Path(deck_path)
|
|
if not path.exists():
|
|
return {"error": f"File not found: {deck_path}"}
|
|
lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
|
|
name = path.stem
|
|
section = "main"
|
|
commanders: dict = {}
|
|
mainboard: dict = {}
|
|
commander_name = None
|
|
for line in lines:
|
|
line = line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
if line == "[metadata]":
|
|
section = "metadata"; continue
|
|
if line.startswith("Name="):
|
|
name = line.split("=", 1)[1].strip(); continue
|
|
if line == "[Commander]":
|
|
section = "commander"; continue
|
|
if line == "[Main]":
|
|
section = "main"; continue
|
|
if line.lower().startswith("sideboard") or line == "[Sideboard]":
|
|
section = "sideboard"; continue
|
|
clean = re.sub(r"\(\w+\)\s*\d*$", "", line).strip()
|
|
clean = re.sub(r"\s*\*\.\*$", "", clean).strip()
|
|
m = re.match(r"^(\d+)x?\s+(.+)$", clean)
|
|
qty, card_name = (int(m.group(1)), m.group(2).strip()) if m else (1, clean)
|
|
if not card_name:
|
|
continue
|
|
if section == "commander":
|
|
commanders[card_name] = qty
|
|
if not commander_name:
|
|
commander_name = card_name
|
|
elif section != "sideboard":
|
|
mainboard[card_name] = qty
|
|
if commander_name:
|
|
name = f"{commander_name} - Text Import"
|
|
total = sum(commanders.values()) + sum(mainboard.values())
|
|
return {"name": name, "commander": commander_name, "commanders": commanders, "mainboard": mainboard, "colorIdentity": [], "totalCards": total}
|
|
|
|
|
|
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}")
|
|
lines += ["", "[Main]"]
|
|
for name, qty in profile.get("mainboard", {}).items():
|
|
lines.append(f"{qty} {name}")
|
|
content = "\n".join(lines)
|
|
safe_name = re.sub(r"[^a-zA-Z0-9 _-]", "", profile.get("name", "imported")).replace(" ", "_").strip()
|
|
if not safe_name:
|
|
safe_name = "imported_deck"
|
|
save_dir = CFG.forge_decks_dir
|
|
if not save_dir or not os.path.isdir(save_dir):
|
|
save_dir = os.path.join(Path(__file__).parent.parent, "imported-decks")
|
|
os.makedirs(save_dir, exist_ok=True)
|
|
out_path = Path(save_dir) / f"{safe_name}.dck"
|
|
out_path.write_text(content, encoding="utf-8")
|
|
log_collect.info(f" Saved .dck: {out_path}")
|
|
return out_path
|
|
|
|
|
|
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()
|
|
if deck_row:
|
|
deck_id = deck_row[0]
|
|
rows = conn.execute("""
|
|
SELECT dc.card_name, dc.quantity, dc.is_commander, dc.role_tag, dc.scryfall_id,
|
|
COALESCE(ce.type_line, cr.type_line, '') as type_line,
|
|
COALESCE(ce.oracle_text, cr.oracle_text, '') as oracle_text,
|
|
COALESCE(ce.cmc, cr.cmc, 0) as cmc,
|
|
COALESCE(ce.color_identity, cr.color_identity, '[]') as color_identity,
|
|
COALESCE(ce.mana_cost, '', '') as mana_cost,
|
|
COALESCE(ce.power, '', '') as power,
|
|
COALESCE(ce.toughness, '', '') as toughness,
|
|
COALESCE(ce.keywords, cr.keywords, '[]') as keywords
|
|
FROM deck_cards dc
|
|
LEFT JOIN (
|
|
SELECT scryfall_id, type_line, oracle_text, cmc, color_identity, mana_cost, power, toughness, keywords
|
|
FROM collection_entries GROUP BY scryfall_id
|
|
) ce ON ce.scryfall_id = dc.scryfall_id
|
|
LEFT JOIN (
|
|
SELECT name, type_line, oracle_text, cmc, color_identity, keywords
|
|
FROM card_records GROUP BY name
|
|
) cr ON cr.name = dc.card_name
|
|
WHERE dc.deck_id = ?
|
|
""", (deck_id,)).fetchall()
|
|
return [_row_to_dict(r) for r in rows]
|
|
|
|
|
|
log_deckgen = logging.getLogger("commander_ai_lab.deckgen")
|
|
|
|
|
|
def _build_dck_lines(
|
|
deck_name: str,
|
|
commander_name: str,
|
|
cards: list[dict],
|
|
*,
|
|
resolve_substitutes: bool = False,
|
|
) -> list[str]:
|
|
"""Build Forge .dck file lines from deck name, commander, and card list."""
|
|
lines: list[str] = [
|
|
"[metadata]",
|
|
f"Name={deck_name}",
|
|
"",
|
|
"[Commander]",
|
|
f"1 {commander_name}",
|
|
"",
|
|
"[Main]",
|
|
]
|
|
for card in cards:
|
|
cname = card.get("name", "")
|
|
if resolve_substitutes:
|
|
if card.get("status") == "substituted" and card.get("selected_substitute"):
|
|
cname = card["selected_substitute"]
|
|
if cname and cname != commander_name:
|
|
lines.append(f"{card.get('count', card.get('quantity', 1))} {cname}")
|
|
return lines
|
|
|
|
|
|
def _write_dck_file(
|
|
deck_name: str,
|
|
commander_name: str,
|
|
cards: list[dict],
|
|
decks_dir: str | None = None,
|
|
*,
|
|
fallback_id: int | str = 0,
|
|
resolve_substitutes: bool = False,
|
|
) -> str | None:
|
|
"""Write a Forge .dck file to decks_dir. Returns path on success, None otherwise."""
|
|
if not decks_dir:
|
|
decks_dir = CFG.forge_decks_dir
|
|
if not decks_dir or not os.path.isdir(decks_dir):
|
|
return None
|
|
|
|
safe_name = re.sub(r"[^a-zA-Z0-9_\-\s]", "", deck_name).strip().replace(" ", "_")
|
|
if not safe_name:
|
|
safe_name = f"Deck_{fallback_id}"
|
|
|
|
dck_path = os.path.join(decks_dir, f"{safe_name}.dck")
|
|
lines = _build_dck_lines(
|
|
deck_name, commander_name, cards,
|
|
resolve_substitutes=resolve_substitutes,
|
|
)
|
|
with open(dck_path, "w", encoding="utf-8") as f:
|
|
f.write("\n".join(lines))
|
|
|
|
log_deckgen.info(f" Exported .dck: {dck_path}")
|
|
return dck_path
|