from __future__ import annotations import hashlib import json import uuid from typing import Any from sqlalchemy.orm import Session from app.core.logging_config import get_logger from app.models.template import DocumentFormat, TemplateFingerprint from app.repositories.template_repository import TemplateFingerprintRepository, TemplateRepository logger = get_logger(__name__) class FingerprintService: """Generate and manage layout fingerprints for template matching.""" def __init__(self, db: Session) -> None: self.db = db self.fingerprint_repo = TemplateFingerprintRepository(db) self.template_repo = TemplateRepository(db) def generate_fingerprint(self, template: DocumentFormat) -> TemplateFingerprint: """Generate a layout fingerprint for a template.""" # Collect page dimensions page_dimensions = { "width": template.page_width, "height": template.page_height, "page_count": template.page_count, "margins": { "top": template.margin_top, "right": template.margin_right, "bottom": template.margin_bottom, "left": template.margin_left, }, } # Collect logo coordinates logo_coordinates = self._extract_logo_coordinates(template) # Collect header coordinates header_coordinates = self._extract_region_coordinates(template, "header") # Collect footer coordinates footer_coordinates = self._extract_region_coordinates(template, "footer") # Collect table coordinates table_coordinates = self._extract_table_coordinates(template) # Collect cell coordinates cell_coordinates = self._extract_cell_coordinates(template) # Compute fingerprint hash fingerprint_data = { "page_dimensions": page_dimensions, "logo_coordinates": logo_coordinates, "header_coordinates": header_coordinates, "footer_coordinates": footer_coordinates, "table_coordinates": table_coordinates, "cell_coordinates": cell_coordinates, } fingerprint_hash = self._compute_hash(fingerprint_data) # Check for existing fingerprint existing = self.fingerprint_repo.get_by_format_id(template.id) if existing: # Update existing existing.page_dimensions = page_dimensions existing.logo_coordinates = logo_coordinates existing.header_coordinates = header_coordinates existing.footer_coordinates = footer_coordinates existing.table_coordinates = table_coordinates existing.cell_coordinates = cell_coordinates existing.fingerprint_hash = fingerprint_hash self.db.flush() self.db.refresh(existing) return existing # Create new fingerprint fingerprint = self.fingerprint_repo.create_fingerprint( format_id=template.id, fingerprint_hash=fingerprint_hash, page_dimensions=page_dimensions, logo_coordinates=logo_coordinates, header_coordinates=header_coordinates, footer_coordinates=footer_coordinates, table_coordinates=table_coordinates, cell_coordinates=cell_coordinates, ) logger.info( "fingerprint_generated", template_id=str(template.id), hash=fingerprint_hash[:16], ) return fingerprint def _extract_logo_coordinates(self, template: DocumentFormat) -> dict[str, Any] | None: """Extract logo image coordinates from template.""" logos = [ir for ir in template.image_regions if ir.image_type == "logo"] if not logos: return None return { "items": [ { "page": ir.page_number, "x": ir.x, "y": ir.y, "width": ir.width, "height": ir.height, } for ir in logos ] } def _extract_region_coordinates( self, template: DocumentFormat, region_type: str, ) -> dict[str, Any] | None: """Extract coordinates for a specific region type.""" regions = [r for r in template.regions if r.region_type == region_type] if not regions: return None return { "items": [ { "page": r.page_number, "x": r.x, "y": r.y, "width": r.width, "height": r.height, } for r in regions ] } def _extract_table_coordinates(self, template: DocumentFormat) -> dict[str, Any] | None: """Extract table coordinates from template.""" if not template.table_formats: return None # Filter out dummy tables created by manual mappings (width=1000, height=1000) valid_tables = [tf for tf in template.table_formats if tf.width != 1000.0 and tf.height != 1000.0] if not valid_tables: return None return { "items": [ { "page": tf.page_number, "x": tf.x, "y": tf.y, "width": tf.width, "height": tf.height, "rows": tf.rows, "columns": tf.columns, } for tf in valid_tables ] } def _extract_cell_coordinates(self, template: DocumentFormat) -> dict[str, Any] | None: """Extract cell coordinates from template.""" # For manually mapped templates, prefer the actual mapped regions field_regions = [r for r in template.regions if r.region_type == "field_mapping"] if field_regions: return { "items": [ { "page": r.page_number, "x": r.x, "y": r.y, "width": r.width, "height": r.height, } for r in field_regions ] } # Otherwise fallback to template cells, but skip if they are all 0,0 (unmapped placeholders) if not template.cells: return None has_real_coords = any(c.width > 0 or c.height > 0 for c in template.cells) if not has_real_coords: return None return { "items": [ { "page": c.page_number, "x": c.x, "y": c.y, "width": c.width, "height": c.height, "row": c.row_no, "col": c.column_no, } for c in template.cells ] } def _compute_hash(self, data: dict[str, Any]) -> str: """Compute a deterministic hash of the fingerprint data.""" # Normalize coordinates to reduce sensitivity to minor variations normalized = self._normalize_coordinates(data) serialized = json.dumps(normalized, sort_keys=True, default=str) return hashlib.sha256(serialized.encode()).hexdigest() def _normalize_coordinates(self, data: Any) -> Any: """Normalize coordinates by rounding to reduce sensitivity to small variations.""" if isinstance(data, dict): return {k: self._normalize_coordinates(v) for k, v in data.items()} elif isinstance(data, list): return [self._normalize_coordinates(item) for item in data] elif isinstance(data, float): return round(data, 1) return data def compute_similarity( self, fingerprint1: TemplateFingerprint, fingerprint2_data: dict[str, Any], ) -> float: """Compute similarity score between a stored fingerprint and new document data.""" scores: list[float] = [] weights: list[float] = [] # Page dimensions similarity (high weight) dim_score = self._compare_dimensions( fingerprint1.page_dimensions, fingerprint2_data.get("page_dimensions"), ) scores.append(dim_score) weights.append(3.0) # Logo coordinates similarity if fingerprint1.logo_coordinates and fingerprint1.logo_coordinates.get("items"): logo_score = self._compare_coordinates( fingerprint1.logo_coordinates, fingerprint2_data.get("logo_coordinates"), ) scores.append(logo_score) weights.append(2.0) # Header coordinates similarity if fingerprint1.header_coordinates and fingerprint1.header_coordinates.get("items"): header_score = self._compare_coordinates( fingerprint1.header_coordinates, fingerprint2_data.get("header_coordinates"), ) scores.append(header_score) weights.append(2.0) # Footer coordinates similarity if fingerprint1.footer_coordinates and fingerprint1.footer_coordinates.get("items"): footer_score = self._compare_coordinates( fingerprint1.footer_coordinates, fingerprint2_data.get("footer_coordinates"), ) scores.append(footer_score) weights.append(1.5) # Table coordinates similarity if fingerprint1.table_coordinates and fingerprint1.table_coordinates.get("items"): table_score = self._compare_coordinates( fingerprint1.table_coordinates, fingerprint2_data.get("table_coordinates"), ) scores.append(table_score) weights.append(2.0) # Cell coordinates similarity if fingerprint1.cell_coordinates and fingerprint1.cell_coordinates.get("items"): cell_score = self._compare_coordinates( fingerprint1.cell_coordinates, fingerprint2_data.get("cell_coordinates"), ) scores.append(cell_score) weights.append(1.5) # Weighted average total_weight = sum(weights) if total_weight == 0: return 0.0 weighted_sum = sum(s * w for s, w in zip(scores, weights)) return weighted_sum / total_weight def _compare_dimensions( self, dims1: dict[str, Any] | None, dims2: dict[str, Any] | None, ) -> float: """Compare page dimensions similarity.""" if not dims1 or not dims2: return 0.0 if (dims1 or dims2) else 1.0 width_ratio = min(dims1.get("width", 0), dims2.get("width", 0)) / max( dims1.get("width", 1), dims2.get("width", 1) ) height_ratio = min(dims1.get("height", 0), dims2.get("height", 0)) / max( dims1.get("height", 1), dims2.get("height", 1) ) page_count_match = 1.0 if dims1.get("page_count") == dims2.get("page_count") else 0.5 return (width_ratio + height_ratio + page_count_match) / 3.0 def _compare_coordinates( self, coords1: dict[str, Any] | None, coords2: dict[str, Any] | None, ) -> float: """Compare coordinate sets for similarity.""" if not coords1 and not coords2: return 1.0 if not coords1 or not coords2: return 0.0 items1 = coords1.get("items", []) items2 = coords2.get("items", []) if not items1 and not items2: return 1.0 if not items1 or not items2: return 0.0 # Compare number of items count_ratio = min(len(items1), len(items2)) / max(len(items1), len(items2)) # Compare positions of matched items position_scores = [] for item1 in items1: best_match = 0.0 for item2 in items2: if item1.get("page") != item2.get("page"): continue score = self._compute_bbox_iou(item1, item2) best_match = max(best_match, score) position_scores.append(best_match) avg_position_score = sum(position_scores) / len(position_scores) if position_scores else 0.0 return (count_ratio + avg_position_score) / 2.0 def _compute_bbox_iou(self, bbox1: dict[str, Any], bbox2: dict[str, Any]) -> float: """Compute Intersection over Union for two bounding boxes.""" x1 = max(bbox1.get("x", 0), bbox2.get("x", 0)) y1 = max(bbox1.get("y", 0), bbox2.get("y", 0)) x2 = min( bbox1.get("x", 0) + bbox1.get("width", 0), bbox2.get("x", 0) + bbox2.get("width", 0), ) y2 = min( bbox1.get("y", 0) + bbox1.get("height", 0), bbox2.get("y", 0) + bbox2.get("height", 0), ) intersection = max(0, x2 - x1) * max(0, y2 - y1) area1 = bbox1.get("width", 0) * bbox1.get("height", 0) area2 = bbox2.get("width", 0) * bbox2.get("height", 0) union = area1 + area2 - intersection if union == 0: return 0.0 return intersection / union