import logging from datetime import timedelta from decimal import Decimal from typing import Optional from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, and_, or_ from models.invoice import Invoice logger = logging.getLogger(__name__) class DuplicateDetector: """Service for detecting duplicate and related invoices.""" # Configuration thresholds DATE_TOLERANCE_DAYS = 3 AMOUNT_TOLERANCE_PERCENT = 5.0 def __init__(self, db: AsyncSession, kitchen_id: int): self.db = db self.kitchen_id = kitchen_id async def check_duplicates(self, invoice: Invoice) -> dict: """ Check for duplicates of the given invoice. Returns: { "firm_duplicate": Invoice or None, "possible_duplicates": list[Invoice], "related_documents": list[Invoice] } """ logger.info( f"Checking duplicates for invoice {invoice.id}: " f"invoice_number={invoice.invoice_number}, " f"supplier_id={invoice.supplier_id}, " f"date={invoice.invoice_date}, " f"total={invoice.total}" ) result = { "firm_duplicate": None, "possible_duplicates": [], "related_documents": [] } # 1. FIRM DUPLICATE: Same invoice_number (with same supplier if available) if invoice.invoice_number: firm = await self._find_firm_duplicate(invoice) if firm: logger.info(f"Found firm duplicate: invoice {firm.id} (number={firm.invoice_number})") result["firm_duplicate"] = firm else: logger.info(f"No firm duplicate found for invoice_number={invoice.invoice_number}") # 2. FUZZY/POSSIBLE DUPLICATE: Similar date + similar total (same supplier if available) if invoice.invoice_date and invoice.total: possible = await self._find_fuzzy_duplicates(invoice) result["possible_duplicates"] = possible # 3. RELATED DOCUMENTS: Cross-match by order_number if invoice.order_number: related = await self._find_related_documents(invoice) result["related_documents"] = related return result async def _find_firm_duplicate(self, invoice: Invoice) -> Optional[Invoice]: """Find exact match by invoice_number (same supplier if available)""" # Build conditions conditions = [ Invoice.kitchen_id == self.kitchen_id, Invoice.invoice_number == invoice.invoice_number, Invoice.id != invoice.id ] # If we have supplier_id, require same supplier for firm match # If no supplier_id, just match by invoice_number alone if invoice.supplier_id: conditions.append(Invoice.supplier_id == invoice.supplier_id) logger.debug(f"Searching for firm duplicate: invoice_number={invoice.invoice_number}, supplier_id={invoice.supplier_id}") else: logger.debug(f"Searching for firm duplicate (no supplier): invoice_number={invoice.invoice_number}") # Order by ID to get the oldest duplicate first, and use first() instead # of scalar_one_or_none() since there may be multiple duplicates query = select(Invoice).where(and_(*conditions)).order_by(Invoice.id) result = await self.db.execute(query) found = result.scalars().first() if not found: # Log what invoices exist with this number for debugging all_with_number = await self.db.execute( select(Invoice).where( Invoice.kitchen_id == self.kitchen_id, Invoice.invoice_number == invoice.invoice_number ) ) all_matches = list(all_with_number.scalars().all()) logger.debug(f"All invoices with number {invoice.invoice_number}: {[(i.id, i.supplier_id) for i in all_matches]}") return found async def _find_fuzzy_duplicates(self, invoice: Invoice) -> list[Invoice]: """Find similar invoices: close date + close total (same supplier if available)""" date_min = invoice.invoice_date - timedelta(days=self.DATE_TOLERANCE_DAYS) date_max = invoice.invoice_date + timedelta(days=self.DATE_TOLERANCE_DAYS) # Calculate amount tolerance amount_tolerance = invoice.total * Decimal(str(self.AMOUNT_TOLERANCE_PERCENT / 100)) amount_min = invoice.total - amount_tolerance amount_max = invoice.total + amount_tolerance # Build conditions conditions = [ Invoice.kitchen_id == self.kitchen_id, Invoice.id != invoice.id, Invoice.invoice_date.between(date_min, date_max), Invoice.total.between(amount_min, amount_max), # Exclude if it's the same invoice_number (already caught by firm) or_( Invoice.invoice_number == None, Invoice.invoice_number != invoice.invoice_number ) ] # If we have supplier_id, require same supplier for fuzzy match if invoice.supplier_id: conditions.append(Invoice.supplier_id == invoice.supplier_id) query = select(Invoice).where(and_(*conditions)) result = await self.db.execute(query) return list(result.scalars().all()) async def _find_related_documents(self, invoice: Invoice) -> list[Invoice]: """Find documents with same order_number but different invoice_number""" query = select(Invoice).where( and_( Invoice.kitchen_id == self.kitchen_id, Invoice.order_number == invoice.order_number, Invoice.id != invoice.id, # Must have different invoice_number to be related (not duplicate) or_( Invoice.invoice_number == None, Invoice.invoice_number != invoice.invoice_number ) ) ) result = await self.db.execute(query) return list(result.scalars().all()) def detect_document_type(raw_text: str, fields: dict) -> str: """ Detect if document is Invoice, Credit Note, or Delivery Note based on OCR text. Args: raw_text: Full OCR text fields: Extracted fields dict from Azure Returns: "invoice", "credit_note", or "delivery_note" """ if not raw_text: return "invoice" text_upper = raw_text.upper() # Check for credit note FIRST (highest priority) credit_keywords = [ "CREDIT NOTE", "CREDIT MEMO", "CR NOTE", "C/N", "CREDIT INVOICE", "CN NO", "CN:", "REFUND" ] # Check if invoice number contains credit note indicator invoice_number = fields.get("invoice_number") if invoice_number: inv_num_upper = str(invoice_number).upper() if any(kw in inv_num_upper for kw in ["CREDIT", "CR NOTE", "CN", "C/N"]): return "credit_note" # Check for credit note keywords in text if any(kw in text_upper for kw in credit_keywords): return "credit_note" # Check for negative total (strong indicator of credit note) total = fields.get("total") net_total = fields.get("net_total") if (total is not None and total < 0) or (net_total is not None and net_total < 0): return "credit_note" # Keywords suggesting delivery note dn_keywords = [ "DELIVERY NOTE", "DELIVERY DOCKET", "DISPATCH NOTE", "DELIVERY ADVICE", "PACKING SLIP", "PACKING LIST", "DN NO", "DN:", "D/N" ] # Keywords suggesting invoice inv_keywords = [ "TAX INVOICE", "VAT INVOICE", "INVOICE NO", "INVOICE DATE", "INVOICE TOTAL", "AMOUNT DUE", "PAYMENT DUE", "BALANCE DUE" ] dn_score = sum(1 for kw in dn_keywords if kw in text_upper) inv_score = sum(1 for kw in inv_keywords if kw in text_upper) # Also check if there's no total amount (delivery notes often don't have) if not fields.get("total"): dn_score += 1 return "delivery_note" if dn_score > inv_score else "invoice"