344 lines
12 KiB
Python
344 lines
12 KiB
Python
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
|
|
|
|
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 template.table_formats
|
|
]
|
|
}
|
|
|
|
def _extract_cell_coordinates(self, template: DocumentFormat) -> dict[str, Any] | None:
|
|
"""Extract cell coordinates from template."""
|
|
if not template.cells:
|
|
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: dict[str, Any]) -> dict[str, 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
|
|
logo_score = self._compare_coordinates(
|
|
fingerprint1.logo_coordinates,
|
|
fingerprint2_data.get("logo_coordinates"),
|
|
)
|
|
scores.append(logo_score)
|
|
weights.append(2.0)
|
|
|
|
# Header coordinates similarity
|
|
header_score = self._compare_coordinates(
|
|
fingerprint1.header_coordinates,
|
|
fingerprint2_data.get("header_coordinates"),
|
|
)
|
|
scores.append(header_score)
|
|
weights.append(2.0)
|
|
|
|
# Footer coordinates similarity
|
|
footer_score = self._compare_coordinates(
|
|
fingerprint1.footer_coordinates,
|
|
fingerprint2_data.get("footer_coordinates"),
|
|
)
|
|
scores.append(footer_score)
|
|
weights.append(1.5)
|
|
|
|
# Table coordinates similarity
|
|
table_score = self._compare_coordinates(
|
|
fingerprint1.table_coordinates,
|
|
fingerprint2_data.get("table_coordinates"),
|
|
)
|
|
scores.append(table_score)
|
|
weights.append(2.5)
|
|
|
|
# Cell coordinates similarity
|
|
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
|