""" Boundary Commission Document Service Handles document storage, indexing, and PDF conversion for boundary documents """ import os import shutil import tempfile from pathlib import Path from typing import Optional, Dict, List from datetime import datetime from PIL import Image import hashlib import uuid class BoundaryDocumentService: """Service for managing boundary commission documents""" def __init__(self, contentstore_base: str, db_config: dict): """ Initialize the boundary document service. Args: contentstore_base: Base path to contentstore directory db_config: Database configuration """ self.db_config = db_config # Try to use the configured contentstore_base/boundary path. # If it's not writable (e.g. /mnt on macOS), fall back to a temp directory, # but keep the same relative content_path structure so URLs still work. try: boundary_dir = os.path.join(contentstore_base, "boundary") os.makedirs(boundary_dir, exist_ok=True) self.contentstore_base = contentstore_base self.boundary_contentstore = boundary_dir except (OSError, PermissionError): temp_root = os.path.join(tempfile.gettempdir(), "boundary_contentstore_root") boundary_dir = os.path.join(temp_root, "boundary") os.makedirs(boundary_dir, exist_ok=True) self.contentstore_base = temp_root self.boundary_contentstore = boundary_dir print(f"⚠️ Could not create boundary contentstore at {os.path.join(contentstore_base, 'boundary')}") print(f" Using temporary directory instead: {boundary_dir}") # JPEG magic bytes (same as Aumentum) self.JPEG_SIGNATURES = [ b"\xFF\xD8\xFF\xE0", b"\xFF\xD8\xFF\xE1", b"\xFF\xD8\xFF\xE8", b"\xFF\xD8\xFF\xE2" ] def _get_db_connection(self): """Return database connection (MySQL or MSSQL).""" db_type = self.db_config.get("type", "mysql").lower() if db_type == "mysql": try: import pymysql host = self.db_config.get("host") or self.db_config.get("server", "localhost") port = self.db_config.get("port", 3306) database = self.db_config.get("database") username = self.db_config.get("username") or self.db_config.get("user") password = self.db_config.get("password") return pymysql.connect( host=host, port=port, user=username, password=password, database=database, charset='utf8mb4' ) except ImportError: import mysql.connector return mysql.connector.connect( host=self.db_config.get("host", "localhost"), port=self.db_config.get("port", 3306), user=self.db_config.get("username", "root"), password=self.db_config.get("password"), database=self.db_config.get("database") ) else: # MSSQL connection (if needed) raise NotImplementedError("MSSQL support not yet implemented for boundary documents") def generate_content_path(self, document_number: str, file_extension: str = ".bin") -> str: """ Generate contentstore path for a document (similar to Aumentum structure). Args: document_number: Unique document identifier file_extension: File extension (.bin, .pdf, etc.) Returns: Relative path like "boundary/2024/01/15/10/30/uuid.bin" """ now = datetime.now() year = str(now.year) month = str(now.month).zfill(2) day = str(now.day).zfill(2) hour = str(now.hour).zfill(2) minute = str(now.minute).zfill(2) # Generate unique filename unique_id = str(uuid.uuid4()) filename = f"{unique_id}{file_extension}" # Create path structure: boundary/YYYY/MM/DD/HH/MM/filename path = os.path.join("boundary", year, month, day, hour, minute, filename) return path.replace("\\", "/") # Use forward slashes def store_document(self, file_content: bytes, document_number: str, file_extension: str = ".bin") -> Dict: """ Store a document file in the contentstore. Args: file_content: File content as bytes document_number: Unique document identifier file_extension: File extension Returns: Dict with content_path and full_path """ # Generate hierarchical content path (e.g. boundary/YYYY/MM/DD/HH/MM/uuid.bin) content_path = self.generate_content_path(document_number, file_extension) # Full path is always built from contentstore_base + relative content_path full_path = os.path.join(self.contentstore_base, content_path) # Ensure directory structure exists os.makedirs(os.path.dirname(full_path), exist_ok=True) # Write file with open(full_path, 'wb') as f: f.write(file_content) return { "content_path": content_path, "full_path": full_path, "file_size": len(file_content) } def generate_document_number(self) -> str: """ Generate a unique document number. Format: BND-DOC-YYYYMMDD-HHMMSS-XXXX Where XXXX is a random 4-digit number for uniqueness. Returns: Unique document number string """ now = datetime.now() date_str = now.strftime("%Y%m%d") time_str = now.strftime("%H%M%S") random_suffix = str(uuid.uuid4().int % 10000).zfill(4) return f"BND-DOC-{date_str}-{time_str}-{random_suffix}" def index_document(self, document_number: Optional[str], document_type: str, title: str, file_path: str, related_boundary_id: Optional[int] = None, related_dispute_id: Optional[int] = None, related_survey_id: Optional[int] = None, related_treaty_id: Optional[int] = None, description: Optional[str] = None, uploaded_by: Optional[str] = None, visibility: str = "open", pseudo_document_number: Optional[str] = None) -> Dict: """ Index a document in the database. Args: document_number: Unique document identifier (auto-generated if None) document_type: Type of document (evidence, treaty, survey_report, etc.) title: Document title file_path: Path to the file (will be moved to contentstore) related_boundary_id: Related boundary ID related_dispute_id: Related dispute ID related_survey_id: Related survey ID related_treaty_id: Related treaty ID description: Document description uploaded_by: User who uploaded the document Returns: Dict with document information """ # Auto-generate document number if not provided if not document_number: document_number = self.generate_document_number() # Normalize pseudo document number (optional) if pseudo_document_number: pseudo_document_number = pseudo_document_number.strip() or None conn = self._get_db_connection() cursor = conn.cursor() try: # Read file content with open(file_path, 'rb') as f: file_content = f.read() # Store in contentstore store_result = self.store_document(file_content, document_number) content_path = store_result["content_path"] # Normalize visibility visibility = (visibility or "open").lower() if visibility not in ("open", "secret"): visibility = "open" # Get file info file_size = len(file_content) file_name = os.path.basename(file_path) mime_type = self._detect_mime_type(file_content) page_count = self._count_pages(file_content, mime_type) # Insert into database cursor.execute(""" INSERT INTO boundary_document (document_number, pseudo_document_number, document_type, title, description, content_path, file_name, file_size, mime_type, page_count, uploaded_by, visibility, related_boundary_id, related_dispute_id, related_survey_id, related_treaty_id) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( document_number, pseudo_document_number, document_type, title, description, content_path, file_name, file_size, mime_type, page_count, uploaded_by, visibility, related_boundary_id, related_dispute_id, related_survey_id, related_treaty_id )) document_id = cursor.lastrowid conn.commit() return { "id": document_id, "document_number": document_number, "content_path": content_path, "file_size": file_size, "page_count": page_count, "mime_type": mime_type } finally: cursor.close() conn.close() def _detect_mime_type(self, content: bytes) -> str: """Detect MIME type from file content.""" if content.startswith(b"%PDF"): return "application/pdf" elif any(content.startswith(sig) for sig in self.JPEG_SIGNATURES): return "image/jpeg" elif content.startswith(b"\x89PNG"): return "image/png" else: return "application/octet-stream" def _count_pages(self, content: bytes, mime_type: str) -> int: """Count pages in a document.""" if mime_type == "application/pdf": try: # Simple PDF page count (count "%%EOF" markers) return content.count(b"%%EOF") except: return 1 elif mime_type in ["image/jpeg", "image/png"]: return 1 else: return 1 def convert_bin_to_pdf(self, bin_path: str, output_pdf_path: Optional[str] = None) -> Optional[str]: """ Convert .bin file (LEADTOOLS JPEG) to PDF (same as Aumentum). Args: bin_path: Path to .bin file output_pdf_path: Optional output path Returns: Path to created PDF file, or None if conversion failed """ try: if output_pdf_path is None: output_pdf_path = bin_path.replace(".bin", ".pdf") # Ensure output directory exists os.makedirs(os.path.dirname(output_pdf_path), exist_ok=True) # Read .bin file and extract JPEG with open(bin_path, 'rb') as f: content = f.read() # Find JPEG start jpeg_start = None for sig in self.JPEG_SIGNATURES: pos = content.find(sig) if pos != -1: jpeg_start = pos break if jpeg_start is None: print(f"⚠️ No JPEG signature found in {bin_path}") return None # Extract JPEG data jpeg_data = content[jpeg_start:] # Convert to PDF using PIL from io import BytesIO img = Image.open(BytesIO(jpeg_data)) rgb = img.convert('RGB') rgb.save(output_pdf_path, "PDF", resolution=100.0) return output_pdf_path except Exception as e: print(f"Error converting {bin_path} to PDF: {e}") return None def convert_image_to_pdf(self, image_path: str, output_pdf_path: Optional[str] = None, mime_type: Optional[str] = None) -> Optional[str]: """ Convert image file (PNG, JPEG, etc.) to PDF. Args: image_path: Path to image file output_pdf_path: Optional output path mime_type: MIME type of the image (auto-detected if None) Returns: Path to created PDF file, or None if conversion failed """ try: if not os.path.exists(image_path): print(f"⚠️ Image file not found: {image_path}") return None if output_pdf_path is None: output_pdf_path = image_path.replace(".png", ".pdf").replace(".jpg", ".pdf").replace(".jpeg", ".pdf") # Ensure output directory exists os.makedirs(os.path.dirname(output_pdf_path), exist_ok=True) # Open image with PIL img = Image.open(image_path) # Convert to RGB if necessary (PNG with transparency, etc.) if img.mode in ('RGBA', 'LA', 'P'): # Create white background for transparent images rgb = Image.new('RGB', img.size, (255, 255, 255)) if img.mode == 'P': img = img.convert('RGBA') rgb.paste(img, mask=img.split()[-1] if img.mode in ('RGBA', 'LA') else None) elif img.mode != 'RGB': rgb = img.convert('RGB') else: rgb = img # Save as PDF rgb.save(output_pdf_path, "PDF", resolution=100.0) print(f"✅ Converted {image_path} to PDF: {output_pdf_path}") return output_pdf_path except Exception as e: print(f"❌ Error converting {image_path} to PDF: {e}") import traceback traceback.print_exc() return None def fetch_pdf_by_document_number(self, document_number: str) -> Dict: """ Fetch and convert boundary document to PDF. Args: document_number: Document number Returns: Dict with pdf_path, document_id, page_count, and error info """ result = { "document_number": document_number, "pdf_path": None, "document_id": None, "page_count": 0, "error": None } try: conn = self._get_db_connection() cursor = conn.cursor() # Get document from database cursor.execute(""" SELECT id, content_path, page_count, mime_type FROM boundary_document WHERE document_number = %s LIMIT 1 """, (document_number,)) doc = cursor.fetchone() if not doc: result["error"] = f"Document not found: {document_number}" cursor.close() conn.close() return result document_id, content_path, page_count, mime_type = doc result["document_id"] = document_id result["page_count"] = page_count or 0 # Build full path full_path = os.path.join(self.contentstore_base, content_path) if not os.path.exists(full_path): result["error"] = f"File not found: {full_path}" cursor.close() conn.close() return result # Handle different file types if mime_type == "application/pdf": # Already PDF, use directly result["pdf_path"] = full_path elif mime_type in ["image/jpeg", "image/png"] or full_path.endswith((".bin", ".jpg", ".jpeg", ".png")): # Convert image to PDF # Use temp directory for PDF conversion to avoid permission issues import tempfile temp_dir = os.getenv("TEMP_PDF_DIR", "/tmp/boundary_pdfs") os.makedirs(temp_dir, exist_ok=True) # Generate temp PDF path base_name = os.path.splitext(os.path.basename(full_path))[0] pdf_path = os.path.join(temp_dir, f"{base_name}_{document_id}.pdf") # Check if PDF already exists if os.path.exists(pdf_path): result["pdf_path"] = pdf_path else: # Convert image to PDF converted = self.convert_image_to_pdf(full_path, pdf_path, mime_type) if converted: result["pdf_path"] = converted else: result["error"] = "PDF conversion failed" else: result["error"] = f"Unsupported file type: {mime_type}" cursor.close() conn.close() return result except Exception as e: result["error"] = str(e) return result