Initial commit: Real Estate Registry app
This commit is contained in:
Georges Haddad
@@ -0,0 +1,40 @@
|
||||
import sqlite3
|
||||
import hashlib
|
||||
import secrets
|
||||
|
||||
def hash_password(password: str) -> str:
|
||||
salt = secrets.token_hex(16)
|
||||
hashed = hashlib.pbkdf2_hmac('sha256', password.encode('utf-8'), salt.encode('utf-8'), 100000).hex()
|
||||
return f"{salt}${hashed}"
|
||||
|
||||
def verify_password(stored_password: str, provided_password: str) -> bool:
|
||||
try:
|
||||
salt, stored_hash = stored_password.split('$')
|
||||
hashed = hashlib.pbkdf2_hmac('sha256', provided_password.encode('utf-8'), salt.encode('utf-8'), 100000).hex()
|
||||
return hashed == stored_hash
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
def get_user_by_username(username: str):
|
||||
from database.connection import get_db
|
||||
with get_db() as conn:
|
||||
cursor = conn.execute("SELECT * FROM users WHERE username = ?", (username,))
|
||||
return cursor.fetchone()
|
||||
|
||||
def create_user(username: str, password: str):
|
||||
from database.connection import get_db
|
||||
with get_db() as conn:
|
||||
hashed = hash_password(password)
|
||||
conn.execute("INSERT INTO users (username, password_hash) VALUES (?, ?)", (username, hashed))
|
||||
|
||||
def delete_user(user_id: int):
|
||||
from database.connection import get_db
|
||||
with get_db() as conn:
|
||||
conn.execute("DELETE FROM users WHERE id = ?", (user_id,))
|
||||
|
||||
def get_all_users():
|
||||
from database.connection import get_db
|
||||
with get_db() as conn:
|
||||
cursor = conn.execute("SELECT id, username, created_at FROM users ORDER BY created_at DESC")
|
||||
return cursor.fetchall()
|
||||
|
||||
@@ -0,0 +1,24 @@
|
||||
import os
|
||||
import sqlite3
|
||||
from datetime import datetime
|
||||
from config import DATABASE_PATH
|
||||
from database.connection import get_db
|
||||
|
||||
def create_backup() -> str:
|
||||
"""Create a backup of the database to a backup file."""
|
||||
db_file = DATABASE_PATH
|
||||
if not os.path.exists(db_file):
|
||||
raise FileNotFoundError("Database file does not exist.")
|
||||
|
||||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
backup_file = f"{db_file}.{timestamp}.bak"
|
||||
|
||||
# Safe SQLite backup
|
||||
with get_db() as conn:
|
||||
bck = sqlite3.connect(backup_file)
|
||||
with bck:
|
||||
conn.backup(bck)
|
||||
bck.close()
|
||||
|
||||
return backup_file
|
||||
|
||||
@@ -0,0 +1,551 @@
|
||||
import base64
|
||||
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. 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
|
||||
8. 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, 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)."
|
||||
)
|
||||
|
||||
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 get_available_providers() -> list[dict]:
|
||||
"""Return list of all available providers (EasyOCR is always available)."""
|
||||
providers = [
|
||||
{"id": "easyocr", "name": "EasyOCR (مجاني)", "model": "local"},
|
||||
]
|
||||
if ANTHROPIC_API_KEY:
|
||||
providers.append({"id": "claude", "name": "Claude (Anthropic)", "model": CLAUDE_MODEL})
|
||||
if GEMINI_API_KEY:
|
||||
providers.append({"id": "gemini", "name": "Gemini (Google)", "model": GEMINI_MODEL})
|
||||
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 "easyocr"
|
||||
ids = [p["id"] for p in providers]
|
||||
if DEFAULT_PROVIDER in ids:
|
||||
return DEFAULT_PROVIDER
|
||||
return ids[0]
|
||||
|
||||
|
||||
# ─── Claude extraction ───────────────────────────────────────────
|
||||
|
||||
async def _extract_with_claude(image_path: str) -> dict:
|
||||
import anthropic
|
||||
|
||||
client = anthropic.AsyncAnthropic(api_key=ANTHROPIC_API_KEY)
|
||||
full_path = _resolve_path(image_path)
|
||||
img_data, media_type = _encode_image(full_path)
|
||||
|
||||
message = await client.messages.create(
|
||||
model=CLAUDE_MODEL,
|
||||
max_tokens=4096,
|
||||
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},
|
||||
],
|
||||
}
|
||||
],
|
||||
)
|
||||
|
||||
raw_text = _strip_code_fences(message.content[0].text)
|
||||
return json.loads(raw_text)
|
||||
|
||||
|
||||
# ─── Gemini extraction ───────────────────────────────────────────
|
||||
|
||||
async def _extract_with_gemini(image_path: 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")
|
||||
|
||||
# Try primary model first; only fall back on 503 (overloaded), NOT on 429 (quota)
|
||||
models_to_try = [GEMINI_MODEL, "gemini-2.0-flash-lite"]
|
||||
|
||||
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,
|
||||
),
|
||||
)
|
||||
return response.text
|
||||
|
||||
last_error = None
|
||||
for model_name in models_to_try:
|
||||
for attempt in range(2):
|
||||
try:
|
||||
raw_text = await asyncio.get_event_loop().run_in_executor(None, lambda m=model_name: _call(m))
|
||||
raw_text = _strip_code_fences(raw_text)
|
||||
return json.loads(raw_text)
|
||||
except Exception as e:
|
||||
last_error = e
|
||||
err_str = str(e)
|
||||
# On quota exhaustion, skip to next model immediately (don't waste retries)
|
||||
if "429" in err_str or "RESOURCE_EXHAUSTED" in err_str:
|
||||
break
|
||||
# On temporary overload, wait and retry same model
|
||||
if "503" in err_str or "UNAVAILABLE" in err_str:
|
||||
await asyncio.sleep(3 * (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.
|
||||
"""
|
||||
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 == "gemini":
|
||||
if not GEMINI_API_KEY:
|
||||
raise ValueError("GEMINI_API_KEY not set")
|
||||
return await _extract_with_gemini(image_path)
|
||||
elif provider == "easyocr":
|
||||
return await _extract_with_easyocr(image_path)
|
||||
else:
|
||||
raise ValueError(f"Unknown provider: {provider}")
|
||||
@@ -0,0 +1,40 @@
|
||||
"""Convert PDF files to per-page images using PyMuPDF (no poppler dependency)."""
|
||||
|
||||
import uuid
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
|
||||
from config import UPLOAD_DIR
|
||||
|
||||
|
||||
def pdf_to_images(pdf_bytes: bytes, original_name: str) -> list[dict]:
|
||||
"""
|
||||
Convert a PDF to individual page images.
|
||||
Returns list of dicts: [{"image_path": "relative/path.png", "page_number": 1}, ...]
|
||||
"""
|
||||
import fitz # PyMuPDF
|
||||
|
||||
doc = fitz.open(stream=pdf_bytes, filetype="pdf")
|
||||
group_id = uuid.uuid4().hex
|
||||
today = date.today().isoformat()
|
||||
dest_dir = Path(UPLOAD_DIR) / today
|
||||
dest_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
pages = []
|
||||
for page_num in range(len(doc)):
|
||||
page = doc[page_num]
|
||||
# Render at 200 DPI for good OCR quality
|
||||
pix = page.get_pixmap(dpi=200)
|
||||
|
||||
filename = f"{group_id}_p{page_num + 1}.png"
|
||||
dest = dest_dir / filename
|
||||
pix.save(str(dest))
|
||||
|
||||
pages.append({
|
||||
"image_path": str(Path(today) / filename),
|
||||
"page_number": page_num + 1,
|
||||
"pdf_group_id": group_id,
|
||||
})
|
||||
|
||||
doc.close()
|
||||
return pages
|
||||
@@ -0,0 +1,152 @@
|
||||
from database.connection import get_db
|
||||
|
||||
|
||||
def normalize_arabic(text: str) -> str:
|
||||
"""Normalize Arabic text for search: collapse alef variants, ta marbuta, alef maqsura."""
|
||||
if not text:
|
||||
return text
|
||||
for ch in "أإآ":
|
||||
text = text.replace(ch, "ا")
|
||||
text = text.replace("ة", "ه")
|
||||
text = text.replace("ى", "ي")
|
||||
return text
|
||||
|
||||
|
||||
def search_persons(query: str) -> list[dict]:
|
||||
"""Search persons by name (first, family, or father name)."""
|
||||
norm = normalize_arabic(query.strip())
|
||||
pattern = f"%{norm}%"
|
||||
raw_pattern = f"%{query.strip()}%"
|
||||
|
||||
with get_db() as conn:
|
||||
rows = conn.execute(
|
||||
"""
|
||||
SELECT p.*,
|
||||
COUNT(DISTINCT pr.id) AS property_count,
|
||||
COUNT(DISTINCT d.id) AS document_count,
|
||||
GROUP_CONCAT(DISTINCT d.search_scope) AS search_scopes
|
||||
FROM persons p
|
||||
LEFT JOIN properties pr ON pr.person_id = p.id
|
||||
LEFT JOIN documents d ON d.person_id = p.id
|
||||
WHERE p.first_name_norm LIKE ?
|
||||
OR p.family_name_norm LIKE ?
|
||||
OR p.father_name LIKE ?
|
||||
OR p.first_name LIKE ?
|
||||
OR p.family_name LIKE ?
|
||||
GROUP BY p.id
|
||||
ORDER BY
|
||||
CASE WHEN p.first_name_norm = ? THEN 0
|
||||
WHEN p.family_name_norm = ? THEN 0
|
||||
ELSE 1 END,
|
||||
p.first_name
|
||||
""",
|
||||
(pattern, pattern, raw_pattern, raw_pattern, raw_pattern, norm, norm),
|
||||
).fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
|
||||
def search_properties(
|
||||
property_number: str = "",
|
||||
district: str = "",
|
||||
block: str = "",
|
||||
) -> list[dict]:
|
||||
"""Search properties by number, district, or block."""
|
||||
conditions = []
|
||||
params = []
|
||||
|
||||
if property_number:
|
||||
conditions.append("pr.property_number LIKE ?")
|
||||
params.append(f"{property_number}%")
|
||||
|
||||
if district:
|
||||
conditions.append("pr.real_estate_district LIKE ?")
|
||||
params.append(f"%{district}%")
|
||||
|
||||
if block:
|
||||
conditions.append("pr.block LIKE ?")
|
||||
params.append(f"%{block}%")
|
||||
|
||||
if not conditions:
|
||||
return []
|
||||
|
||||
where = " AND ".join(conditions)
|
||||
with get_db() as conn:
|
||||
rows = conn.execute(
|
||||
f"""
|
||||
SELECT pr.*,
|
||||
p.first_name, p.father_name, p.family_name, p.family_origin,
|
||||
d.search_scope
|
||||
FROM properties pr
|
||||
LEFT JOIN persons p ON p.id = pr.person_id
|
||||
LEFT JOIN documents d ON d.id = pr.document_id
|
||||
WHERE {where}
|
||||
ORDER BY pr.real_estate_district, pr.property_number
|
||||
""",
|
||||
params,
|
||||
).fetchall()
|
||||
|
||||
results = []
|
||||
for r in rows:
|
||||
row_dict = dict(r)
|
||||
if not row_dict.get("qaza") and row_dict.get("search_scope"):
|
||||
row_dict["qaza"] = row_dict["search_scope"]
|
||||
results.append(row_dict)
|
||||
return results
|
||||
|
||||
|
||||
def get_person_with_properties(person_id: int) -> dict | None:
|
||||
"""Fetch a person and all their properties."""
|
||||
with get_db() as conn:
|
||||
person = conn.execute(
|
||||
"SELECT * FROM persons WHERE id = ?", (person_id,)
|
||||
).fetchone()
|
||||
if not person:
|
||||
return None
|
||||
|
||||
props = conn.execute(
|
||||
"""
|
||||
SELECT pr.*, d.image_path, d.id AS document_id, d.search_scope
|
||||
FROM properties pr
|
||||
LEFT JOIN documents d ON d.id = pr.document_id
|
||||
WHERE pr.person_id = ?
|
||||
ORDER BY pr.real_estate_district, pr.row_order
|
||||
""",
|
||||
(person_id,),
|
||||
).fetchall()
|
||||
|
||||
docs = conn.execute(
|
||||
"SELECT id, image_path, request_number, request_date, status, page_info, search_scope FROM documents WHERE person_id = ?",
|
||||
(person_id,),
|
||||
).fetchall()
|
||||
|
||||
properties_list = []
|
||||
doc_ids_with_props = set()
|
||||
for p in props:
|
||||
p_dict = dict(p)
|
||||
if not p_dict.get("qaza") and p_dict.get("search_scope"):
|
||||
p_dict["qaza"] = p_dict["search_scope"]
|
||||
properties_list.append(p_dict)
|
||||
doc_ids_with_props.add(p_dict.get("document_id"))
|
||||
|
||||
# Add a "no properties" dummy row for documents that yielded no properties
|
||||
# so the search request is still recorded in the properties table
|
||||
docs_list = [dict(d) for d in docs]
|
||||
for d in docs_list:
|
||||
if d["id"] not in doc_ids_with_props:
|
||||
properties_list.append({
|
||||
"document_id": d["id"],
|
||||
"party_name": "-",
|
||||
"property_number": "-",
|
||||
"section": "-",
|
||||
"block": "-",
|
||||
"real_estate_district": "-",
|
||||
"qaza": d.get("search_scope") or "غير محدد",
|
||||
"num_shares": "-",
|
||||
"ownership_type": "لا يملك"
|
||||
})
|
||||
|
||||
return {
|
||||
"person": dict(person),
|
||||
"properties": properties_list,
|
||||
"documents": docs_list,
|
||||
}
|
||||
Reference in New Issue
Block a user