| """ |
| 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() |
|
|
