""" FastAPI REST API server for Pokemon Card Authentication. This server wraps the DL prediction pipeline (ResNet50 + EfficientNet-B7) to provide a clean REST interface for the frontend application. """ import hashlib import os import sys from contextlib import asynccontextmanager from pathlib import Path from threading import Lock, Thread from typing import Any, Dict, List, Optional from urllib.parse import urlparse from urllib.request import Request, urlopen BASE_DIR = Path(__file__).resolve().parent # Load .env from the Backend directory (ignored by git; overrides nothing already set) try: from dotenv import load_dotenv load_dotenv(BASE_DIR / ".env", override=False) except ImportError: pass def _find_model_package_root() -> Path: candidates = [ BASE_DIR.parent / "Model", # Code/Model in monorepo BASE_DIR.parent.parent / "Code" / "Model", # Repo root /Code/Model BASE_DIR, # If src/ is vendored into Code/Backend/ ] for candidate in candidates: if (candidate / "src" / "dl" / "prediction_pipeline.py").is_file(): return candidate raise RuntimeError( "Could not locate model source package root containing " "'src/dl/prediction_pipeline.py'. For Railway, prefer deploying " "with Root Directory set to the repository root so `Code/Model/` is " "included; alternatively vendor `Code/Model/src` into `Code/Backend/src`." ) def _find_models_dir(model_package_root: Path) -> Path: candidates = [ model_package_root / "data" / "models", # Monorepo Model directory BASE_DIR / "data" / "models", # Vendored into Backend for deployment ] for candidate in candidates: if candidate.is_dir(): return candidate raise RuntimeError( "Could not locate trained models directory. Ensure " "model files are present in the deploy build context, or vendor " "them into `Code/Backend/data/models`. Looked for: " f"{candidates}" ) def _discover_local_checkpoint(dl_models_dir: Path) -> Optional[Path]: """ Discover a local checkpoint in preferred order. Priority: 1) *_best.pth 2) *_final.pth 3) any *.pth """ if not dl_models_dir.exists(): return None for pattern in ("*_best.pth", "*_final.pth", "*.pth"): candidates = sorted(dl_models_dir.glob(pattern)) if candidates: return candidates[-1] return None def _compute_sha256(file_path: Path) -> str: """Compute SHA256 hash for file integrity checks.""" sha256 = hashlib.sha256() with open(file_path, "rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): sha256.update(chunk) return sha256.hexdigest() def _resolve_model_filename(download_url: str, filename_override: Optional[str]) -> str: """Resolve destination filename from override or URL path.""" if filename_override: candidate = Path(filename_override).name if candidate: return candidate candidate = Path(urlparse(download_url).path).name if candidate: return candidate return "downloaded_model_best.pth" def _download_file(download_url: str, destination: Path, bearer_token: Optional[str] = None, timeout_seconds: int = 120) -> None: """Download file from URL to destination path.""" headers = {} if bearer_token: headers["Authorization"] = f"Bearer {bearer_token}" destination.parent.mkdir(parents=True, exist_ok=True) tmp_destination = destination.with_suffix(destination.suffix + ".tmp") request = Request(download_url, headers=headers) with urlopen(request, timeout=timeout_seconds) as response, open(tmp_destination, "wb") as out_file: while True: chunk = response.read(1024 * 1024) if not chunk: break out_file.write(chunk) tmp_destination.replace(destination) def _download_checkpoint_from_env(dl_models_dir: Path) -> Optional[Path]: """ Download checkpoint when DL_MODEL_URL is configured. Optional env vars: - DL_MODEL_FILENAME: override downloaded filename - DL_MODEL_SHA256: expected checksum (lowercase hex) - DL_MODEL_BEARER_TOKEN: bearer token for private URLs """ download_url = os.getenv("DL_MODEL_URL", "").strip() if not download_url: return None filename_override = os.getenv("DL_MODEL_FILENAME", "").strip() or None expected_sha256 = os.getenv("DL_MODEL_SHA256", "").strip().lower() or None bearer_token = os.getenv("DL_MODEL_BEARER_TOKEN", "").strip() or None filename = _resolve_model_filename(download_url, filename_override) destination = dl_models_dir / filename if destination.exists() and expected_sha256: existing_hash = _compute_sha256(destination).lower() if existing_hash != expected_sha256: print(f"⚠️ Existing checkpoint hash mismatch, re-downloading: {destination.name}") destination.unlink() if not destination.exists(): print(f"Downloading DL checkpoint from DL_MODEL_URL to {destination}") _download_file(download_url, destination, bearer_token=bearer_token) if expected_sha256: actual_sha256 = _compute_sha256(destination).lower() if actual_sha256 != expected_sha256: try: destination.unlink() except OSError: pass raise RuntimeError( f"Downloaded checkpoint hash mismatch for {destination.name}. " f"Expected {expected_sha256}, got {actual_sha256}" ) return destination def _should_load_model_on_startup() -> bool: """ Decide whether to eagerly load the DL model during startup. Env var: - DL_LOAD_ON_STARTUP=true|false (default: true) """ raw = os.getenv("DL_LOAD_ON_STARTUP", "").strip().lower() if raw in ("", "1", "true", "yes", "on"): return True if raw in ("0", "false", "no", "off"): return False print(f"⚠️ Invalid DL_LOAD_ON_STARTUP value '{raw}', defaulting to eager loading.") return True MODEL_PACKAGE_ROOT = _find_model_package_root() MODELS_DIR = _find_models_dir(MODEL_PACKAGE_ROOT) # Add model package root to path for importing `src.*` modules sys.path.insert(0, str(MODEL_PACKAGE_ROOT)) from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field import numpy as np import cv2 import base64 import json import time from src.dl.prediction_pipeline import create_dl_pipeline from src.preprocessing.card_detector import detect_card_boundary_strict # Import validators import sys backend_src_path = str(BASE_DIR / "src") if backend_src_path not in sys.path: sys.path.insert(0, backend_src_path) from validators.feature_based_validator import FeatureBasedValidator from validators.multilayer_validation import run_multilayer_validation # Stage 5: OCR + enrichment (optional; imported lazily to keep startup fast) try: from ocr.card_ocr import CardOCR from enrichment.tcg_lookup import TCGLookup _ENRICHMENT_AVAILABLE = True except ImportError: _ENRICHMENT_AVAILABLE = False @asynccontextmanager async def app_lifespan(_: FastAPI): """Run startup initialization via FastAPI lifespan to avoid deprecated startup events.""" await startup_event() yield # Initialize FastAPI app app = FastAPI( title="Pokemon Card Authentication API", description="AI-powered Pokemon card authentication using ResNet50 + EfficientNet-B7", version="2.0.0", lifespan=app_lifespan, ) # CORS middleware for frontend connectivity app.add_middleware( CORSMiddleware, allow_origins=[ "http://localhost:3000", "http://127.0.0.1:3000", "https://pokemonauthenticator.com", "https://www.pokemonauthenticator.com", ], allow_origin_regex=r"^https://.*\.(vercel\.app|vercel\.com)$", allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Global DL pipeline instance dl_pipeline = None model_load_error = None model_version_info = None # DL model version metadata model_filename = None # DL model filename model_registry = None # Cached model registry metadata model_load_lock = Lock() model_load_mode = "eager" # "eager" (default) or "lazy" model_loading = False # True while a model load is in progress # Global validators (validation layers kept from earlier pipeline revisions) feature_validator = None def _load_version_registry(registry_path: Path) -> Optional[dict]: """Load version registry if it exists.""" if not registry_path.exists(): print(f"⚠️ Version registry not found: {registry_path}") return None try: with open(registry_path, 'r') as f: registry = json.load(f) print(f"✅ Loaded version registry (schema v{registry.get('schema_version')})") return registry except Exception as e: print(f"⚠️ Failed to load version registry: {e}") return None def _get_model_version_info(registry: Optional[dict], model_filename: str) -> Optional[Dict[str, Any]]: """Extract version info for a specific model from registry.""" if registry is None: return None try: # Prefer exact filename match across all model types. for model_entries in registry.get('models', {}).values(): for model_entry in model_entries: if model_entry.get('filename') == model_filename: return model_entry # Fallback: Extract version token (YYYYMMDD_HHMMSS) from filename. stem = Path(model_filename).stem import re match = re.search(r"(\d{8}_\d{6})", stem) if not match: return None version = match.group(1) for model_entries in registry.get('models', {}).values(): for model_entry in model_entries: if model_entry.get('version') == version: return model_entry except Exception as e: print(f"⚠️ Failed to extract version info: {e}") return None return None def _initialize_feature_validator() -> None: """Initialize pre-DL validation layers.""" global feature_validator if feature_validator is not None: return print("\n" + "=" * 80) print("Initializing validators...") print("=" * 80) try: feature_validator = FeatureBasedValidator(confidence_threshold=0.75) print("✅ Pokemon card validators loaded (color-based back validation)") except Exception as e: print(f"⚠️ Failed to load validators: {e}") import traceback traceback.print_exc() feature_validator = None def _load_dl_pipeline_on_demand(): """Lazy-load DL pipeline on first authenticate request.""" global dl_pipeline, model_load_error, model_version_info, model_filename, model_registry, model_loading if dl_pipeline is not None: return dl_pipeline # Avoid repeated expensive retries when the model has already failed to load. if model_load_error: return None with model_load_lock: if dl_pipeline is not None: return dl_pipeline if model_load_error: return None model_loading = True try: print("\n" + "=" * 80) print("Loading DL model...") print("=" * 80) if model_registry is None: registry_path = MODELS_DIR / "version_registry.json" model_registry = _load_version_registry(registry_path) dl_models_dir = MODELS_DIR / "dl" dl_model_path = None download_error = None # Find local DL checkpoint first discovered_checkpoint = _discover_local_checkpoint(dl_models_dir) if discovered_checkpoint is not None: dl_model_path = str(discovered_checkpoint) model_filename = discovered_checkpoint.name print(f"Loading local DL model: {model_filename}") else: try: downloaded_checkpoint = _download_checkpoint_from_env(dl_models_dir) if downloaded_checkpoint is not None: dl_model_path = str(downloaded_checkpoint) model_filename = downloaded_checkpoint.name print(f"Loading downloaded DL model: {model_filename}") except Exception as e: download_error = str(e) print(f"⚠️ DL checkpoint download failed: {download_error}") if dl_model_path: try: dl_pipeline = create_dl_pipeline( model_path=dl_model_path, preprocessing_config={"target_size": 256}, ) print(f"✅ DL pipeline loaded: {model_filename}") # Extract version info version_entry = _get_model_version_info(model_registry, model_filename) if version_entry: model_version_info = version_entry print(f"✅ DL Model version: {version_entry.get('version')} ({version_entry.get('status')})") except Exception as e: import traceback print(f"⚠️ DL model failed to load: {e}") traceback.print_exc() dl_pipeline = None if dl_pipeline is None: error_msg_parts = [f"No DL model found in {MODELS_DIR / 'dl'}."] if download_error: error_msg_parts.append(f"DL_MODEL_URL bootstrap failed: {download_error}.") error_msg_parts.append( "Provide a checkpoint in that directory, or set DL_MODEL_URL " "(optional: DL_MODEL_FILENAME, DL_MODEL_SHA256, DL_MODEL_BEARER_TOKEN)." ) error_msg_parts.append("Train locally with: cd ../Model && python -m src.dl.train_dl") model_load_error = " ".join(error_msg_parts) print(f"❌ {model_load_error}") return None model_load_error = None print("=" * 80) return dl_pipeline except Exception as e: import traceback print(f"⚠️ Unexpected DL model load failure: {e}") traceback.print_exc() dl_pipeline = None model_load_error = f"Unexpected DL model load failure: {e}" return None finally: model_loading = False def _start_background_model_load_if_needed() -> bool: """ Trigger model loading in a daemon thread. Returns: True if a new background load was started, False otherwise. """ global model_loading, model_load_error with model_load_lock: if dl_pipeline is not None or model_load_error or model_loading: return False model_loading = True def _run_loader(): global model_loading try: _load_dl_pipeline_on_demand() finally: model_loading = False try: Thread(target=_run_loader, daemon=True, name="dl-model-loader").start() return True except Exception as e: model_loading = False model_load_error = f"Failed to start background model load: {e}" print(f"⚠️ {model_load_error}") return False async def startup_event(): """Initialize lightweight components; defer DL model to first request.""" global model_registry, model_load_mode load_on_startup = _should_load_model_on_startup() model_load_mode = "eager" if load_on_startup else "lazy" print("=" * 80) print(f"Starting API (DL model load mode: {model_load_mode})...") print(f"Models directory: {MODELS_DIR}") print(f"Models directory exists: {MODELS_DIR.exists()}") # Load version registry registry_path = MODELS_DIR / "version_registry.json" model_registry = _load_version_registry(registry_path) # Initialize Pokemon card validators (validation layers unchanged) _initialize_feature_validator() if load_on_startup: print("Eager mode enabled: loading DL model during startup.") _load_dl_pipeline_on_demand() else: print("Lazy mode enabled: DL model initialization deferred to first /api/authenticate request.") # Warm up card enrichment index in background (avoids 30s delay on first /api/card-info call) if _ENRICHMENT_AVAILABLE: def _warmup_enrichment(): try: TCGLookup() # triggers _GitHubCardIndex._get_index() — loads cache or builds from GitHub print("Card enrichment index ready.") except Exception as exc: print(f"⚠️ Card enrichment warm-up failed (non-fatal): {exc}") Thread(target=_warmup_enrichment, daemon=True).start() print("Card enrichment warm-up started in background.") print("=" * 80) # Pydantic models for request/response validation class AuthenticateRequest(BaseModel): """Request body for card authentication.""" front_image: str = Field(..., description="Base64 encoded front image") back_image: str = Field(..., description="Base64 encoded back image") class CardDetectRequest(BaseModel): """Request body for card edge detection.""" image: str = Field(..., description="Base64 encoded image") class CardDetectResponse(BaseModel): """Response body for card edge detection.""" card_detected: bool = Field(..., description="True if card edges are detected") class PredictionResult(BaseModel): """Individual image prediction result.""" prediction: int = Field(..., description="-1=no_card, 0=counterfeit, 1=authentic") label: str = Field(..., description="'authentic', 'counterfeit', or 'no_card'") confidence: float = Field(..., ge=0, le=1, description="Confidence score") probabilities: Dict[str, float] = Field(..., description="Class probabilities") inference_time_ms: float = Field(..., description="Inference time in milliseconds") component_scores: Optional[Dict[str, float]] = Field(None, description="Per-head DL scores") class QualityCheckResult(BaseModel): """Image quality check result.""" blur_score: float = Field(..., description="Laplacian variance (higher = sharper)") brightness: float = Field(..., description="Mean pixel value (0-255)") contrast: float = Field(..., description="Std deviation of pixels") is_acceptable: bool = Field(..., description="Whether image passes quality checks") class PokemonBackValidation(BaseModel): """Pokemon back color validation result.""" passed: bool = Field(..., description="Whether back image passes Pokemon back validation") confidence: float = Field(..., ge=0, le=1, description="Confidence score for validation") reason: str = Field(..., description="Validation failure/success reason") class ModelVersionInfo(BaseModel): """Model version and training metadata.""" version: str = Field(..., description="Model version (timestamp)") model_type: str = Field(..., description="Model type (dl_multihead)") model_class: str = Field(default="", description="Python class name") training_date: str = Field(default="", description="ISO timestamp of training") status: str = Field(..., description="Deployment status (production, staging, training)") accuracy: Optional[float] = Field(None, description="Test accuracy") f1_score: Optional[float] = Field(None, description="Test F1 score") roc_auc: Optional[float] = Field(None, description="Test ROC AUC") dataset_size: Optional[int] = Field(None, description="Number of training samples") n_features: Optional[Any] = Field(None, description="Number of features or 'end-to-end'") pipeline_type: Optional[str] = Field(None, description="Pipeline type: 'dl'") backbone: Optional[str] = Field(None, description="DL backbone architecture") class RejectionReason(BaseModel): """Detailed information about why a card was rejected as 'no_card'.""" category: str = Field(..., description="Rejection category: 'geometry', 'back_pattern', 'front_is_back', 'mismatch'") message: str = Field(..., description="User-friendly error message") details: Dict[str, Any] = Field(default_factory=dict, description="Technical details for debugging") class CardInfoRequest(BaseModel): """Request body for card enrichment (OCR + TCG lookup).""" front_image: str = Field(..., description="Base64 encoded front image (data URI or raw base64)") class CardInfoResponse(BaseModel): """Response body for card identity enrichment.""" found: bool = Field(..., description="True if card identity was successfully resolved") name: Optional[str] = Field(None, description="Card name from OCR / API") set_name: Optional[str] = Field(None, description="Set name (e.g. 'Base Set')") set_code: Optional[str] = Field(None, description="Set code (e.g. 'base1')") collector_number: Optional[str] = Field(None, description="Collector number (e.g. '58')") rarity: Optional[str] = Field(None, description="Rarity string (e.g. 'Rare Holo')") hp: Optional[str] = Field(None, description="HP value (e.g. '120')") types: Optional[List[str]] = Field(None, description="Card types (e.g. ['Fire'])") market_price: Optional[Dict] = Field(None, description="Price tiers: {normal, holofoil, reverse_holofoil, currency}") tcg_image_url: Optional[str] = Field(None, description="Small card image URL from pokemontcg.io") tcg_card_url: Optional[str] = Field(None, description="TCGPlayer card page URL") lookup_confidence: float = Field(0.0, ge=0, le=1, description="Confidence of the TCG lookup match") ocr_raw: Optional[str] = Field(None, description="Raw OCR text for debugging") error: Optional[str] = Field(None, description="Error code if lookup failed") class AuthenticateResponse(BaseModel): """Response body for card authentication.""" is_authentic: bool = Field(..., description="Final authentication result") confidence: float = Field(..., ge=0, le=1, description="Overall confidence") label: str = Field(..., description="'authentic', 'counterfeit', or 'no_card'") probabilities: Dict[str, float] = Field(..., description="Average probabilities") front_analysis: PredictionResult = Field(..., description="Front card analysis") back_analysis: PredictionResult = Field(..., description="Back card analysis") processing_time_ms: float = Field(..., description="Total processing time") quality_checks: Dict[str, QualityCheckResult] = Field(..., description="Quality checks for both images") pokemon_back_validation: Optional[PokemonBackValidation] = Field(None, description="Pokemon back validation result (if performed)") model_version: Optional[ModelVersionInfo] = Field(None, description="DL model version information") rejection_reason: Optional[RejectionReason] = Field(None, description="Detailed rejection reason (if label='no_card')") processed_sides: Optional[List[str]] = Field(None, description="Side(s) that passed validation and were processed by DL inference") @app.get("/") async def root(): """Root endpoint.""" return { "message": "Pokemon Card Authentication API", "version": "2.0.0", "status": "running", "endpoints": { "health": "/api/health", "warmup": "/api/warmup", "card_detect": "/api/card-detect", "authenticate": "/api/authenticate", "card_info": "/api/card-info", "docs": "/docs" } } @app.get("/api/health") async def health_check(): """Health check endpoint to verify API and model status.""" if dl_pipeline is None: response = { "status": "degraded" if model_load_error else "ok", "model_loaded": False, "model_loading": model_loading, "model_load_mode": model_load_mode, "api_version": "2.0.0", "error": model_load_error, "models_dir": str(MODELS_DIR), "models_dir_exists": MODELS_DIR.exists(), } if model_version_info: info = dict(model_version_info) if 'trained_at' in info and not info.get('training_date'): info['training_date'] = info['trained_at'] response["model_version"] = ModelVersionInfo(**info).model_dump() return response response = { "status": "ok", "model_loaded": True, "model_loading": model_loading, "model_load_mode": model_load_mode, "api_version": "2.0.0", "model_name": model_filename or "dl_model", } # Add version info if available if model_version_info: info = dict(model_version_info) if 'trained_at' in info and not info.get('training_date'): info['training_date'] = info['trained_at'] response["model_version"] = ModelVersionInfo(**info).model_dump() return response @app.post("/api/warmup") async def warmup_model(): """Trigger asynchronous DL model loading.""" if dl_pipeline is not None: return { "status": "ready", "model_loaded": True, "model_loading": False, "model_load_mode": model_load_mode, } if model_load_error: return { "status": "error", "model_loaded": False, "model_loading": False, "model_load_mode": model_load_mode, "error": model_load_error, } started = _start_background_model_load_if_needed() return { "status": "warming" if (started or model_loading) else "pending", "model_loaded": False, "model_loading": True if (started or model_loading) else False, "model_load_mode": model_load_mode, } @app.post("/api/card-detect", response_model=CardDetectResponse) async def card_detect(request: CardDetectRequest): """ Detect card edges in a single image. Args: request: Contains base64-encoded image Returns: Card detection result """ img = decode_base64_image(request.image) if img is None: raise HTTPException(status_code=400, detail="Failed to decode image") corners = detect_card_boundary_strict( img, min_area_ratio=0.001, max_area_ratio=0.999, aspect_ratio_range=(0.30, 1.0), solidity_threshold=0.60, fill_ratio_threshold=0.40, ) return CardDetectResponse(card_detected=corners is not None) @app.post("/api/authenticate", response_model=AuthenticateResponse) async def authenticate_card(request: AuthenticateRequest): """ Authenticate a Pokemon card using front and back images. Args: request: Contains base64-encoded front and back images Returns: Authentication result with confidence scores and quality checks Raises: HTTPException: If model not loaded or processing fails """ if dl_pipeline is None and model_load_mode == "lazy": if model_loading or _start_background_model_load_if_needed(): raise HTTPException( status_code=503, detail=( "DL model warm-up in progress. Retry in 20-60 seconds. " "You can poll /api/health (model_loaded/model_loading) or call /api/warmup." ), ) pipeline = _load_dl_pipeline_on_demand() if pipeline is None: raise HTTPException( status_code=503, detail=model_load_error or ( "No DL model loaded. Add checkpoint to Code/Model/data/models/dl " "or set DL_MODEL_URL, then restart backend." ), ) print("Using DL pipeline for authentication") start_time = time.time() try: # Decode base64 images front_img = decode_base64_image(request.front_image) back_img = decode_base64_image(request.back_image) # Validate images if front_img is None: raise HTTPException(status_code=400, detail="Failed to decode front image") if back_img is None: raise HTTPException(status_code=400, detail="Failed to decode back image") def _no_card_result() -> Dict[str, Any]: return { "prediction": -1, "label": "no_card", "confidence": 0.0, "probabilities": {"authentic": 0.0, "counterfeit": 0.0}, "inference_time_ms": 0.0, } validation = run_multilayer_validation( front_image=front_img, back_image=back_img, feature_validator=feature_validator, require_both_sides=False, ) def _to_pokemon_back_validation() -> Optional[PokemonBackValidation]: if ( validation.pokemon_back_validation is not None and not validation.pokemon_back_validation.passed ): return PokemonBackValidation( passed=False, confidence=validation.pokemon_back_validation.confidence, reason=validation.pokemon_back_validation.reason, ) if ( validation.front_not_back_validation is not None and not validation.front_not_back_validation.passed ): return PokemonBackValidation( passed=False, confidence=validation.front_not_back_validation.confidence, reason=( "Front image appears to be a card back: " f"{validation.front_not_back_validation.reason}" ), ) return None front_result = _no_card_result() back_result = _no_card_result() if "front" in validation.processed_sides: front_result = pipeline.predict(front_img, is_back=False) if "back" in validation.processed_sides: back_result = pipeline.predict(back_img, is_back=True) pokemon_back_validation = _to_pokemon_back_validation() processing_time_ms = (time.time() - start_time) * 1000 processed_sides = validation.processed_sides or None if validation.rejected: response_data: Dict[str, Any] = { "is_authentic": False, "confidence": 0.0, "label": "no_card", "probabilities": {"authentic": 0.0, "counterfeit": 0.0}, "front_analysis": PredictionResult(**front_result), "back_analysis": PredictionResult(**back_result), "processing_time_ms": processing_time_ms, "quality_checks": { "front": QualityCheckResult(**validation.front.quality), "back": QualityCheckResult(**validation.back.quality), }, "rejection_reason": RejectionReason( category=validation.rejection_category, message=validation.rejection_message, details=validation.rejection_details, ), "processed_sides": processed_sides, } if pokemon_back_validation is not None: response_data["pokemon_back_validation"] = pokemon_back_validation return AuthenticateResponse(**response_data) valid_authentic_probs: List[float] = [] if front_result.get("label") != "no_card": valid_authentic_probs.append(float(front_result["probabilities"]["authentic"])) if back_result.get("label") != "no_card": valid_authentic_probs.append(float(back_result["probabilities"]["authentic"])) if not valid_authentic_probs: response_data = { "is_authentic": False, "confidence": 0.0, "label": "no_card", "probabilities": {"authentic": 0.0, "counterfeit": 0.0}, "front_analysis": PredictionResult(**front_result), "back_analysis": PredictionResult(**back_result), "processing_time_ms": processing_time_ms, "quality_checks": { "front": QualityCheckResult(**validation.front.quality), "back": QualityCheckResult(**validation.back.quality), }, "rejection_reason": RejectionReason( category="geometry", message="Validated side(s) were classified as non-Pokemon cards", details={"processed_sides": validation.processed_sides}, ), "processed_sides": processed_sides, } if pokemon_back_validation is not None: response_data["pokemon_back_validation"] = pokemon_back_validation return AuthenticateResponse(**response_data) avg_authentic_prob = sum(valid_authentic_probs) / len(valid_authentic_probs) avg_counterfeit_prob = 1.0 - avg_authentic_prob final_label = "authentic" if avg_authentic_prob >= 0.5 else "counterfeit" final_confidence = max(avg_authentic_prob, avg_counterfeit_prob) response_data = { "is_authentic": avg_authentic_prob >= 0.5, "confidence": final_confidence, "label": final_label, "probabilities": { "authentic": avg_authentic_prob, "counterfeit": avg_counterfeit_prob, }, "front_analysis": PredictionResult(**front_result), "back_analysis": PredictionResult(**back_result), "processing_time_ms": processing_time_ms, "quality_checks": { "front": QualityCheckResult(**validation.front.quality), "back": QualityCheckResult(**validation.back.quality), }, "processed_sides": processed_sides, } if pokemon_back_validation is not None: response_data["pokemon_back_validation"] = pokemon_back_validation if model_version_info: info = dict(model_version_info) if "trained_at" in info and not info.get("training_date"): info["training_date"] = info["trained_at"] response_data["model_version"] = ModelVersionInfo(**info) return AuthenticateResponse(**response_data) except HTTPException: raise except Exception as e: raise HTTPException( status_code=500, detail=f"Authentication failed: {str(e)}" ) @app.post("/api/card-info", response_model=CardInfoResponse) async def card_info_endpoint(request: CardInfoRequest) -> CardInfoResponse: """ Enrich a Pokemon card with identity, rarity, and market pricing. Runs OCR on the front image to extract card name and collector number, then queries the pokemontcg.io API (all eras, 1999–present) for pricing and rarity data. Note: The frontend calls this endpoint **only when authentication returns is_authentic=true**. The backend itself does not enforce this gate so that the endpoint remains independently testable. Args: request: Contains base64-encoded front image Returns: CardInfoResponse with found=True and enrichment data on success, or found=False with an error code on failure. """ if not _ENRICHMENT_AVAILABLE: return CardInfoResponse( found=False, lookup_confidence=0.0, error="enrichment_unavailable", ) img = decode_base64_image(request.front_image) if img is None: return CardInfoResponse( found=False, lookup_confidence=0.0, error="invalid_image", ) try: ocr_result = CardOCR().extract(img) except Exception as exc: print(f"⚠️ CardOCR error: {exc}") return CardInfoResponse( found=False, lookup_confidence=0.0, error="ocr_error", ) ocr_name = ocr_result.get("name") ocr_number = ocr_result.get("collector_number") ocr_hp = ocr_result.get("hp") raw_text = ocr_result.get("raw_text") if not ocr_name: print(f"[card-info] ocr_failed — raw_text: {repr(raw_text)}") return CardInfoResponse( found=False, lookup_confidence=0.0, ocr_raw=raw_text, error="ocr_failed", ) try: card = TCGLookup().search(ocr_name, ocr_number, ocr_hp) except Exception as exc: print(f"⚠️ TCGLookup error: {exc}") return CardInfoResponse( found=False, lookup_confidence=0.3, ocr_raw=raw_text, error="lookup_error", ) if card is None: print(f"[card-info] not_found — name={repr(ocr_name)} number={repr(ocr_number)} hp={repr(ocr_hp)}") return CardInfoResponse( found=False, lookup_confidence=0.3, ocr_raw=raw_text, error="not_found", ) return CardInfoResponse( found=True, name=card.name, set_name=card.set_name, set_code=card.set_code, collector_number=card.collector_number, rarity=card.rarity, hp=card.hp, types=card.types or [], market_price=card.market_price.to_dict() if card.market_price else None, tcg_image_url=card.tcg_image_url, tcg_card_url=card.tcg_card_url, lookup_confidence=card.lookup_confidence, ocr_raw=raw_text, ) def decode_base64_image(base64_str: str) -> Optional[np.ndarray]: """ Decode base64 string to OpenCV image (BGR format). Args: base64_str: Base64 encoded image string (with or without data URI prefix) Returns: NumPy array in BGR format, or None if decoding fails """ try: # Remove data URI prefix if present (data:image/jpeg;base64,...) if ',' in base64_str: base64_str = base64_str.split(',')[1] # Decode base64 to bytes img_bytes = base64.b64decode(base64_str) # Convert bytes to numpy array nparr = np.frombuffer(img_bytes, np.uint8) # Decode to OpenCV image img = cv2.imdecode(nparr, cv2.IMREAD_COLOR) return img except Exception as e: print(f"Error decoding base64 image: {e}") return None if __name__ == "__main__": import uvicorn uvicorn.run( app, host="0.0.0.0", port=8000, log_level="info" )