`python-matcher/api/server.py`

244 lines
"""
FastAPI server for handbag image matching
Provides endpoints for visual similarity search using vector embeddings
"""
import os
import sys
import io
import psycopg2
import requests
from typing import Optional
from fastapi import FastAPI, UploadFile, File, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from PIL import Image

# Add parent directory to path
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from models.handbag_embedder import image_to_vec

# Configuration
DB_DSN = os.environ.get("DATABASE_URL", "postgresql://localhost/handbags")
API_PORT = int(os.environ.get("API_PORT", "8000"))

# Initialize FastAPI
app = FastAPI(
    title="Handbag Visual Matching API",
    description="Vector-based image similarity search for handbag identification",
    version="1.0.0"
)

# CORS - lock down in production
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # TODO: Restrict to specific domains
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


def get_conn():
    """Get database connection"""
    return psycopg2.connect(DB_DSN)


class HandbagURLRequest(BaseModel):
    image_url: str = Field(..., description="URL of handbag image to match")
    top_k: int = Field(5, ge=1, le=50, description="Number of top matches to return")
    min_similarity: float = Field(0.0, ge=0.0, le=1.0, description="Minimum similarity score (0-1)")


class HandbagMatch(BaseModel):
    id: int
    sku: str
    brand: str
    model_name: str
    pattern_name: Optional[str]
    colorway: Optional[str]
    material: Optional[str]
    hardware: Optional[str]
    size: Optional[str]
    condition: Optional[str]
    price_usd: Optional[float]
    image_url: str
    similarity: float
    source: Optional[str]


def query_handbags(vec, top_k=5, min_similarity=0.0):
    """
    Query handbags by vector similarity

    Args:
        vec: 768-dim vector as list
        top_k: Number of results
        min_similarity: Minimum similarity threshold (0-1)

    Returns:
        List of matching handbags with similarity scores
    """
    vec_str = "[" + ",".join(str(x) for x in vec) + "]"

    conn = get_conn()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT
                    id, sku, brand, model_name, pattern_name,
                    colorway, material, hardware, size, condition,
                    price_usd, image_url, source,
                    1 - (embedding <-> %s::vector) AS similarity
                FROM handbags
                WHERE embedding IS NOT NULL
                  AND is_active = true
                  AND 1 - (embedding <-> %s::vector) >= %s
                ORDER BY embedding <-> %s::vector
                LIMIT %s;
                """,
                (vec_str, vec_str, min_similarity, vec_str, top_k)
            )

            cols = [d[0] for d in cur.description]
            rows = cur.fetchall()
            return [dict(zip(cols, r)) for r in rows]
    finally:
        conn.close()


@app.get("/")
def root():
    """Health check endpoint"""
    return {
        "service": "Handbag Visual Matching API",
        "version": "1.0.0",
        "status": "healthy"
    }


@app.get("/stats")
def get_stats():
    """Get database statistics"""
    conn = get_conn()
    try:
        with conn.cursor() as cur:
            cur.execute("""
                SELECT
                    COUNT(*) as total_handbags,
                    COUNT(embedding) as embedded_handbags,
                    COUNT(DISTINCT brand) as unique_brands,
                    COUNT(DISTINCT model_name) as unique_models
                FROM handbags
                WHERE is_active = true;
            """)
            row = cur.fetchone()
            return {
                "total_handbags": row[0],
                "embedded_handbags": row[1],
                "unique_brands": row[2],
                "unique_models": row[3],
                "embedding_coverage": f"{(row[1]/row[0]*100):.1f}%" if row[0] > 0 else "0%"
            }
    finally:
        conn.close()


@app.post("/handbags/match/url", response_model=dict)
def match_handbag_by_url(req: HandbagURLRequest):
    """
    Match handbag from image URL

    Returns top K similar handbags from catalog
    """
    try:
        headers = {'User-Agent': 'Mozilla/5.0 (compatible; HandbagMatcher/1.0)'}
        resp = requests.get(req.image_url, timeout=10, headers=headers)
        resp.raise_for_status()
        img = Image.open(io.BytesIO(resp.content)).convert("RGB")
    except Exception as e:
        raise HTTPException(
            status_code=400,
            detail=f"Could not fetch or read image: {str(e)}"
        )

    # Generate embedding
    vec = image_to_vec(img)

    # Query database
    results = query_handbags(vec, top_k=req.top_k, min_similarity=req.min_similarity)

    return {
        "query_image": req.image_url,
        "matches": results,
        "count": len(results)
    }


@app.post("/handbags/match/upload", response_model=dict)
async def match_handbag_upload(
    file: UploadFile = File(..., description="Handbag image file"),
    top_k: int = Query(5, ge=1, le=50, description="Number of matches"),
    min_similarity: float = Query(0.0, ge=0.0, le=1.0, description="Minimum similarity")
):
    """
    Match handbag from uploaded image file

    Accepts JPEG, PNG, WebP formats
    Returns top K similar handbags from catalog
    """
    # Read file
    content = await file.read()

    try:
        img = Image.open(io.BytesIO(content)).convert("RGB")
    except Exception as e:
        raise HTTPException(
            status_code=400,
            detail=f"Invalid image file: {str(e)}"
        )

    # Generate embedding
    vec = image_to_vec(img)

    # Query database
    results = query_handbags(vec, top_k=top_k, min_similarity=min_similarity)

    return {
        "query_image": file.filename,
        "matches": results,
        "count": len(results)
    }


@app.get("/handbags/{sku}")
def get_handbag_by_sku(sku: str):
    """Get handbag details by SKU"""
    conn = get_conn()
    try:
        with conn.cursor() as cur:
            cur.execute("""
                SELECT id, sku, brand, model_name, pattern_name,
                       colorway, material, hardware, size, condition,
                       price_usd, price_jpy, image_url, source,
                       verified, created_at, updated_at
                FROM handbags
                WHERE sku = %s AND is_active = true;
            """, (sku,))

            row = cur.fetchone()
            if not row:
                raise HTTPException(status_code=404, detail="Handbag not found")

            cols = [d[0] for d in cur.description]
            return dict(zip(cols, row))
    finally:
        conn.close()


if __name__ == "__main__":
    import uvicorn
    print(f"🚀 Starting Handbag Matching API on port {API_PORT}")
    print(f"📊 Database: {DB_DSN}")
    uvicorn.run(app, host="0.0.0.0", port=API_PORT)