import re import json import logging from datetime import datetime from typing import Optional logger = logging.getLogger(__name__) def is_arabic_text(text: str) -> bool: """Check if text contains Arabic characters.""" arabic_pattern = re.compile(r'[\u0600-\u06FF\u0750-\u077F\u08A0-\u08FF]+') return bool(arabic_pattern.search(text)) def detect_document_type(text: str, key_value_pairs: list) -> str: """Detect document type based on content patterns.""" text_lower = text.lower() # Arabic keywords if any(kw in text for kw in ["فاتورة", "invoice", "إيصال"]): return "invoice" if any(kw in text for kw in ["عقد", "contract", "اتفاقية"]): return "contract" if any(kw in text for kw in ["تقرير", "report", "تحليل"]): return "report" if any(kw in text for kw in ["هوية", "جواز", "passport", "id card"]): return "identity_document" if any(kw in text for kw in ["شهادة", "certificate", "وثيقة"]): return "certificate" if len(key_value_pairs) > 5: return "form" return "general_document" def extract_dates(text: str) -> list: """Extract date patterns from Arabic/English text.""" patterns = [ r'\d{1,2}[/\-\.]\d{1,2}[/\-\.]\d{2,4}', r'\d{4}[/\-\.]\d{1,2}[/\-\.]\d{1,2}', r'\d{1,2}\s+(?:يناير|فبراير|مارس|أبريل|مايو|يونيو|يوليو|أغسطس|سبتمبر|أكتوبر|نوفمبر|ديسمبر)\s+\d{4}', r'\d{1,2}\s+(?:January|February|March|April|May|June|July|August|September|October|November|December)\s+\d{4}' ] dates = [] for pattern in patterns: matches = re.findall(pattern, text, re.IGNORECASE) dates.extend(matches) return list(set(dates)) def extract_numbers_and_amounts(text: str) -> dict: """Extract monetary amounts and important numbers.""" amounts = re.findall(r'[\$€£]\s*[\d,]+\.?\d*|[\d,]+\.?\d*\s*(?:USD|EUR|SAR|AED|EGP|ريال|درهم|دولار|جنيه)', text) numbers = re.findall(r'\b\d{4,}\b', text) return { "amounts": list(set(amounts)), "significant_numbers": list(set(numbers))[:10] } def extract_entities(text: str) -> dict: """Basic named entity extraction for Arabic/English documents.""" entities = { "emails": re.findall(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', text), "phone_numbers": re.findall( r'(?:\+?\d{1,3}[-.\s]?)?\(?\d{1,4}\)?[-.\s]?\d{1,4}[-.\s]?\d{1,9}', text ), "urls": re.findall(r'https?://[^\s]+|www\.[^\s]+', text), "arabic_names": [], "organizations": [] } # Simple Arabic name detection (honorifics) arabic_name_pattern = re.compile( r'(?:السيد|السيدة|الدكتور|الأستاذ|المهندس|د\.|أ\.)\s+[\u0600-\u06FF\s]{3,30}' ) entities["arabic_names"] = arabic_name_pattern.findall(text) # Limit phone numbers to reasonable length entities["phone_numbers"] = [ p for p in entities["phone_numbers"] if 7 <= len(re.sub(r'\D', '', p)) <= 15 ][:5] return entities def structure_extraction(raw_extraction: dict) -> dict: """ Convert raw Azure/Tesseract extraction into structured JSON. This is the main structuring function. """ try: full_text = raw_extraction.get("full_text", "") pages = raw_extraction.get("pages", []) tables = raw_extraction.get("tables", []) kv_pairs = raw_extraction.get("key_value_pairs", []) filename = raw_extraction.get("filename", "unknown") # Language detection has_arabic = is_arabic_text(full_text) language = "arabic" if has_arabic else "english" if has_arabic and re.search(r'[a-zA-Z]{3,}', full_text): language = "mixed" # Document type detection doc_type = detect_document_type(full_text, kv_pairs) # Extract structured elements dates = extract_dates(full_text) financials = extract_numbers_and_amounts(full_text) entities = extract_entities(full_text) # Build per-page summary page_summaries = [] for page in pages: lines = page.get("lines", []) page_text = " ".join(ln["text"] for ln in lines) page_summaries.append({ "page_number": page.get("page_number", 0), "line_count": len(lines), "word_count": len(page_text.split()), "has_arabic": is_arabic_text(page_text), "preview": page_text[:200] + "..." if len(page_text) > 200 else page_text }) # Build structured tables structured_tables = [] for table in tables: cells = table.get("cells", []) if not cells: continue rows = {} for cell in cells: row_idx = cell["row"] col_idx = cell["column"] if row_idx not in rows: rows[row_idx] = {} rows[row_idx][col_idx] = cell["text"] structured_tables.append({ "table_index": table.get("table_index", 0), "rows": table.get("row_count", 0), "columns": table.get("column_count", 0), "data": [rows[r] for r in sorted(rows.keys())] }) # Final structured document structured = { "document_id": f"doc_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}", "metadata": { "filename": filename, "file_size_bytes": raw_extraction.get("file_size_bytes", 0), "file_extension": raw_extraction.get("file_extension", ""), "processed_at": datetime.utcnow().isoformat(), "extraction_source": raw_extraction.get("source", "unknown"), "extraction_confidence": raw_extraction.get("confidence", 0.0) }, "document_analysis": { "document_type": doc_type, "language": language, "is_arabic": has_arabic, "total_pages": len(pages), "total_words": raw_extraction.get("word_count", 0), "total_tables": len(tables), "total_key_value_pairs": len(kv_pairs) }, "content": { "full_text": full_text, "page_summaries": page_summaries, "tables": structured_tables, "key_value_pairs": kv_pairs }, "extracted_entities": entities, "extracted_dates": dates, "financial_data": financials, "raw_stats": { "char_count": len(full_text), "line_count": full_text.count('\n'), "arabic_char_ratio": round( len(re.findall(r'[\u0600-\u06FF]', full_text)) / max(len(full_text), 1), 4 ) } } return structured except Exception as e: logger.error(f"Structuring failed: {e}") return { "document_id": "error", "error": str(e), "raw_text": raw_extraction.get("full_text", "") }