218 lines
8.4 KiB
Python
218 lines
8.4 KiB
Python
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, TemplateMatch
|
|
from app.models.template import DocumentFormat
|
|
from app.repositories.document_repository import DocumentRepository, TemplateMatchRepository
|
|
from app.repositories.template_repository import TemplateFingerprintRepository, TemplateRepository
|
|
from app.services.fingerprint_service import FingerprintService
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class MatchingService:
|
|
"""Match documents against existing templates using fingerprint comparison."""
|
|
|
|
def __init__(self, db: Session) -> None:
|
|
self.db = db
|
|
self.doc_repo = DocumentRepository(db)
|
|
self.template_repo = TemplateRepository(db)
|
|
self.match_repo = TemplateMatchRepository(db)
|
|
self.fingerprint_repo = TemplateFingerprintRepository(db)
|
|
self.fingerprint_service = FingerprintService(db)
|
|
|
|
def match_document(
|
|
self,
|
|
document_id: uuid.UUID,
|
|
min_confidence: float = 0.75,
|
|
max_results: int = 5,
|
|
) -> list[TemplateMatch]:
|
|
"""Match a document against all existing templates."""
|
|
document = self.doc_repo.get_with_pages(document_id)
|
|
if not document:
|
|
raise ValueError(f"Document '{document_id}' not found")
|
|
|
|
if not document.pages:
|
|
raise ValueError(f"Document '{document_id}' has no processed pages")
|
|
|
|
# Generate document fingerprint data
|
|
doc_fingerprint_data = self._build_document_fingerprint(document)
|
|
|
|
# Get all templates with fingerprints
|
|
templates = self.template_repo.get_all_with_fingerprints()
|
|
fingerprints = self.fingerprint_repo.get_all_fingerprints()
|
|
|
|
# Map format_id -> fingerprint
|
|
fp_map = {fp.format_id: fp for fp in fingerprints}
|
|
|
|
matches: list[tuple[DocumentFormat, float, dict[str, Any]]] = []
|
|
|
|
for template in templates:
|
|
fp = fp_map.get(template.id)
|
|
if not fp:
|
|
continue
|
|
|
|
score = self.fingerprint_service.compute_similarity(fp, doc_fingerprint_data)
|
|
if score >= min_confidence:
|
|
match_details = {
|
|
"page_dimensions_score": self.fingerprint_service._compare_dimensions(
|
|
fp.page_dimensions, doc_fingerprint_data.get("page_dimensions")
|
|
),
|
|
"logo_score": self.fingerprint_service._compare_coordinates(
|
|
fp.logo_coordinates, doc_fingerprint_data.get("logo_coordinates")
|
|
),
|
|
"header_score": self.fingerprint_service._compare_coordinates(
|
|
fp.header_coordinates, doc_fingerprint_data.get("header_coordinates")
|
|
),
|
|
"footer_score": self.fingerprint_service._compare_coordinates(
|
|
fp.footer_coordinates, doc_fingerprint_data.get("footer_coordinates")
|
|
),
|
|
"table_score": self.fingerprint_service._compare_coordinates(
|
|
fp.table_coordinates, doc_fingerprint_data.get("table_coordinates")
|
|
),
|
|
"cell_score": self.fingerprint_service._compare_coordinates(
|
|
fp.cell_coordinates, doc_fingerprint_data.get("cell_coordinates")
|
|
),
|
|
}
|
|
matches.append((template, score, match_details))
|
|
|
|
# Sort by score descending
|
|
matches.sort(key=lambda x: x[1], reverse=True)
|
|
matches = matches[:max_results]
|
|
|
|
# Store match results
|
|
result_matches: list[TemplateMatch] = []
|
|
for idx, (template, score, details) in enumerate(matches):
|
|
template_match = self.match_repo.create_match(
|
|
document_id=document_id,
|
|
format_id=template.id,
|
|
confidence_score=score,
|
|
match_details=details,
|
|
selected=(idx == 0), # Auto-select best match
|
|
)
|
|
result_matches.append(template_match)
|
|
|
|
logger.info(
|
|
"document_matched",
|
|
document_id=str(document_id),
|
|
matches_found=len(result_matches),
|
|
best_score=result_matches[0].confidence_score if result_matches else 0.0,
|
|
)
|
|
|
|
return result_matches
|
|
|
|
def _build_document_fingerprint(self, document: Document) -> dict[str, Any]:
|
|
"""Build fingerprint data from a document for comparison."""
|
|
first_page = document.pages[0] if document.pages else None
|
|
|
|
page_dimensions = None
|
|
if first_page:
|
|
page_dimensions = {
|
|
"width": first_page.width,
|
|
"height": first_page.height,
|
|
"page_count": document.page_count or len(document.pages),
|
|
}
|
|
|
|
# Extract logo coordinates from images
|
|
logo_coordinates = None
|
|
logos = []
|
|
for page in document.pages:
|
|
for img in page.images:
|
|
if img.image_type == "logo":
|
|
logos.append({
|
|
"page": page.page_number,
|
|
"x": img.x,
|
|
"y": img.y,
|
|
"width": img.width,
|
|
"height": img.height,
|
|
})
|
|
if logos:
|
|
logo_coordinates = {"items": logos}
|
|
|
|
# Extract header coordinates
|
|
header_coordinates = None
|
|
headers = []
|
|
for page in document.pages:
|
|
header_blocks = [b for b in page.text_blocks if b.block_type == "header"]
|
|
if 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)
|
|
headers.append({
|
|
"page": page.page_number,
|
|
"x": min_x,
|
|
"y": min_y,
|
|
"width": max_x - min_x,
|
|
"height": max_y - min_y,
|
|
})
|
|
if headers:
|
|
header_coordinates = {"items": headers}
|
|
|
|
# Extract footer coordinates
|
|
footer_coordinates = None
|
|
footers = []
|
|
for page in document.pages:
|
|
footer_blocks = [b for b in page.text_blocks if b.block_type == "footer"]
|
|
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)
|
|
footers.append({
|
|
"page": page.page_number,
|
|
"x": min_x,
|
|
"y": min_y,
|
|
"width": max_x - min_x,
|
|
"height": max_y - min_y,
|
|
})
|
|
if footers:
|
|
footer_coordinates = {"items": footers}
|
|
|
|
# Extract table coordinates
|
|
table_coordinates = None
|
|
tables = []
|
|
for page in document.pages:
|
|
for table in page.tables:
|
|
tables.append({
|
|
"page": page.page_number,
|
|
"x": table.x,
|
|
"y": table.y,
|
|
"width": table.width,
|
|
"height": table.height,
|
|
"rows": table.rows,
|
|
"columns": table.columns,
|
|
})
|
|
if tables:
|
|
table_coordinates = {"items": tables}
|
|
|
|
# Extract cell coordinates from text blocks
|
|
cell_coordinates = None
|
|
cells = []
|
|
for page in document.pages:
|
|
for block in page.text_blocks:
|
|
if block.block_type == "text":
|
|
cells.append({
|
|
"page": page.page_number,
|
|
"x": block.x,
|
|
"y": block.y,
|
|
"width": block.width,
|
|
"height": block.height,
|
|
})
|
|
if cells:
|
|
cell_coordinates = {"items": cells}
|
|
|
|
return {
|
|
"page_dimensions": page_dimensions,
|
|
"logo_coordinates": logo_coordinates,
|
|
"header_coordinates": header_coordinates,
|
|
"footer_coordinates": footer_coordinates,
|
|
"table_coordinates": table_coordinates,
|
|
"cell_coordinates": cell_coordinates,
|
|
}
|