from __future__ import annotations import time from typing import List, Optional from fastapi import APIRouter, File, HTTPException, UploadFile from pydantic import BaseModel, Field from app.config import get_settings from app.core.logger import get_logger from app.services.qr_decoder_service import QRDecoderService logger = get_logger(__name__) router = APIRouter() _settings = get_settings() MAX_UPLOAD_BYTES = _settings.max_upload_bytes _service = QRDecoderService(max_file_size_mb=_settings.max_upload_mb) class QRDecodeUrlRequest(BaseModel): url: str = Field( ..., description="Public URL of the QR code image to decode", ) class DecodedQR(BaseModel): data: str type: str = "QRCODE" bounding_box: Optional[List[List[float]]] = None decoder: Optional[str] = None class QRDecodeResponse(BaseModel): success: bool time_ms: float decoded: List[DecodedQR] count: int error_message: Optional[str] = None @router.post( "/qr-decode/file", response_model=QRDecodeResponse, summary="Decode QR code from an uploaded image file", description="Upload an image file containing a QR code and get its decoded content.", ) async def decode_qr_file( file: UploadFile = File(..., description="Image file containing a QR code"), ) -> QRDecodeResponse: start = time.perf_counter() if not file.content_type or not file.content_type.startswith("image/"): elapsed = round((time.perf_counter() - start) * 1000, 3) raise HTTPException( status_code=400, detail=QRDecodeResponse( success=False, time_ms=elapsed, decoded=[], count=0, error_message="Uploaded file must be an image.", ).model_dump(), ) raw = await file.read() if len(raw) == 0: elapsed = round((time.perf_counter() - start) * 1000, 3) raise HTTPException( status_code=400, detail=QRDecodeResponse( success=False, time_ms=elapsed, decoded=[], count=0, error_message="Uploaded file is empty.", ).model_dump(), ) if len(raw) > MAX_UPLOAD_BYTES: elapsed = round((time.perf_counter() - start) * 1000, 3) raise HTTPException( status_code=413, detail=QRDecodeResponse( success=False, time_ms=elapsed, decoded=[], count=0, error_message=f"File exceeds maximum size of {MAX_UPLOAD_BYTES // (1024 * 1024)} MB.", ).model_dump(), ) result = await _service.decode_from_bytes(raw, source_label=file.filename or "upload") if not result.success: logger.warning("QR decode failed for uploaded file", extra={ "filename": file.filename, "error": result.error_message, }) raise HTTPException( status_code=422, detail=QRDecodeResponse( success=False, time_ms=result.processing_time_ms, decoded=[], count=0, error_message=result.error_message, ).model_dump(), ) items = [DecodedQR(**item) for item in result.decoded_data] logger.info("QR decode successful", extra={ "filename": file.filename, "count": len(items), }) return QRDecodeResponse( success=True, time_ms=result.processing_time_ms, decoded=items, count=len(items), error_message=None, ) @router.post( "/qr-decode/url", response_model=QRDecodeResponse, summary="Decode QR code from an image URL", description="Provide a public URL to an image containing a QR code and get its decoded content.", ) async def decode_qr_url( body: QRDecodeUrlRequest, ) -> QRDecodeResponse: start = time.perf_counter() if not body.url.lower().startswith(("http://", "https://")): elapsed = round((time.perf_counter() - start) * 1000, 3) raise HTTPException( status_code=400, detail=QRDecodeResponse( success=False, time_ms=elapsed, decoded=[], count=0, error_message="URL must start with http:// or https://.", ).model_dump(), ) result = await _service.decode_from_url(body.url) if not result.success: logger.warning("QR decode failed for URL", extra={ "url": body.url, "error": result.error_message, }) raise HTTPException( status_code=422, detail=QRDecodeResponse( success=False, time_ms=result.processing_time_ms, decoded=[], count=0, error_message=result.error_message, ).model_dump(), ) items = [DecodedQR(**item) for item in result.decoded_data] logger.info("QR decode successful", extra={ "url": body.url, "count": len(items), }) return QRDecodeResponse( success=True, time_ms=result.processing_time_ms, decoded=items, count=len(items), error_message=None, )