786 lines
32 KiB
Python
786 lines
32 KiB
Python
import base64
|
||
import importlib.util
|
||
import json
|
||
import re
|
||
from pathlib import Path
|
||
|
||
from config import (
|
||
ANTHROPIC_API_KEY, GEMINI_API_KEY,
|
||
CLAUDE_MODEL, GEMINI_MODEL, DEFAULT_PROVIDER, UPLOAD_DIR,
|
||
)
|
||
|
||
SYSTEM_PROMPT = """You are an expert OCR assistant specializing in Lebanese government real estate documents written in Arabic (right-to-left). Your task is to extract structured data from scanned document images of "بطاقة معلومات عن الملكية العقارية" (Real Estate Property Information Cards) issued by the Lebanese Ministry of Real Estate Affairs.
|
||
|
||
CRITICAL INSTRUCTIONS:
|
||
1. Extract ALL text exactly as written in Arabic — do not translate or transliterate.
|
||
2. For dates, preserve the original format as printed (e.g., ٢٩-٣-٢٠٢٦ or 29/03/2026).
|
||
3. For the properties table, extract EVERY row — documents often have 30+ rows. If the document prominently states "لا يملك", set owns_properties to false and return an empty array [] for properties.
|
||
4. If a field is illegible or absent, use null.
|
||
5. Numbers may be in Arabic-Indic digits (٠١٢٣٤٥٦٧٨٩) or Western digits — preserve exactly as-is.
|
||
6. The header and footer sections contain these labeled fields:
|
||
- الاسم (first_name), اسم الأب (father_name), اسم الأم (mother_name)
|
||
- الشهرة (family_name / family_origin), الجنسية (nationality)
|
||
- تاريخ الولادة (birth_date), رقم السجل (registry_number), مكان السجل (registry_place)
|
||
- رقم الطلب (request_number), تاريخ الطلب (request_date)
|
||
- اسم المستدعي/الوكيل (applicant_name_raw)
|
||
- الغاية من الطلب (request_purpose)
|
||
- نطاق البحث contains القضاء or المحافظات (qaza/district/governorate) — extract as search_scope
|
||
- in the bottom footer, find the page fraction (e.g. "صفحة 1 من 3" or "1 / 3") and extract as page_info.
|
||
- معلومات محولة لغاية (data_valid_until) found near the bottom.
|
||
- أمانة السجل (registry_office) found at the bottom right.
|
||
- عدد العقارات: (declared_property_count) located right under the properties table.
|
||
7. If the property owner is a company, religious entity, or organization (e.g., شركة, وقف, مطرانية, جمعية) rather than a natural person, extract its full name into the `first_name` field and leave `father_name`, `mother_name`, `family_name`, etc. as null.
|
||
8. The properties table has 8 columns in order from RIGHT to LEFT as they appear on the page:
|
||
col1(rightmost)=اسم الفريق, col2=رقم العقار, col3=القسم, col4=البلوك,
|
||
col5=المنطقة العقارية, col6=القضاء, col7=عدد الأسهم, col8(leftmost)=نوع الملكية
|
||
Map these to JSON keys: party_name, property_number, section, block, real_estate_district, qaza, num_shares, ownership_type
|
||
9. Return ONLY valid JSON matching the schema. No markdown, no explanation."""
|
||
|
||
USER_PROMPT = (
|
||
"Extract all data from this Lebanese real estate property card. "
|
||
"Return a single JSON object with these keys: "
|
||
"request_number, request_date, applicant_name_raw, request_purpose, data_valid_until, "
|
||
"registry_office, page_info, search_scope, owns_properties, declared_property_count, "
|
||
"person (object with: first_name (or company/entity name), father_name, mother_name, family_name, "
|
||
"family_origin, nationality, birth_date, registry_number, registry_place), "
|
||
"properties (array of objects with: party_name, property_number, section, block, "
|
||
"real_estate_district, qaza, num_shares, ownership_type), "
|
||
"extraction_notes. "
|
||
"Use null for missing fields. Include every property row from the table (or empty array if none)."
|
||
)
|
||
|
||
CORRELATION_SYSTEM_PROMPT = """You are verifying whether two scanned Lebanese real estate document pages belong to the same multi-page request and the same person.
|
||
|
||
Rules:
|
||
1. Compare both page images and the extracted metadata together.
|
||
2. Strong signals: matching request number, matching search scope, matching page numbering in the same sequence, matching applicant/person names, matching footer/header identifiers, and obvious continuation of the same document layout.
|
||
3. Weak OCR differences are common in Arabic letters such as س and ن, ب and ت, or Arabic-Indic digits. Do not reject a match solely because of one likely OCR confusion.
|
||
4. Reject when there are clear contradictions in request number, search scope, person identity, or unrelated page numbering.
|
||
5. Return JSON only.
|
||
"""
|
||
|
||
|
||
def _get_ai_verification_provider(preferred_provider: str = "") -> str:
|
||
"""Return a provider capable of vision reasoning for verification."""
|
||
providers = [p["id"] for p in get_available_providers() if provider_supports_ai_verification(p["id"])]
|
||
if preferred_provider in providers:
|
||
return preferred_provider
|
||
if DEFAULT_PROVIDER in providers:
|
||
return DEFAULT_PROVIDER
|
||
if providers:
|
||
return providers[0]
|
||
raise ValueError("No AI verification provider configured")
|
||
|
||
|
||
def provider_supports_ai_verification(provider_id: str) -> bool:
|
||
return provider_id == "claude" or provider_id == "gemini" or provider_id.startswith("gemini-")
|
||
|
||
|
||
def _build_correlation_user_prompt(current_context: dict, candidate_context: dict) -> str:
|
||
return (
|
||
"Determine whether PAGE_A and PAGE_B belong to the same multi-page request/document for the same person. "
|
||
"Treat minor OCR mistakes as possible noise. Return one JSON object with keys: "
|
||
"same_document (boolean), confidence ('high'|'medium'|'low'), verdict_ar (short Arabic sentence), "
|
||
"reasons_ar (array of short Arabic bullet strings), mismatch_flags (array of short Arabic strings), "
|
||
"recommended_action ('auto-link'|'manual-review').\n\n"
|
||
f"PAGE_A_CONTEXT={json.dumps(current_context, ensure_ascii=False)}\n"
|
||
f"PAGE_B_CONTEXT={json.dumps(candidate_context, ensure_ascii=False)}"
|
||
)
|
||
|
||
|
||
async def verify_page_correlation(
|
||
current_image_path: str,
|
||
candidate_image_path: str,
|
||
current_context: dict,
|
||
candidate_context: dict,
|
||
provider: str = "",
|
||
) -> dict:
|
||
"""Ask a vision model if two pages belong to the same multi-page request."""
|
||
provider = _get_ai_verification_provider(provider)
|
||
|
||
if provider == "claude":
|
||
import anthropic
|
||
|
||
client = anthropic.AsyncAnthropic(api_key=ANTHROPIC_API_KEY)
|
||
current_full_path = _resolve_path(current_image_path)
|
||
candidate_full_path = _resolve_path(candidate_image_path)
|
||
current_img_data, current_media_type = _encode_image(current_full_path)
|
||
candidate_img_data, candidate_media_type = _encode_image(candidate_full_path)
|
||
|
||
message = await client.messages.create(
|
||
model=CLAUDE_MODEL,
|
||
max_tokens=1200,
|
||
system=[{"type": "text", "text": CORRELATION_SYSTEM_PROMPT}],
|
||
messages=[
|
||
{
|
||
"role": "user",
|
||
"content": [
|
||
{"type": "text", "text": _build_correlation_user_prompt(current_context, candidate_context)},
|
||
{
|
||
"type": "image",
|
||
"source": {
|
||
"type": "base64",
|
||
"media_type": current_media_type,
|
||
"data": current_img_data,
|
||
},
|
||
},
|
||
{
|
||
"type": "image",
|
||
"source": {
|
||
"type": "base64",
|
||
"media_type": candidate_media_type,
|
||
"data": candidate_img_data,
|
||
},
|
||
},
|
||
],
|
||
}
|
||
],
|
||
)
|
||
result = _parse_json_lenient(message.content[0].text)
|
||
elif provider == "gemini":
|
||
import asyncio
|
||
from google import genai
|
||
from google.genai import types
|
||
|
||
client = genai.Client(api_key=GEMINI_API_KEY)
|
||
current_full_path = _resolve_path(current_image_path)
|
||
candidate_full_path = _resolve_path(candidate_image_path)
|
||
|
||
with open(current_full_path, "rb") as f:
|
||
current_bytes = f.read()
|
||
with open(candidate_full_path, "rb") as f:
|
||
candidate_bytes = f.read()
|
||
|
||
mime_map = {
|
||
".jpg": "image/jpeg",
|
||
".jpeg": "image/jpeg",
|
||
".png": "image/png",
|
||
".webp": "image/webp",
|
||
}
|
||
current_mime = mime_map.get(Path(current_full_path).suffix.lower(), "image/jpeg")
|
||
candidate_mime = mime_map.get(Path(candidate_full_path).suffix.lower(), "image/jpeg")
|
||
|
||
def _call():
|
||
response = client.models.generate_content(
|
||
model=GEMINI_MODEL,
|
||
contents=[
|
||
types.Content(
|
||
parts=[
|
||
types.Part(text=CORRELATION_SYSTEM_PROMPT + "\n\n" + _build_correlation_user_prompt(current_context, candidate_context)),
|
||
types.Part(inline_data=types.Blob(mime_type=current_mime, data=current_bytes)),
|
||
types.Part(inline_data=types.Blob(mime_type=candidate_mime, data=candidate_bytes)),
|
||
]
|
||
)
|
||
],
|
||
config=types.GenerateContentConfig(
|
||
temperature=0.1,
|
||
max_output_tokens=1600,
|
||
),
|
||
)
|
||
return response.text
|
||
|
||
raw_text = await asyncio.get_event_loop().run_in_executor(None, _call)
|
||
result = _parse_json_lenient(raw_text)
|
||
else:
|
||
raise ValueError(f"Provider {provider} does not support AI correlation verification")
|
||
|
||
return {
|
||
"provider": provider,
|
||
"same_document": bool(result.get("same_document")),
|
||
"confidence": result.get("confidence") or "medium",
|
||
"verdict_ar": result.get("verdict_ar") or "تعذر توليد خلاصة واضحة.",
|
||
"reasons_ar": result.get("reasons_ar") or [],
|
||
"mismatch_flags": result.get("mismatch_flags") or [],
|
||
"recommended_action": result.get("recommended_action") or "manual-review",
|
||
}
|
||
|
||
def _resolve_path(image_path: str) -> str:
|
||
if Path(image_path).is_absolute():
|
||
return image_path
|
||
return str(Path(UPLOAD_DIR) / image_path)
|
||
|
||
|
||
def _encode_image(image_path: str) -> tuple[str, str]:
|
||
path = Path(image_path)
|
||
suffix = path.suffix.lower()
|
||
media_type_map = {
|
||
".jpg": "image/jpeg",
|
||
".jpeg": "image/jpeg",
|
||
".png": "image/png",
|
||
".webp": "image/webp",
|
||
".gif": "image/gif",
|
||
}
|
||
media_type = media_type_map.get(suffix, "image/jpeg")
|
||
with open(image_path, "rb") as f:
|
||
data = base64.standard_b64encode(f.read()).decode("utf-8")
|
||
return data, media_type
|
||
|
||
|
||
def _strip_code_fences(text: str) -> str:
|
||
text = text.strip()
|
||
if text.startswith("```"):
|
||
lines = text.split("\n")
|
||
text = "\n".join(lines[1:-1] if lines[-1].strip() == "```" else lines[1:])
|
||
return text.strip()
|
||
|
||
|
||
def _parse_json_lenient(text: str) -> dict:
|
||
"""Parse JSON even if the model added prose, trailing commas, or truncation."""
|
||
text = _strip_code_fences(text or "")
|
||
if not text:
|
||
raise ValueError("Empty response from model")
|
||
|
||
# Fast path
|
||
try:
|
||
return json.loads(text)
|
||
except Exception:
|
||
pass
|
||
|
||
# Extract the first {...} block (greedy to last })
|
||
start = text.find("{")
|
||
end = text.rfind("}")
|
||
if start != -1 and end > start:
|
||
candidate = text[start:end + 1]
|
||
try:
|
||
return json.loads(candidate)
|
||
except Exception:
|
||
# Remove trailing commas before } or ]
|
||
cleaned = re.sub(r",(\s*[}\]])", r"\1", candidate)
|
||
try:
|
||
return json.loads(cleaned)
|
||
except Exception:
|
||
pass
|
||
|
||
raise ValueError(f"Could not parse model response as JSON: {text[:200]}")
|
||
|
||
|
||
# Individual Gemini models exposed to the UI
|
||
_GEMINI_MODEL_OPTIONS = [
|
||
("gemini-2.5-pro", "Gemini 2.5 Pro (أفضل دقة)"),
|
||
("gemini-2.5-flash", "Gemini 2.5 Flash (سريع)"),
|
||
("gemini-2.0-flash", "Gemini 2.0 Flash"),
|
||
("gemini-2.0-flash-lite", "Gemini 2.0 Flash Lite (احتياطي)"),
|
||
]
|
||
|
||
|
||
def _easyocr_is_available() -> bool:
|
||
return importlib.util.find_spec("easyocr") is not None
|
||
|
||
|
||
def get_available_providers() -> list[dict]:
|
||
"""Return only providers that are actually usable in the current runtime."""
|
||
providers = []
|
||
if _easyocr_is_available():
|
||
providers.append({"id": "easyocr", "name": "EasyOCR (مجاني)", "model": "local"})
|
||
if ANTHROPIC_API_KEY:
|
||
providers.append({"id": "claude", "name": "Claude (Anthropic)", "model": CLAUDE_MODEL})
|
||
if GEMINI_API_KEY:
|
||
for model_id, label in _GEMINI_MODEL_OPTIONS:
|
||
providers.append({"id": model_id, "name": label, "model": model_id})
|
||
return providers
|
||
|
||
|
||
def get_default_provider() -> str:
|
||
"""Return the default provider, falling back to whichever is available."""
|
||
providers = get_available_providers()
|
||
if not providers:
|
||
return GEMINI_MODEL if DEFAULT_PROVIDER == "gemini" else DEFAULT_PROVIDER
|
||
ids = [p["id"] for p in providers]
|
||
# Map legacy "gemini" default to the configured GEMINI_MODEL
|
||
resolved = GEMINI_MODEL if DEFAULT_PROVIDER == "gemini" else DEFAULT_PROVIDER
|
||
if resolved in ids:
|
||
return resolved
|
||
if DEFAULT_PROVIDER in ids:
|
||
return DEFAULT_PROVIDER
|
||
return ids[0]
|
||
|
||
|
||
# ─── Claude extraction ───────────────────────────────────────────
|
||
|
||
async def _extract_with_claude(image_path: str) -> dict:
|
||
import anthropic
|
||
import asyncio
|
||
|
||
client = anthropic.AsyncAnthropic(api_key=ANTHROPIC_API_KEY)
|
||
full_path = _resolve_path(image_path)
|
||
img_data, media_type = _encode_image(full_path)
|
||
|
||
last_error = None
|
||
for attempt in range(3):
|
||
try:
|
||
message = await client.messages.create(
|
||
model=CLAUDE_MODEL,
|
||
max_tokens=8192,
|
||
system=[
|
||
{
|
||
"type": "text",
|
||
"text": SYSTEM_PROMPT,
|
||
"cache_control": {"type": "ephemeral"},
|
||
}
|
||
],
|
||
messages=[
|
||
{
|
||
"role": "user",
|
||
"content": [
|
||
{
|
||
"type": "image",
|
||
"source": {
|
||
"type": "base64",
|
||
"media_type": media_type,
|
||
"data": img_data,
|
||
},
|
||
},
|
||
{"type": "text", "text": USER_PROMPT},
|
||
],
|
||
}
|
||
],
|
||
)
|
||
return _parse_json_lenient(message.content[0].text)
|
||
except Exception as e:
|
||
last_error = e
|
||
err_str = str(e)
|
||
# Transient: overload/rate-limit/network — back off and retry
|
||
if any(code in err_str for code in ("529", "503", "502", "500", "overload", "Overload", "timeout", "Timeout")):
|
||
await asyncio.sleep(2 * (attempt + 1))
|
||
continue
|
||
if "429" in err_str and attempt < 2:
|
||
await asyncio.sleep(5 * (attempt + 1))
|
||
continue
|
||
raise
|
||
raise last_error
|
||
|
||
|
||
# ─── Gemini extraction ───────────────────────────────────────────
|
||
|
||
async def _extract_with_gemini(image_path: str, model: str = "") -> dict:
|
||
import asyncio
|
||
from google import genai
|
||
from google.genai import types
|
||
|
||
client = genai.Client(api_key=GEMINI_API_KEY)
|
||
full_path = _resolve_path(image_path)
|
||
|
||
with open(full_path, "rb") as f:
|
||
image_bytes = f.read()
|
||
|
||
suffix = Path(full_path).suffix.lower()
|
||
mime_map = {".jpg": "image/jpeg", ".jpeg": "image/jpeg",
|
||
".png": "image/png", ".webp": "image/webp"}
|
||
mime_type = mime_map.get(suffix, "image/jpeg")
|
||
|
||
# Full fallback chain: start from requested model, cascade through cheaper/available ones
|
||
_ALL_GEMINI_FALLBACK = [
|
||
"gemini-2.5-pro",
|
||
"gemini-2.5-flash",
|
||
"gemini-2.0-flash",
|
||
"gemini-2.0-flash-lite",
|
||
]
|
||
# If a specific model was requested, start from it; otherwise start from the configured default
|
||
start_model = model if model else GEMINI_MODEL
|
||
# Build ordered list: requested model first, then remaining fallbacks in order
|
||
models_to_try = [start_model] + [m for m in _ALL_GEMINI_FALLBACK if m != start_model]
|
||
|
||
def _call(model_name: str):
|
||
response = client.models.generate_content(
|
||
model=model_name,
|
||
contents=[
|
||
types.Content(parts=[
|
||
types.Part(text=SYSTEM_PROMPT + "\n\n" + USER_PROMPT),
|
||
types.Part(inline_data=types.Blob(mime_type=mime_type, data=image_bytes)),
|
||
])
|
||
],
|
||
config=types.GenerateContentConfig(
|
||
temperature=0.1,
|
||
max_output_tokens=8192,
|
||
),
|
||
)
|
||
text = response.text
|
||
if not text:
|
||
try:
|
||
parts = []
|
||
for cand in (response.candidates or []):
|
||
for part in (cand.content.parts or []):
|
||
if getattr(part, "text", None):
|
||
parts.append(part.text)
|
||
text = "".join(parts)
|
||
except Exception:
|
||
pass
|
||
if not text:
|
||
raise ValueError("Gemini returned empty response")
|
||
return text
|
||
|
||
last_error = None
|
||
for model_name in models_to_try:
|
||
for attempt in range(3):
|
||
try:
|
||
raw_text = await asyncio.get_event_loop().run_in_executor(None, lambda m=model_name: _call(m))
|
||
return _parse_json_lenient(raw_text)
|
||
except Exception as e:
|
||
last_error = e
|
||
err_str = str(e)
|
||
# Quota exhaustion — skip to next model immediately
|
||
if "429" in err_str or "RESOURCE_EXHAUSTED" in err_str:
|
||
break
|
||
# Transient errors — back off and retry same model
|
||
if any(code in err_str for code in ("503", "502", "500", "UNAVAILABLE", "DEADLINE", "timeout", "Timeout", "empty response", "Could not parse")):
|
||
await asyncio.sleep(2 * (attempt + 1))
|
||
continue
|
||
raise
|
||
|
||
raise last_error
|
||
|
||
|
||
# ─── EasyOCR extraction (free, local) ────────────────────────────
|
||
|
||
# Cache EasyOCR reader (heavy to initialize)
|
||
_easyocr_reader = None
|
||
|
||
|
||
def _get_easyocr_reader():
|
||
global _easyocr_reader
|
||
if _easyocr_reader is None:
|
||
import easyocr
|
||
_easyocr_reader = easyocr.Reader(["ar", "en"], gpu=False)
|
||
return _easyocr_reader
|
||
|
||
|
||
# Known label strings for header fields
|
||
_LABEL_MAP = {
|
||
"الاسم": "first_name",
|
||
"اسم الأب": "father_name",
|
||
"اسم الأم": "mother_name",
|
||
"الشهرة": "family_origin",
|
||
"الجنسية": "nationality",
|
||
"تاريخ الولادة": "birth_date",
|
||
"رقم السجل": "registry_number",
|
||
"مكان السجل": "registry_place",
|
||
}
|
||
|
||
# Table header labels that mark the start of the properties table
|
||
_TABLE_HEADERS = {"اسم الفريق", "رقم العقار", "القسم", "البلوك",
|
||
"المنطقة العقارية", "القضاء", "عدد الأسهم", "نوع الملكية"}
|
||
|
||
# Column-header text → JSON key (right→left order on page)
|
||
_COL_KEYS = [
|
||
("اسم الفريق", "party_name"),
|
||
("رقم العقار", "property_number"),
|
||
("القسم", "section"),
|
||
("البلوك", "block"),
|
||
("المنطقة العقارية", "real_estate_district"),
|
||
("القضاء", "qaza"),
|
||
("عدد الأسهم", "num_shares"),
|
||
("نوع الملكية", "ownership_type"),
|
||
]
|
||
|
||
|
||
def _parse_easyocr_results(results: list) -> dict:
|
||
"""
|
||
Parse EasyOCR bounding-box results into structured document data.
|
||
Each result is (bbox, text, confidence).
|
||
bbox = [[x1,y1],[x2,y2],[x3,y3],[x4,y4]]
|
||
|
||
Strategy:
|
||
1. Build a list of items with (text, x_center, y_center, x_left, x_right).
|
||
2. Detect the table-header row → derive column X-boundary ranges.
|
||
3. Header section (above table): find label items, then take the nearest
|
||
right-adjacent item as the value (RTL: label is in the centre/left,
|
||
value is to the right of it on the same row).
|
||
4. Table body rows: assign each cell to a column by matching its X-center
|
||
to the column boundary derived from the header row.
|
||
"""
|
||
if not results:
|
||
return _empty_result("No text detected")
|
||
|
||
items = []
|
||
for bbox, text, conf in results:
|
||
xs = [pt[0] for pt in bbox]
|
||
ys = [pt[1] for pt in bbox]
|
||
items.append({
|
||
"text": text.strip(),
|
||
"x": (min(xs) + max(xs)) / 2,
|
||
"y": (min(ys) + max(ys)) / 2,
|
||
"x_left": min(xs),
|
||
"x_right": max(xs),
|
||
})
|
||
|
||
items.sort(key=lambda t: t["y"])
|
||
|
||
# ── 1. Locate the table header row ──────────────────────────────────────
|
||
header_row_items = []
|
||
header_y = None
|
||
|
||
# Simple heuristic to find where the table starts:
|
||
# First, locate the word "العقارات" or "المملوكة" or "نطاق البحث"
|
||
for item in items:
|
||
if "العقارات" in item["text"] or "المملوكة" in item["text"]:
|
||
header_y = item["y"]
|
||
break
|
||
|
||
if header_y is None:
|
||
for item in items:
|
||
if item["text"] in _TABLE_HEADERS:
|
||
header_y = item["y"]
|
||
break
|
||
|
||
for item in items:
|
||
if header_y is not None and 0 < item["y"] - header_y < 80:
|
||
# The table headers are right below "العقارات المملوكة"
|
||
# Let's collect anything that looks like a header column label
|
||
if any(h in item["text"] for h in ["العقار", "الفريق", "القسم", "البلوك", "المنطقة", "قضاء", "أسهم", "نوع"]):
|
||
header_row_items.append(item)
|
||
|
||
if header_row_items:
|
||
# Sort header cells right→left (descending x)
|
||
header_row_items.sort(key=lambda t: -t["x"])
|
||
|
||
# ── 2. Build column X-boundaries from header row ─────────────────────────
|
||
col_boundaries = [] # list of (x_left, x_right, json_key)
|
||
|
||
# Define a fuzzy matching logic or regex to assign headers
|
||
key_keywords = {
|
||
"party_name": ["فريق"],
|
||
"property_number": ["رقم", "عقار"],
|
||
"section": ["قسم"],
|
||
"block": ["بلوك"],
|
||
"real_estate_district": ["منطقة", "عقارية"],
|
||
"qaza": ["قضاء"],
|
||
"num_shares": ["أسهم", "سهم"],
|
||
"ownership_type": ["نوع", "ملكية"],
|
||
}
|
||
|
||
if header_row_items:
|
||
for it in header_row_items:
|
||
assigned_key = None
|
||
for key, keywords in key_keywords.items():
|
||
if any(kw in it["text"] for kw in keywords):
|
||
assigned_key = key
|
||
break
|
||
if assigned_key:
|
||
# Add large horizontal padding to ensure columns grab cells correctly
|
||
col_boundaries.append((it["x_left"] - 40, it["x_right"] + 40, assigned_key))
|
||
|
||
table_start_y = header_row_items[0]["y"] if header_row_items else (header_y if header_y is not None else 600)
|
||
|
||
# ── 3. Extract header/person fields (above the table) ────────────────────
|
||
header_items = [it for it in items if it["y"] < (header_y if header_y is not None else table_start_y) - 5]
|
||
|
||
# Group header items into rows
|
||
header_rows: list[list[dict]] = []
|
||
if header_items:
|
||
current = [header_items[0]]
|
||
for it in header_items[1:]:
|
||
if abs(it["y"] - current[0]["y"]) < 15:
|
||
current.append(it)
|
||
else:
|
||
header_rows.append(sorted(current, key=lambda t: -t["x"])) # right→left
|
||
current = [it]
|
||
header_rows.append(sorted(current, key=lambda t: -t["x"]))
|
||
|
||
person: dict = {}
|
||
request_number = None
|
||
request_date = None
|
||
applicant_name = None
|
||
|
||
for row in header_rows:
|
||
# In RTL layout the label is printed first (rightmost), value follows to its left.
|
||
# We scan each item; if it matches a known label we take the next item as the value.
|
||
i = 0
|
||
while i < len(row):
|
||
txt = row[i]["text"]
|
||
|
||
# Strip trailing colon if present
|
||
clean = txt.rstrip(":").strip()
|
||
|
||
if clean in _LABEL_MAP and i + 1 < len(row):
|
||
value = row[i + 1]["text"].rstrip(":").strip()
|
||
# Skip if the value looks like another label
|
||
if value not in _LABEL_MAP:
|
||
person[_LABEL_MAP[clean]] = value
|
||
i += 2
|
||
continue
|
||
|
||
# رقم الطلب
|
||
if "رقم الطلب" in txt and i + 1 < len(row):
|
||
request_number = row[i + 1]["text"].strip()
|
||
i += 2
|
||
continue
|
||
# تاريخ الطلب
|
||
if "تاريخ الطلب" in txt and i + 1 < len(row):
|
||
request_date = row[i + 1]["text"].strip()
|
||
i += 2
|
||
continue
|
||
# اسم المستدعي / الوكيل
|
||
if "المستدعي" in txt or "الوكيل" in txt:
|
||
if i + 1 < len(row):
|
||
applicant_name = row[i + 1]["text"].strip()
|
||
i += 2
|
||
continue
|
||
|
||
i += 1
|
||
|
||
# Also try full-text regex as fallback for request fields
|
||
full_text = " ".join(it["text"] for it in items)
|
||
if not request_number:
|
||
m = re.search(r"رقم الطلب[:\s]+([\d٠-٩]+)", full_text)
|
||
if m:
|
||
request_number = m.group(1)
|
||
if not request_date:
|
||
m = re.search(r"تاريخ الطلب[:\s]+([\d٠-٩\-/]+)", full_text)
|
||
if m:
|
||
request_date = m.group(1)
|
||
|
||
page_info = None
|
||
m_page = re.search(r"(صفحة\s*[\d٠-٩]+\s*من\s*[\d٠-٩]+|[\d٠-٩]+\s*/\s*[\d٠-٩]+)", full_text)
|
||
if m_page:
|
||
page_info = m_page.group(1).strip()
|
||
|
||
# Extract search scope (نطاق البحث / القضاء)
|
||
search_scope = None
|
||
for row in header_rows:
|
||
for j, item in enumerate(row):
|
||
if "القضاء" in item["text"] or "نطاق" in item["text"]:
|
||
# Value is the next item in the row
|
||
if j + 1 < len(row):
|
||
val = row[j + 1]["text"].rstrip(":").strip()
|
||
if val and "نطاق" not in val and "القضاء" not in val:
|
||
search_scope = val
|
||
break
|
||
if search_scope:
|
||
break
|
||
# Regex fallback
|
||
if not search_scope:
|
||
m_scope = re.search(r"القضاء[:\s]+([^\d٠-٩\-/]+?)(?:\s|$)", full_text)
|
||
if m_scope:
|
||
search_scope = m_scope.group(1).strip().rstrip(":")
|
||
|
||
# ── 4. Extract table body rows ────────────────────────────────────────────
|
||
table_items = [it for it in items if it["y"] > table_start_y + 15]
|
||
properties = _group_into_table_rows(table_items, col_boundaries)
|
||
|
||
return {
|
||
"request_number": request_number,
|
||
"request_date": request_date,
|
||
"applicant_name_raw": applicant_name,
|
||
"page_info": page_info,
|
||
"search_scope": search_scope,
|
||
"person": person if person else {"first_name": None},
|
||
"properties": properties,
|
||
"extraction_notes": "Extracted with EasyOCR (local). Please review carefully.",
|
||
}
|
||
|
||
|
||
def _group_into_table_rows(items: list[dict], col_boundaries: list) -> list[dict]:
|
||
"""
|
||
Group text items into table rows by Y-proximity, then assign each cell to
|
||
the correct column using col_boundaries (list of (x_left, x_right, key)).
|
||
Falls back to positional ordering when no boundaries are available.
|
||
"""
|
||
if not items:
|
||
return []
|
||
|
||
items_sorted = sorted(items, key=lambda t: t["y"])
|
||
rows: list[list[dict]] = []
|
||
current_row = [items_sorted[0]]
|
||
|
||
for item in items_sorted[1:]:
|
||
if abs(item["y"] - current_row[0]["y"]) < 14:
|
||
current_row.append(item)
|
||
else:
|
||
rows.append(current_row)
|
||
current_row = [item]
|
||
if current_row:
|
||
rows.append(current_row)
|
||
|
||
properties = []
|
||
key_order = [k for _, k in _COL_KEYS]
|
||
|
||
for row_items in rows:
|
||
texts = [c["text"] for c in row_items]
|
||
joined = " ".join(texts)
|
||
|
||
# Skip header row and single-word rows
|
||
if any(h in joined for h in _TABLE_HEADERS):
|
||
continue
|
||
if len(row_items) < 2:
|
||
continue
|
||
|
||
if col_boundaries:
|
||
# Assign each cell to a column by X-overlap with column boundaries
|
||
prop: dict = {k: None for _, k in _COL_KEYS}
|
||
for cell in row_items:
|
||
best_key = None
|
||
best_overlap = 0
|
||
for x_left, x_right, key in col_boundaries:
|
||
# Compute horizontal overlap between cell and column boundary
|
||
overlap = min(cell["x_right"], x_right) - max(cell["x_left"], x_left)
|
||
if overlap > best_overlap:
|
||
best_overlap = overlap
|
||
best_key = key
|
||
if best_key and best_overlap > 0:
|
||
# Concatenate if multiple cells map to same column
|
||
existing = prop.get(best_key)
|
||
prop[best_key] = (existing + " " + cell["text"]) if existing else cell["text"]
|
||
else:
|
||
# Fallback: positional right→left assignment
|
||
cells = sorted(row_items, key=lambda t: -t["x"])
|
||
prop = {key_order[i]: cells[i]["text"] if i < len(cells) else None
|
||
for i in range(len(key_order))}
|
||
|
||
properties.append(prop)
|
||
|
||
return properties
|
||
|
||
|
||
def _empty_result(note: str) -> dict:
|
||
return {
|
||
"request_number": None,
|
||
"request_date": None,
|
||
"applicant_name_raw": None,
|
||
"page_info": None,
|
||
"search_scope": None,
|
||
"person": {"first_name": None},
|
||
"properties": [],
|
||
"extraction_notes": note,
|
||
}
|
||
|
||
|
||
async def _extract_with_easyocr(image_path: str) -> dict:
|
||
import asyncio
|
||
|
||
full_path = _resolve_path(image_path)
|
||
|
||
def _call():
|
||
reader = _get_easyocr_reader()
|
||
results = reader.readtext(full_path)
|
||
return _parse_easyocr_results(results)
|
||
|
||
return await asyncio.get_event_loop().run_in_executor(None, _call)
|
||
|
||
|
||
# ─── Public API ───────────────────────────────────────────────────
|
||
|
||
async def extract_document(image_path: str, provider: str = "") -> dict:
|
||
"""
|
||
Extract structured data from a document image using the specified provider.
|
||
Falls back to the default provider if none specified.
|
||
provider can be: "claude", "easyocr", "gemini" (uses GEMINI_MODEL),
|
||
or a specific Gemini model ID like "gemini-2.5-flash".
|
||
"""
|
||
if not provider:
|
||
provider = get_default_provider()
|
||
|
||
if provider == "claude":
|
||
if not ANTHROPIC_API_KEY:
|
||
raise ValueError("ANTHROPIC_API_KEY not set")
|
||
return await _extract_with_claude(image_path)
|
||
elif provider == "easyocr":
|
||
return await _extract_with_easyocr(image_path)
|
||
elif provider == "gemini" or provider.startswith("gemini-"):
|
||
if not GEMINI_API_KEY:
|
||
raise ValueError("GEMINI_API_KEY not set")
|
||
# Pass the specific model if the provider ID encodes one
|
||
model_override = provider if provider.startswith("gemini-") else ""
|
||
return await _extract_with_gemini(image_path, model=model_override)
|
||
else:
|
||
raise ValueError(f"Unknown provider: {provider}")
|