File size: 2,510 Bytes
581d33f 2648cef 581d33f 2648cef 581d33f 2648cef 581d33f 2648cef 581d33f 2648cef 581d33f 2648cef 581d33f 2648cef 581d33f 2648cef 581d33f 2648cef 581d33f 2648cef 581d33f 2648cef 3bcbec3 581d33f 2648cef 581d33f | 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 | from fastapi import FastAPI, UploadFile, File, Form, HTTPException
from fastapi.responses import JSONResponse
from PIL import Image
import numpy as np
import cv2
import io
import requests
from typing import Optional
from pydantic import BaseModel
from rapidocr_onnxruntime import RapidOCR
app = FastAPI(title="OCR API", description="RapidOCR API Service")
engine = RapidOCR()
def process_ocr(image: np.ndarray, use_det: bool, use_cls: bool, use_rec: bool) -> list:
"""Xử lý OCR và trả về kết quả"""
img_bgr = cv2.cvtColor(image, cv2.COLOR_RGB2BGR)
result, _ = engine(img_bgr, use_det=use_det, use_cls=use_cls, use_rec=use_rec)
if not result:
return []
texts = []
for item in result:
if len(item) == 3:
box, text, score = item
texts.append({
"text": text,
"confidence": float(score),
"bbox": box.tolist() if hasattr(box, 'tolist') else box
})
elif len(item) == 2:
_, text = item
texts.append({
"text": str(text),
"confidence": None,
"bbox": None
})
return texts
@app.post("/ocr")
async def ocr_from_file(
file: UploadFile = File(..., description="Image file to process"),
use_det: bool = Form(True, description="Use text detection"),
use_cls: bool = Form(True, description="Use classification"),
use_rec: bool = Form(True, description="Use recognition")
):
if not file.content_type or not file.content_type.startswith('image/'):
raise HTTPException(status_code=400, detail="File must be an image")
try:
contents = await file.read()
image = Image.open(io.BytesIO(contents))
if image.mode != 'RGB':
image = image.convert('RGB')
img_np = np.array(image)
results = process_ocr(img_np, use_det, use_cls, use_rec)
return JSONResponse(content={
"success": True,
"texts": [item["text"] for item in results],
"details": results,
"num_texts": len(results)
})
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error: {str(e)}")
@app.get("/health")
async def health_check():
"""Health check endpoint"""
return {"status": "healthy", "service": "XOCR AI"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=7860) |