""" TürkçeOCR (`turkceocr.server` / `turkceocr.server`) FastAPI REST Çıkarım ve Doğrulama Servisi. Uç Noktalar (Endpoints): - `GET /health` : Model ailesi durumu, özel HF depoları ve 13 yapısal özel token. - `POST /v1/verify` : Ham VLM/OCR Markdown metnini `TurkceVerify` (TCKN/VKN/IBAN/İ-ı) ve `TurkishFinancialSolver` (Matrah + KDV = Genel Toplam) ile onarır. - `POST /v1/ocr/demo` : Gerçekçi kaşeli/bozulmuş Türkçe belge sentezler, `HighResDiacriticTiler` ile yamalara böler, `TurkceOCRLine` ve `TurkceVerify` hattından geçirir. - `POST /v1/ocr/image` : Base64 kodlu görüntüyü yüksek çözünürlüklü yamalara böler, `TurkceOCRLine` satır okumasını ve `TurkceVerify` onarımını çalıştırır. """ from __future__ import annotations import argparse import base64 import io from typing import Any, Dict, List, Optional from fastapi import FastAPI, HTTPException from PIL import Image from pydantic import BaseModel, Field from turkceocr.cli import inject_realistic_ocr_corruptions, synthesize_demo_document from turkceocr.models.vl_model import SUPPORTED_VL_BACKBONES, TURKCEOCR_SPECIAL_TOKENS from turkceocr.pipeline import TurkceOCRPipeline from turkceocr.postprocess.entity_engine import UnifiedTurkishEntityEngine from turkceocr.postprocess.visual_redactor import VisualRedactor TURKCEOCR_HF_REPOS: List[str] = [ "ersanbil/TurkceOCR", "ersanbil/kvkk-pii-ner", "ersanbil/TurkceOCR-Bench", "ersanbil/TurkceOCR-Demo", ] class HealthResponse(BaseModel): status: str = "ok" brand: str = "TürkçeOCR" version: str = "0.2.0" hf_repos: List[str] = Field(default_factory=lambda: list(TURKCEOCR_HF_REPOS)) special_tokens: List[str] = Field(default_factory=lambda: list(TURKCEOCR_SPECIAL_TOKENS)) class VerifyRequest(BaseModel): markdown: str = Field(..., description="Onarılacak ham OCR / VLM Markdown metni.") repair_financials: bool = Field( default=True, description="Matrah + KDV = Genel Toplam cebirsel ve görsel denklem çözücüsünü çalıştır.", ) backbone: str = Field(default="lightonocr", description="Kullanılacak VLM yapılandırması.") class DemoOCRRequest(BaseModel): doc_type: str = Field( default="erp_table", description="Sentezlenecek belge tipi: erp_table, invoice, pos, handwritten, gazette, born_digital_2col.", ) apply_degradation: bool = Field( default=True, description="TurkceDegrade kaşe, imza ve fiziksel bozulma motorunu uygula.", ) backbone: str = Field(default="lightonocr", description="lightonocr veya paddleocr_vl.") seed: int = Field(default=2026, description="Rastgelelik tohumu.") class ImageOCRRequest(BaseModel): image_base64: str = Field(..., description="Base64 kodlu PNG/JPEG belge görüntüsü.") raw_markdown: Optional[str] = Field( default=None, description="İsteğe bağlı ön-VLM Markdown çıktısı (verilmezse satır kırpımlarından üretilir).", ) backbone: str = Field(default="lightonocr", description="lightonocr veya paddleocr_vl.") repair_financials: bool = Field(default=True, description="Finansal denklem onarımını çalıştır.") class RedactDocumentRequest(BaseModel): markdown: Optional[str] = Field( default=None, description="Analiz edilecek ve KVKK kapsamında karartılacak metin/Markdown.", ) image_base64: Optional[str] = Field( default=None, description="İsteğe bağlı Base64 kodlu belge görüntüsü (verilirse 2B #000000 karartma uygulanır).", ) doc_type: str = Field( default="invoice", description="Görüntü veya metin verilmezse sentezlenecek demo belge tipi.", ) backbone: str = Field(default="lightonocr", description="lightonocr veya paddleocr_vl.") def _serialize_financial_check(fc: Any) -> Optional[Dict[str, Any]]: if fc is None: return None return { "is_consistent": bool(fc.is_consistent), "was_repaired": bool(fc.was_repaired), "subtotal_matrah": float(fc.subtotal_matrah), "vat_kdv": float(fc.vat_kdv), "grand_total": float(fc.grand_total), "inferred_vat_rate": fc.inferred_vat_rate, "repair_note": str(fc.repair_note), } def create_turkceocr_app() -> FastAPI: """TürkçeOCR FastAPI uygulamasını oluşturur ve yapılandırır.""" api = FastAPI( title="TürkçeOCR Inference & Verification API", description=( "İki Katmanlı Açık Kaynak Türkçe OCR ve Doküman Zekası Ailesi " "(TürkçeOCR-VL-1B, TürkçeOCR-Line SVTRv2-TR ve TurkceVerify)" ), version="0.2.0", ) pipeline_cache: Dict[str, TurkceOCRPipeline] = {} def _get_pipeline(backbone: str = "lightonocr") -> TurkceOCRPipeline: norm = backbone if backbone in SUPPORTED_VL_BACKBONES else "lightonocr" if norm not in pipeline_cache: pipeline_cache[norm] = TurkceOCRPipeline(vl_backbone=norm) return pipeline_cache[norm] @api.get("/health", response_model=HealthResponse) def health_check() -> Dict[str, Any]: return { "status": "ok", "brand": "TürkçeOCR", "version": "0.2.0", "hf_repos": list(TURKCEOCR_HF_REPOS), "special_tokens": list(TURKCEOCR_SPECIAL_TOKENS), } @api.post("/v1/verify") def verify_markdown(req: VerifyRequest) -> Dict[str, Any]: pipeline = _get_pipeline(req.backbone) page_out = pipeline.postprocess_document_markdown( raw_markdown=req.markdown, image=None, repair_financials=req.repair_financials, ) rep = page_out.verification_report fin_dict = _serialize_financial_check(page_out.financial_check) return { "brand": "TürkçeOCR", "raw_markdown": page_out.raw_markdown, "markdown": page_out.markdown, "verified_markdown": page_out.markdown, "tckns": rep.valid_tckns, "valid_tckns": rep.valid_tckns, "vkns": rep.valid_vkns, "valid_vkns": rep.valid_vkns, "ibans": rep.valid_ibans, "valid_ibans": rep.valid_ibans, "repaired_fields": rep.repaired_fields, "financial_check": fin_dict, } @api.post("/v1/ocr/demo") def ocr_demo_document(req: DemoOCRRequest) -> Dict[str, Any]: try: sample = synthesize_demo_document( doc_type=req.doc_type, apply_degradation=req.apply_degradation, seed=req.seed, ) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc pipeline = _get_pipeline(req.backbone) corrupted_md = inject_realistic_ocr_corruptions(sample) page_out = pipeline.postprocess_document_markdown( raw_markdown=corrupted_md, image=sample.image, repair_financials=True, ) line_results: List[Dict[str, Any]] = [] for idx, (crop_img, gt_line) in enumerate(sample.line_crops[:4]): pred_text, conf = pipeline.recognize_line_image(crop_img) line_results.append( { "line_index": idx, "ground_truth_line": gt_line, "predicted_text": pred_text, "confidence": round(float(conf), 4), } ) rep = page_out.verification_report return { "brand": "TürkçeOCR", "doc_type": sample.doc_type, "layout_variant": sample.layout_variant, "image_size": list(sample.image.size), "tile_count": len(page_out.tile_bboxes), "tile_bboxes": [list(b) for b in page_out.tile_bboxes], "raw_markdown": page_out.raw_markdown, "markdown": page_out.markdown, "verified_markdown": page_out.markdown, "expected_kie": sample.expected_kie, "tckns": rep.valid_tckns, "valid_tckns": rep.valid_tckns, "vkns": rep.valid_vkns, "valid_vkns": rep.valid_vkns, "ibans": rep.valid_ibans, "valid_ibans": rep.valid_ibans, "repaired_fields": rep.repaired_fields, "financial_check": _serialize_financial_check(page_out.financial_check), "line_recognitions": line_results, } @api.post("/v1/ocr/image") def ocr_base64_image(req: ImageOCRRequest) -> Dict[str, Any]: try: payload = req.image_base64.split(",")[-1].strip() raw_bytes = base64.b64decode(payload) img = Image.open(io.BytesIO(raw_bytes)).convert("RGB") except Exception as exc: raise HTTPException( status_code=400, detail=f"Geçersiz base64 görüntü verisi: {exc}", ) from exc pipeline = _get_pipeline(req.backbone) w, h = img.size strip_coords = [ (0, int(h * 0.04), w, int(h * 0.16)), (0, int(h * 0.20), w, int(h * 0.34)), (0, int(h * 0.68), w, int(h * 0.84)), ] line_results: List[Dict[str, Any]] = [] decoded_lines: List[str] = [] for idx, bbox in enumerate(strip_coords): crop = img.crop(bbox) pred_text, conf = pipeline.recognize_line_image(crop) line_results.append( { "line_index": idx, "bbox": list(bbox), "predicted_text": pred_text, "confidence": round(float(conf), 4), } ) if pred_text.strip(): decoded_lines.append(pred_text.strip()) effective_raw_md = ( req.raw_markdown if req.raw_markdown is not None else ("\n".join(decoded_lines) or "# TürkçeOCR Görüntü Çıktısı") ) page_out = pipeline.postprocess_document_markdown( raw_markdown=effective_raw_md, image=img, repair_financials=req.repair_financials, ) rep = page_out.verification_report return { "brand": "TürkçeOCR", "image_size": [w, h], "tile_count": len(page_out.tile_bboxes), "tile_bboxes": [list(b) for b in page_out.tile_bboxes], "raw_markdown": page_out.raw_markdown, "markdown": page_out.markdown, "verified_markdown": page_out.markdown, "tckns": rep.valid_tckns, "valid_tckns": rep.valid_tckns, "vkns": rep.valid_vkns, "valid_vkns": rep.valid_vkns, "ibans": rep.valid_ibans, "valid_ibans": rep.valid_ibans, "repaired_fields": rep.repaired_fields, "financial_check": _serialize_financial_check(page_out.financial_check), "line_recognitions": line_results, } entity_engine = UnifiedTurkishEntityEngine(use_ner=False) visual_redactor = VisualRedactor(entity_engine=entity_engine) @api.post("/v1/ocr/redact") def redact_ocr_document(req: RedactDocumentRequest) -> Dict[str, Any]: pipeline = _get_pipeline(req.backbone) img: Optional[Image.Image] = None raw_md: str = req.markdown or "" if req.image_base64: try: payload = req.image_base64.split(",")[-1].strip() raw_bytes = base64.b64decode(payload) img = Image.open(io.BytesIO(raw_bytes)).convert("RGB") except Exception as exc: raise HTTPException( status_code=400, detail=f"Geçersiz base64 görüntü verisi: {exc}", ) from exc if img is None and not raw_md: sample = synthesize_demo_document(doc_type=req.doc_type, apply_degradation=True, seed=2026) img = sample.image raw_md = inject_realistic_ocr_corruptions(sample) elif img is None: img = Image.new("RGB", (900, 600), color=(255, 255, 255)) page_out = pipeline.postprocess_document_markdown( raw_markdown=raw_md, image=img, repair_financials=True, ) redaction_res = visual_redactor.redact_document( image=img, raw_text=page_out.raw_markdown, normalized_text=page_out.markdown, ) buf = io.BytesIO() redaction_res.redacted_image.save(buf, format="PNG") redacted_b64 = base64.b64encode(buf.getvalue()).decode("ascii") return { "brand": "TürkçeOCR", "raw_markdown": page_out.raw_markdown, "verified_markdown": page_out.markdown, "redacted_markdown": redaction_res.redacted_text, "entity_count": len(redaction_res.entities), "entities": [e.to_dict() for e in redaction_res.entities], "redaction_box_count": len(redaction_res.redaction_boxes), "redaction_boxes": [b.to_dict() for b in redaction_res.redaction_boxes], "redaction_coverage": round(float(redaction_res.redaction_coverage), 4), "redacted_image_base64": redacted_b64, } return api app = create_turkceocr_app() def main() -> None: """`turkceocr-serve` konsol sunucu başlatıcısı.""" import uvicorn parser = argparse.ArgumentParser(description="TürkçeOCR FastAPI Sunucusu") parser.add_argument("--host", type=str, default="127.0.0.1") parser.add_argument("--port", type=int, default=8000) args = parser.parse_args() uvicorn.run(app, host=args.host, port=args.port) if __name__ == "__main__": main()