""" Commander AI Lab — V3 Deck Generator Service ═════════════════════════════════════════════ Generates Commander decks using Anthropic Claude structured JSON output, then runs ownership check and smart substitution. """ import json import logging import re import sqlite3 from typing import List, Optional, Tuple from urllib.request import urlopen, Request from ..clients.anthropic_client import AnthropicClient from ..schemas.deck_schema import DECK_LIST_SCHEMA, GeneratedDeckList from ..schemas.substitution_schema import ( CardAlternative, DeckCardWithStatus, SubstitutionBatchResult, ) from ..prompts.deck_gen_system import SYSTEM_PROMPT from ..prompts.deck_gen_user import build_user_prompt from ..prompts.substitution_prompt import ( SUBSTITUTION_SYSTEM, build_substitution_prompt, ) from ..config import ( DECK_GEN_MODEL, DECK_GEN_TEMPERATURE, DECK_GEN_MAX_TOKENS, SUBSTITUTION_MIN_SIMILARITY, SUBSTITUTION_MAX_ALTERNATIVES, SUBSTITUTION_MODEL, SUBSTITUTION_USE_ANTHROPIC_FALLBACK, ) logger = logging.getLogger("coach.deckgen") class DeckGeneratorV3: """ V3 Deck Generator using Anthropic Claude structured output. Pipeline: 1. Resolve commander via Scryfall 2. Build collection summary 3. Call Anthropic Claude with structured JSON schema 4. Validate card names via Scryfall 5. Check ownership against collection DB 6. Run Smart Substitution (embedding + Anthropic Claude fallback) 7. Return finalized deck with substitution data """ def __init__(self, anthropic_client: AnthropicClient, db_conn_factory, embedding_index=None): """ Args: anthropic_client: Configured AnthropicClient instance db_conn_factory: Callable that returns sqlite3.Connection embedding_index: Optional MTGEmbeddingIndex for substitution """ self.anthropic = anthropic_client self.db_conn_factory = db_conn_factory self.embedding_index = embedding_index # ── Step 1: Resolve Commander ──────────────────────────── def resolve_commander(self, commander_name: str) -> dict: """Look up commander on Scryfall to get canonical name, color identity, type.""" try: encoded = commander_name.replace(" ", "+") url = f"https://api.scryfall.com/cards/named?fuzzy={encoded}" req = Request(url) req.add_header("User-Agent", "CommanderAILab/3.0") req.add_header("Accept", "application/json") with urlopen(req, timeout=10) as resp: data = json.loads(resp.read()) return { "name": data.get("name", commander_name), "scryfall_id": data.get("id", ""), "color_identity": data.get("color_identity", []), "type_line": data.get("type_line", ""), "mana_cost": data.get("mana_cost", ""), "image_url": data.get("image_uris", {}).get("normal", ""), "oracle_text": data.get("oracle_text", ""), } except Exception as e: logger.warning("Scryfall lookup failed for '%s': %s", commander_name, e) return { "name": commander_name, "scryfall_id": "", "color_identity": [], "type_line": "", "mana_cost": "", "image_url": "", "oracle_text": "", } # ── Step 2: Build Collection Summary ───────────────────── def build_collection_summary(self, color_identity: List[str], max_per_group: int = 20) -> str: """ Build a text summary of owned cards matching the deck's color identity. Groups by card type for the LLM prompt. """ conn = self.db_conn_factory() try: rows = conn.execute( "SELECT name, type_line, cmc, oracle_text, quantity, tcg_price " "FROM collection_entries WHERE quantity > 0" ).fetchall() except Exception as e: logger.warning("Failed to load collection: %s", e) return "" if not rows: return "" # Filter by color identity ci_set = set(color_identity) if color_identity else None filtered = [] for row in rows: name = row["name"] type_line = row["type_line"] or "" # Basic lands and colorless are always allowed if ci_set: # Simple color check from type_line and name # For proper filtering we'd check mana cost but this is good enough for summary card_colors = set() oracle = row["oracle_text"] or "" mana_cost = "" try: mana_cost = row["mana_cost"] or "" except (IndexError, KeyError): pass for sym in ["W", "U", "B", "R", "G"]: if f"{{{sym}}}" in (mana_cost or oracle): card_colors.add(sym) if card_colors and not card_colors.issubset(ci_set): continue filtered.append(row) if not filtered: return "" # Group by card type groups = {} for row in filtered: type_line = row["type_line"] or "" if "Creature" in type_line: group = "Creatures" elif "Instant" in type_line: group = "Instants" elif "Sorcery" in type_line: group = "Sorceries" elif "Artifact" in type_line: group = "Artifacts" elif "Enchantment" in type_line: group = "Enchantments" elif "Planeswalker" in type_line: group = "Planeswalkers" elif "Land" in type_line: group = "Lands" else: group = "Other" groups.setdefault(group, []).append(row) # Sort each group by popularity/price and build text lines = [f"COLLECTION SUMMARY ({len(filtered)} cards in deck colors):"] for group_name in ["Creatures", "Instants", "Sorceries", "Artifacts", "Enchantments", "Planeswalkers", "Lands", "Other"]: cards = groups.get(group_name, []) if not cards: continue # Sort by price descending (higher value = likely better card) cards.sort(key=lambda r: r["tcg_price"] or 0, reverse=True) card_strs = [] for c in cards[:max_per_group]: price = f"${c['tcg_price']:.2f}" if c["tcg_price"] else "$?" card_strs.append(f"{c['name']} ({price})") lines.append(f" {group_name} ({len(cards)}): {', '.join(card_strs)}") return "\n".join(lines) # ── Step 3: Generate Deck via Anthropic Claude ───────────────── def generate_deck( self, commander_name: str, strategy: str = "", target_bracket: int = 3, budget_usd: float = None, budget_mode: str = "total", omit_cards: List[str] = None, use_collection: bool = True, model: str = None, ) -> dict: """ Generate a complete Commander deck. Returns dict with: - commander: dict with Scryfall data - raw_deck: parsed GeneratedDeckList from Anthropic Claude - cards: list of DeckCardWithStatus (after ownership check) - stats: deck statistics - tokens_used: {prompt, completion} """ # Step 1: Resolve commander cmdr = self.resolve_commander(commander_name) color_identity = cmdr["color_identity"] logger.info("Commander: %s, CI: %s", cmdr["name"], color_identity) # Step 2: Collection summary collection_block = "" if use_collection: collection_block = self.build_collection_summary(color_identity) logger.info("Collection summary: %d chars", len(collection_block) if collection_block else 0) # Step 3: Build prompts user_prompt = build_user_prompt( commander=cmdr["name"], commander_type=cmdr["type_line"], color_identity=color_identity, strategy=strategy, target_bracket=target_bracket, budget_usd=budget_usd, budget_mode=budget_mode, omit_cards=omit_cards, collection_summary=collection_block, ) # Step 4: Call Anthropic Claude with structured output use_model = model or DECK_GEN_MODEL logger.info("Calling Anthropic Claude (%s) for deck generation...", use_model) response = self.anthropic.chat_structured( system_prompt=SYSTEM_PROMPT, user_prompt=user_prompt, json_schema=DECK_LIST_SCHEMA, schema_name="DeckList", model=use_model, temperature=DECK_GEN_TEMPERATURE, max_tokens=DECK_GEN_MAX_TOKENS, ) if not response.ok: raise ValueError( f"Anthropic Claude returned invalid response: {response.content[:300]}" ) raw_deck = response.parsed_json card_count = len(raw_deck.get("cards", [])) truncated = card_count < 90 # Commander decks should have ~100 cards logger.info("Deck generated: %d cards, bracket %d%s", card_count, (raw_deck.get("bracket", {}).get("level", 0) if isinstance(raw_deck.get("bracket"), dict) else raw_deck.get("bracket", 0)), " (TRUNCATED — response may have hit token limit)" if truncated else "") # Step 5: Check ownership and build enriched card list cards_with_status = self._check_ownership(raw_deck.get("cards", [])) # Step 5b: Fill basic lands to hit exactly 100 cards cards_with_status = self._fill_basic_lands(cards_with_status, color_identity) # Step 5c: Secondary safety-net deduplication seen = {} deduped = [] for c in cards_with_status: key = c.name.lower() if key in seen: deduped[seen[key]].count += c.count logger.warning("Secondary dedupe caught duplicate: '%s'", c.name) else: seen[key] = len(deduped) deduped.append(c) cards_with_status = deduped # Step 6: Compute stats stats = self._compute_stats(cards_with_status) return { "commander": cmdr, "color_identity": color_identity, "raw_deck": raw_deck, "cards": [c.model_dump() for c in cards_with_status], "stats": stats, "strategy_summary": raw_deck.get("strategy_summary", ""), "bracket": raw_deck.get("bracket", {}), "archetype": raw_deck.get("archetype", ""), "reasoning": raw_deck.get("reasoning", {}), "estimated_total_usd": raw_deck.get("estimated_total_usd", 0), "tokens_used": { "prompt": response.prompt_tokens, "completion": response.completion_tokens, }, "model": response.model, "citations": response.citations, "truncated": truncated, } # ── Step 5: Ownership Check ────────────────────────────── def _check_ownership(self, cards: List[dict]) -> List[DeckCardWithStatus]: """ Cross-reference generated cards against the collection DB. Sets owned/scryfall_id/owned_qty and real price. Deduplicates cards by name (case-insensitive) — if the LLM returns the same card twice, the counts are merged. """ conn = self.db_conn_factory() enriched = [] seen_names: dict[str, int] = {} # name_lower → index in enriched for card in cards: name = card.get("name", "") if not name: continue name_lower = name.lower() if name_lower in seen_names: idx = seen_names[name_lower] enriched[idx].count += card.get("count", 1) logger.warning("Duplicate card '%s' merged (LLM returned it twice)", name) continue # Look up in collection row = conn.execute( "SELECT name, scryfall_id, quantity, tcg_price, type_line, cmc " "FROM collection_entries WHERE name = ? COLLATE NOCASE LIMIT 1", (name,) ).fetchone() is_owned = bool(row and row["quantity"] and row["quantity"] > 0) enriched_card = DeckCardWithStatus( name=row["name"] if row else name, count=card.get("count", 1), category=card.get("category", ""), role_tags=card.get("role_tags", []), reason=card.get("reason", ""), estimated_price_usd=( round(row["tcg_price"], 2) if row and row["tcg_price"] else card.get("estimated_price_usd", 0) ), synergy_with=card.get("synergy_with", []), owned=is_owned, owned_qty=int(row["quantity"]) if row and row["quantity"] else 0, scryfall_id=row["scryfall_id"] if row else "", status="owned" if is_owned else "missing", ) seen_names[name_lower] = len(enriched) enriched.append(enriched_card) return enriched # ── Step 5b: Basic Land Fill ────────────────────────────── _COLOR_TO_BASIC = { 'W': 'Plains', 'U': 'Island', 'B': 'Swamp', 'R': 'Mountain', 'G': 'Forest', } _BASIC_NAMES = set(_COLOR_TO_BASIC.values()) | {'Wastes'} def _fill_basic_lands( self, cards: List[DeckCardWithStatus], color_identity: List[str], target_total: int = 100, ) -> List[DeckCardWithStatus]: """ Strip LLM-provided basics, then add the right number evenly across the commander's colors to reach exactly target_total. """ non_basics = [c for c in cards if c.name not in self._BASIC_NAMES] current_total = sum(c.count for c in non_basics) basics_needed = max(0, target_total - current_total) if basics_needed == 0: return non_basics ci = [c.upper() for c in color_identity] basic_names = [self._COLOR_TO_BASIC[c] for c in ci if c in self._COLOR_TO_BASIC] if not basic_names: basic_names = ['Wastes'] per_color = basics_needed // len(basic_names) remainder = basics_needed % len(basic_names) for i, bname in enumerate(basic_names): qty = per_color + (1 if i < remainder else 0) if qty > 0: non_basics.append(DeckCardWithStatus( name=bname, count=qty, category='Land', role_tags=[], reason='Basic land for mana fixing', estimated_price_usd=0.10, owned=True, owned_qty=qty, status='owned', )) logger.info("Basic land fill: %d basics added across %d color(s)", basics_needed, len(basic_names)) return non_basics # ── Step 6: Smart Substitution ─────────────────────────── def run_substitution( self, cards: List[DeckCardWithStatus], commander: dict, strategy: str = "", ) -> SubstitutionBatchResult: """ Run the Smart Substitution Engine on missing cards. Pipeline: a) Identify missing cards b) For each: try embedding similarity (free, fast) c) For low-confidence matches: batch Anthropic Claude fallback d) Auto-select best substitute per card Returns SubstitutionBatchResult with all cards updated. """ color_identity = commander.get("color_identity", []) missing = [c for c in cards if not c.owned] owned = [c for c in cards if c.owned] if not missing: return SubstitutionBatchResult( total_cards=len(cards), owned_count=len(owned), substituted_count=0, missing_count=0, cards=cards, ) logger.info("Substitution: %d missing / %d owned", len(missing), len(owned)) # Build list of owned card names for substitution pool conn = self.db_conn_factory() owned_pool_rows = conn.execute( "SELECT DISTINCT name FROM collection_entries WHERE quantity > 0" ).fetchall() owned_pool = [r["name"] for r in owned_pool_rows] # Exclude cards already in the deck deck_names = set(c.name.lower() for c in cards) available_pool = [n for n in owned_pool if n.lower() not in deck_names] # a) Embedding pass needs_anthropic_fallback = [] for card in missing: embedding_alts = self._embedding_substitution( card, color_identity, available_pool ) card.alternatives = embedding_alts # Check if best match meets threshold if embedding_alts and embedding_alts[0].similarity_score >= SUBSTITUTION_MIN_SIMILARITY: card.selected_substitute = embedding_alts[0].name card.status = "substituted" # Remove selected from pool if card.selected_substitute.lower() in [n.lower() for n in available_pool]: available_pool = [n for n in available_pool if n.lower() != card.selected_substitute.lower()] else: needs_anthropic_fallback.append(card) # b) Anthropic Claude fallback for low-confidence matches if needs_anthropic_fallback and SUBSTITUTION_USE_ANTHROPIC_FALLBACK: self._anthropic_substitution( needs_anthropic_fallback, available_pool, commander, strategy ) # Count results substituted = sum(1 for c in cards if c.status == "substituted") still_missing = sum(1 for c in cards if c.status == "missing") return SubstitutionBatchResult( total_cards=len(cards), owned_count=len(owned), substituted_count=substituted, missing_count=still_missing, cards=cards, ) def _embedding_substitution( self, card: DeckCardWithStatus, color_identity: List[str], available_pool: List[str], ) -> List[CardAlternative]: """Find embedding-similar owned cards as substitutes.""" if not self.embedding_index or not self.embedding_index.loaded: return [] # Use the embeddings search to find similar cards matches = self.embedding_index.search_similar( query_card=card.name, color_filter=color_identity if color_identity else None, type_filter=card.category if card.category else None, exclude_cards=[card.name], top_n=SUBSTITUTION_MAX_ALTERNATIVES * 3, # Over-fetch, then filter to owned ) # Filter to only cards in the available pool pool_lower = set(n.lower() for n in available_pool) alternatives = [] for match in matches: if match.name.lower() in pool_lower: # Compute role overlap match_roles = [] # Would need card's role_tags from DB role_overlap = list(set(card.role_tags) & set(match_roles)) if match_roles else [] alternatives.append(CardAlternative( name=match.name, similarity_score=match.similarity, reason=f"Similar to {card.name} (cosine sim: {match.similarity:.3f})", source="embedding", role_overlap=role_overlap, cmc_delta=match.mana_value - (card.estimated_price_usd or 0), )) if len(alternatives) >= SUBSTITUTION_MAX_ALTERNATIVES: break return alternatives def _anthropic_substitution( self, missing_cards: List[DeckCardWithStatus], available_pool: List[str], commander: dict, strategy: str, ): """ Batch Anthropic Claude fallback for cards with low embedding confidence. Mutates the cards in-place with substitution data. """ if not available_pool: return # Build the request missing_dicts = [ { "name": c.name, "role_tags": c.role_tags, "category": c.category, "reason": c.reason, } for c in missing_cards ] user_prompt = build_substitution_prompt( missing_cards=missing_dicts, allowed_cards=available_pool[:200], # Limit to avoid token overflow commander=commander.get("name", ""), color_identity=commander.get("color_identity", []), strategy=strategy, ) try: from ..schemas.substitution_schema import SUBSTITUTION_RESPONSE_SCHEMA response = self.anthropic.chat_structured( system_prompt=SUBSTITUTION_SYSTEM, user_prompt=user_prompt, json_schema=SUBSTITUTION_RESPONSE_SCHEMA, schema_name="SubstitutionResponse", model=SUBSTITUTION_MODEL, temperature=0.1, max_tokens=4096, ) if not response.ok: logger.warning("Anthropic Claude substitution fallback failed — non-JSON response") return subs = response.parsed_json.get("substitutions", []) logger.info("Anthropic Claude suggested %d substitutions", len(subs)) # Apply substitutions card_by_name = {c.name.lower(): c for c in missing_cards} for sub in subs: orig = sub.get("original", "").lower() substitute = sub.get("substitute", "") reason = sub.get("reason", "") if orig in card_by_name and substitute: card = card_by_name[orig] # Add as top alternative anthropic_alt = CardAlternative( name=substitute, similarity_score=0.80, # Nominal score for Anthropic Claude suggestions reason=reason, source="Anthropic Claude", role_overlap=sub.get("role_overlap", []), ) # Insert at front card.alternatives.insert(0, anthropic_alt) card.selected_substitute = substitute card.status = "substituted" except Exception as e: logger.error("Anthropic Claude substitution fallback error: %s", e) # ── Stats ──────────────────────────────────────────────── @staticmethod def _compute_stats(cards: List[DeckCardWithStatus]) -> dict: """Compute deck statistics from the enriched card list.""" by_category = {} by_status = {"owned": 0, "substituted": 0, "missing": 0} total_price = 0.0 owned_price = 0.0 missing_price = 0.0 role_counts = {} for card in cards: cat = card.category or "Other" by_category[cat] = by_category.get(cat, 0) + card.count by_status[card.status] = by_status.get(card.status, 0) + 1 price = card.estimated_price_usd * card.count total_price += price if card.owned: owned_price += price else: missing_price += price for role in card.role_tags: role_counts[role] = role_counts.get(role, 0) + 1 land_count = by_category.get("Land", 0) nonland_count = sum(v for k, v in by_category.items() if k != "Land") return { "total_cards": sum(c.count for c in cards), "land_count": land_count, "nonland_count": nonland_count, "by_category": by_category, "by_status": by_status, "role_counts": role_counts, "total_price_usd": round(total_price, 2), "owned_price_usd": round(owned_price, 2), "missing_price_usd": round(missing_price, 2), }