from __future__ import annotations import uuid from typing import Any from sqlalchemy.orm import Session from app.core.logging_config import get_logger from app.models.document import Document from app.models.template import ( DocumentCell, DocumentFormat, DocumentRegion, ImageRegion, TableColumn, TableFormat, TableRow, Watermark, ) from app.repositories.document_repository import DocumentRepository from app.repositories.template_repository import ( DocumentCellRepository, DocumentRegionRepository, ImageRegionRepository, TableColumnRepository, TableFormatRepository, TableRowRepository, TemplateRepository, WatermarkRepository, ) from app.services.fingerprint_service import FingerprintService logger = get_logger(__name__) class TemplateService: """Generate reusable document templates from processed documents.""" def __init__(self, db: Session) -> None: self.db = db self.doc_repo = DocumentRepository(db) self.template_repo = TemplateRepository(db) self.cell_repo = DocumentCellRepository(db) self.region_repo = DocumentRegionRepository(db) self.table_format_repo = TableFormatRepository(db) self.table_column_repo = TableColumnRepository(db) self.table_row_repo = TableRowRepository(db) self.watermark_repo = WatermarkRepository(db) self.image_region_repo = ImageRegionRepository(db) self.fingerprint_service = FingerprintService(db) def generate_template( self, document: Document, user_id: uuid.UUID | None = None, ) -> DocumentFormat: """Generate a reusable template from a processed document.""" if not document.pages: raise ValueError(f"Document '{document.id}' has no processed pages") # Check if template already exists for this document existing = self.template_repo.get_by_source_document(document.id) if existing: logger.info( "template_already_exists", document_id=str(document.id), template_id=str(existing.id), ) return existing first_page = document.pages[0] template_name = f"Template_{document.original_filename}_{uuid.uuid4().hex[:8]}" # Create template template = self.template_repo.create_template( name=template_name, page_width=first_page.width, page_height=first_page.height, page_count=document.page_count or len(document.pages), description=f"Auto-generated template from {document.original_filename}", source_document_id=document.id, created_by=user_id, is_active=False, ) # Process each page for page in document.pages: self._process_page_for_template(template, page) # Generate fingerprint self.fingerprint_service.generate_fingerprint(template) logger.info( "template_generated", template_id=str(template.id), document_id=str(document.id), cells=len(template.cells), regions=len(template.regions), ) return template def _process_page_for_template( self, template: DocumentFormat, page: Any, ) -> None: """Process a document page and create template components.""" page_number = page.page_number # Create cells from text blocks self._create_cells_from_text_blocks(template, page, page_number) # Create regions from headers, footers self._create_regions(template, page, page_number) # Create table formats self._create_table_formats(template, page, page_number) # Create image regions self._create_image_regions(template, page, page_number) # Detect watermarks from text blocks self._create_watermarks(template, page, page_number) def _create_cells_from_text_blocks( self, template: DocumentFormat, page: Any, page_number: int, ) -> None: """Create template cells from extracted text blocks.""" for seq, block in enumerate(page.text_blocks): if block.block_type in ("header", "footer", "watermark"): continue # Determine if this is a dynamic field is_dynamic = self._is_dynamic_field(block.text) field_name = self._generate_field_name(block.text, seq) if is_dynamic else None self.cell_repo.create_cell( format_id=template.id, page_number=page_number, x=block.x, y=block.y, width=block.width, height=block.height, data_type=self._infer_data_type(block.text), font_family=block.font_family, font_size=block.font_size, font_style=block.font_style, font_color=block.font_color, alignment=self._infer_alignment(block.x, template.page_width), static_text=block.text if not is_dynamic else None, field_name=field_name, sequence=seq, is_dynamic=is_dynamic, ) def _create_regions( self, template: DocumentFormat, page: Any, page_number: int, ) -> None: """Create template regions from headers and footers.""" header_blocks = [b for b in page.text_blocks if b.block_type == "header"] footer_blocks = [b for b in page.text_blocks if b.block_type == "footer"] if header_blocks: # Compute bounding box for all header blocks min_x = min(b.x for b in header_blocks) min_y = min(b.y for b in header_blocks) max_x = max(b.x + b.width for b in header_blocks) max_y = max(b.y + b.height for b in header_blocks) content = { "blocks": [ { "text": b.text, "x": b.x, "y": b.y, "width": b.width, "height": b.height, "font_family": b.font_family, "font_size": b.font_size, } for b in header_blocks ] } self.region_repo.create_region( format_id=template.id, page_number=page_number, region_type="header", x=min_x, y=min_y, width=max_x - min_x, height=max_y - min_y, content=content, sequence=0, ) if footer_blocks: min_x = min(b.x for b in footer_blocks) min_y = min(b.y for b in footer_blocks) max_x = max(b.x + b.width for b in footer_blocks) max_y = max(b.y + b.height for b in footer_blocks) content = { "blocks": [ { "text": b.text, "x": b.x, "y": b.y, "width": b.width, "height": b.height, "font_family": b.font_family, "font_size": b.font_size, } for b in footer_blocks ] } self.region_repo.create_region( format_id=template.id, page_number=page_number, region_type="footer", x=min_x, y=min_y, width=max_x - min_x, height=max_y - min_y, content=content, sequence=1, ) def _create_table_formats( self, template: DocumentFormat, page: Any, page_number: int, ) -> None: """Create table format definitions from detected tables.""" for table in page.tables: table_format = self.table_format_repo.create_table_format( format_id=template.id, page_number=page_number, x=table.x, y=table.y, width=table.width, height=table.height, rows=table.rows, columns=table.columns, ) # Create columns col_width = table.width / max(table.columns, 1) for col_idx in range(table.columns): self.table_column_repo.create_column( table_format_id=table_format.id, column_index=col_idx, width=col_width, data_type="text", alignment="left", ) # Create rows row_height = table.height / max(table.rows, 1) for row_idx in range(table.rows): self.table_row_repo.create_row( table_format_id=table_format.id, row_index=row_idx, height=row_height, is_header=(row_idx == 0), ) def _create_image_regions( self, template: DocumentFormat, page: Any, page_number: int, ) -> None: """Create image region definitions from detected images.""" for img in page.images: self.image_region_repo.create_image_region( format_id=template.id, page_number=page_number, x=img.x, y=img.y, width=img.width, height=img.height, image_path=img.image_path, image_type=img.image_type, is_static=True, ) def _create_watermarks( self, template: DocumentFormat, page: Any, page_number: int, ) -> None: """Create watermark definitions from detected watermark text blocks.""" watermark_blocks = [b for b in page.text_blocks if b.block_type == "watermark"] for block in watermark_blocks: self.watermark_repo.create_watermark( format_id=template.id, page_number=page_number, text=block.text, x=block.x, y=block.y, width=block.width, height=block.height, opacity=0.3, rotation=0.0, font_family=block.font_family, font_size=block.font_size, font_color=block.font_color or "#CCCCCC", ) def _is_dynamic_field(self, text: str) -> bool: """Determine if a text block represents a dynamic (variable) field.""" if not text: return False # Common patterns indicating dynamic content dynamic_patterns = [ "{{", "}}", "${", "##", "__________", "___", "...........", ] for pattern in dynamic_patterns: if pattern in text: return True # Short single-word values that might be labels are static # Longer values with numbers/dates tend to be dynamic import re # Date patterns if re.search(r"\d{1,2}[/\-\.]\d{1,2}[/\-\.]\d{2,4}", text): return True # Currency patterns if re.search(r"[$€£¥]\s*[\d,]+\.?\d*", text): return True # Phone patterns if re.search(r"\+?\d[\d\s\-()]{7,}", text): return True return False def _generate_field_name(self, text: str, sequence: int) -> str: """Generate a field name from text content.""" import re # Clean text clean = re.sub(r"[^a-zA-Z0-9\s]", "", text) clean = clean.strip().lower() words = clean.split()[:3] if words: return "_".join(words) return f"field_{sequence}" def _infer_data_type(self, text: str) -> str: """Infer the data type from text content.""" import re if not text: return "text" stripped = text.strip() # Number if re.match(r"^-?[\d,]+\.?\d*$", stripped.replace(",", "")): return "number" # Date if re.search(r"\d{1,2}[/\-\.]\d{1,2}[/\-\.]\d{2,4}", stripped): return "date" # Currency if re.search(r"^[$€£¥]\s*[\d,]+\.?\d*$", stripped): return "currency" # Email if re.search(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", stripped): return "email" return "text" def _infer_alignment(self, x: float, page_width: float) -> str: """Infer text alignment based on horizontal position.""" if page_width <= 0: return "left" relative_x = x / page_width if relative_x < 0.15: return "left" elif relative_x > 0.6: return "right" elif 0.35 < relative_x < 0.65: return "center" return "left"