Spaces:
Running
Running
File size: 4,975 Bytes
ebb9029 faefb1f ebb9029 b7dddbe ebb9029 8f9855d ebb9029 8f9855d ebb9029 8f9855d ebb9029 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 | 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,
)
|