Phase 2b: Extract deck service to services/deck_service.py (#11)

services/deck_service.py

@@ -0,0 +1,194 @@
+"""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 not deck_row:
+        return []
+    deck_id = deck_row[0]
+    rows = conn.execute("""
+        SELECT dc.card_name, dc.quantity, dc.is_commander, dc.role_tag, dc.scryfall_id,
+               ce.type_line, ce.oracle_text, ce.cmc, ce.color_identity, ce.mana_cost
+        FROM deck_cards dc
+        LEFT JOIN (
+            SELECT scryfall_id, type_line, oracle_text, cmc, color_identity, mana_cost
+            FROM collection_entries GROUP BY scryfall_id
+        ) ce ON ce.scryfall_id = dc.scryfall_id
+        WHERE dc.deck_id = ?
+    """, (deck_id,)).fetchall()
+    return [_row_to_dict(r) for r in rows]