Spaces:
Sleeping
Sleeping
File size: 2,603 Bytes
862cbed | 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 | """Image upload route — accepts an image, runs OCR, produces XML."""
from __future__ import annotations
import tempfile
from pathlib import Path
from typing import Any
from fastapi import APIRouter, HTTPException, UploadFile
from src.app.api import get_job_service
from src.app.domain.models import RawProviderPayload
router = APIRouter(tags=["ocr"])
ALLOWED_EXTENSIONS = {".png", ".jpg", ".jpeg", ".tiff", ".tif", ".webp", ".bmp"}
MAX_IMAGE_SIZE = 20 * 1024 * 1024 # 20 MB
@router.post("/ocr", status_code=201)
async def ocr_image(image: UploadFile) -> dict[str, Any]:
"""Upload an image, run PaddleOCR, and produce ALTO/PAGE XML.
This is the main user-facing endpoint — no JSON payload needed.
"""
if not image.filename:
raise HTTPException(status_code=422, detail="No filename provided")
ext = Path(image.filename).suffix.lower()
if ext not in ALLOWED_EXTENSIONS:
raise HTTPException(
status_code=422,
detail=f"Unsupported format '{ext}'. Use: {', '.join(sorted(ALLOWED_EXTENSIONS))}",
)
content = await image.read(MAX_IMAGE_SIZE + 1)
if len(content) > MAX_IMAGE_SIZE:
raise HTTPException(status_code=413, detail="Image too large (max 20 MB)")
with tempfile.NamedTemporaryFile(suffix=ext, delete=False) as tmp:
tmp.write(content)
tmp_path = Path(tmp.name)
try:
from PIL import Image as PILImage
with PILImage.open(tmp_path) as img:
image_width, image_height = img.size
from src.app.ocr import paddle_result_to_payload, run_paddle_ocr
results = run_paddle_ocr(tmp_path)
if not results:
raise HTTPException(
status_code=422,
detail="PaddleOCR found no text in this image",
)
payload = paddle_result_to_payload(results)
raw = RawProviderPayload(
provider_id="paddleocr",
adapter_id="adapter.word_box_json.v1",
runtime_type="local",
payload=payload,
image_width=image_width,
image_height=image_height,
)
svc = get_job_service()
job = svc.create_job(
provider_id="paddleocr",
provider_family="word_box_json",
source_filename=image.filename,
)
result = svc.run_job(
job, raw,
image_width=image_width,
image_height=image_height,
image_path=tmp_path,
)
return result.to_summary()
finally:
tmp_path.unlink(missing_ok=True)
|