Homepage Redesign

archive/platform-services/vehicles/api/routes/vin.py

@@ -0,0 +1,110 @@
+from fastapi import APIRouter, Depends, HTTPException
+import asyncpg
+from ..dependencies import get_db, get_cache
+from ..services.cache_service import CacheService
+from ..models.responses import VINDecodeRequest, VINDecodeResponse, VINDecodeResult
+import logging
+import re
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/vehicles", tags=["VIN Decoding"])
+
+def validate_vin(vin: str) -> bool:
+    """Validate VIN format"""
+    if len(vin) != 17:
+        return False
+    
+    # VIN cannot contain I, O, Q
+    if any(char in vin.upper() for char in ['I', 'O', 'Q']):
+        return False
+    
+    # Must be alphanumeric
+    if not re.match(r'^[A-HJ-NPR-Z0-9]{17}$', vin.upper()):
+        return False
+    
+    return True
+
+@router.post("/vindecode", response_model=VINDecodeResponse)
+async def decode_vin(
+    request: VINDecodeRequest,
+    db: asyncpg.Connection = Depends(get_db),
+    cache: CacheService = Depends(get_cache)
+):
+    """Decode VIN using PostgreSQL function with MSSQL parity
+    
+    Uses the vehicles.f_decode_vin() function to decode VIN with confidence scoring
+    """
+    vin = request.vin.upper().strip()
+    
+    # Validate VIN format
+    if not validate_vin(vin):
+        return VINDecodeResponse(
+            vin=vin,
+            result=None,
+            success=False,
+            error="Invalid VIN format"
+        )
+    
+    # Check cache first
+    cache_key = f"vin:decode:{vin}"
+    cached_result = await cache.get(cache_key)
+    if cached_result:
+        logger.debug(f"VIN decode result for {vin} retrieved from cache")
+        return VINDecodeResponse(**cached_result)
+    
+    try:
+        # Call PostgreSQL VIN decode function
+        query = """
+            SELECT * FROM vehicles.f_decode_vin($1)
+        """
+        
+        row = await db.fetchrow(query, vin)
+        
+        if row:
+            result = VINDecodeResult(
+                make=row['make'],
+                model=row['model'],
+                year=row['year'],
+                trim_name=row['trim_name'],
+                engine_description=row['engine_description'],
+                transmission_description=row['transmission_description'],
+                horsepower=row.get('horsepower'),
+                torque=row.get('torque'),
+                top_speed=row.get('top_speed'),
+                fuel=row.get('fuel'),
+                confidence_score=float(row['confidence_score']) if row['confidence_score'] else 0.0,
+                vehicle_type=row.get('vehicle_type')
+            )
+            
+            response = VINDecodeResponse(
+                vin=vin,
+                result=result,
+                success=True
+            )
+            
+            # Cache successful decode for 30 days
+            await cache.set(cache_key, response.dict(), ttl=30*24*3600)
+            
+            logger.info(f"Successfully decoded VIN {vin}: {result.make} {result.model} {result.year}")
+            return response
+        else:
+            # No result found
+            response = VINDecodeResponse(
+                vin=vin,
+                result=None,
+                success=False,
+                error="VIN not found in database"
+            )
+            
+            # Cache negative result for 1 hour
+            await cache.set(cache_key, response.dict(), ttl=3600)
+            return response
+            
+    except Exception as e:
+        logger.error(f"Failed to decode VIN {vin}: {e}")
+        return VINDecodeResponse(
+            vin=vin,
+            result=None,
+            success=False,
+            error="Internal server error during VIN decoding"
+        )