""" Hermes Tools — Unified compliance pipeline and engine access. This module provides: - `CompliancePipeline`: Thread-safe orchestrator for PDF parsing, HS classification, and sanctions screening in a single call. - Re-exports from `base` for backward compatibility. Usage: from hermes.tools import CompliancePipeline, get_pipeline pipeline = get_pipeline() review_item = pipeline.process_incoming_document( file_path="invoice.pdf", context={"shipper": "Acme Corp", "consignee": "Global Imports LLC"}, ) """ from __future__ import annotations import logging import tempfile import threading from pathlib import Path from typing import Annotated, Any, Final, Optional from hermes.tools.base.registry import tool_registry from hermes.tools.base.tool import BaseTool logger: Final = logging.getLogger(__name__) # ── Domain Exceptions ───────────────────────────────────────────────── class PipelineError(Exception): """Base exception for pipeline execution errors.""" class PipelineValidationError(PipelineError): """Raised when pipeline input validation fails. Attributes: field: The field that failed validation. reason: Description of the validation failure. """ def __init__(self, field: str, reason: str) -> None: self.field = field self.reason = reason super().__init__(f"Validation failed for '{field}': {reason}") class PipelinePDFParsingError(PipelineError): """Raised when PDF parsing fails during pipeline execution. Attributes: file_path: Path to the PDF that failed parsing. original_error: The underlying parsing exception. """ def __init__(self, file_path: str, original_error: Exception) -> None: self.file_path = file_path self.original_error = original_error super().__init__(f"PDF parsing failed for {file_path}: {original_error}") class PipelineClassificationError(PipelineError): """Raised when HS code classification fails. Attributes: description: The product description that failed classification. original_error: The underlying classification exception. """ def __init__(self, description: str, original_error: Exception) -> None: self.description = description self.original_error = original_error super().__init__( f"HS classification failed for '{description[:50]}...': {original_error}" ) class PipelineSanctionsError(PipelineError): """Raised when sanctions screening fails. Attributes: party_name: The party name that failed screening. original_error: The underlying screening exception. """ def __init__(self, party_name: str, original_error: Exception) -> None: self.party_name = party_name self.original_error = original_error super().__init__( f"Sanctions screening failed for '{party_name}': {original_error}" ) # ── Constants ───────────────────────────────────────────────────────── MAX_TEMP_FILE_SIZE: Final[int] = 50 * 1024 * 1024 # 50 MB # ── Compliance Pipeline ─────────────────────────────────────────────── class CompliancePipeline: """Thread-safe pipeline orchestrating PDF parsing, HS classification, and sanctions screening. Provides a single method `process_incoming_document` that: 1. Parses the PDF to extract document fields 2. Classifies extracted items to HS codes 3. Screens all parties against sanctions lists 4. Assembles a ReviewItem for human approval All errors are wrapped in pipeline-specific exceptions for consistent error handling upstream. Attributes: _parser: PDF parser singleton. _hs_engine: HS classification engine singleton. _sanctions_engine: Sanctions screening engine singleton. """ def __init__(self) -> None: """Initialize the pipeline with engine singletons.""" from hermes.tools.pdf_parser import get_parser from hermes.tools.hs_classifier import get_engine as get_hs_engine from hermes.tools.sanctions import get_engine as get_sanctions_engine self._parser = get_parser() self._hs_engine = get_hs_engine() self._sanctions_engine = get_sanctions_engine() self._lock: threading.Lock = threading.Lock() def process_incoming_document( self, file_path: str, context: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: """Process a trade document through the full compliance pipeline. Orchestrates PDF parsing → HS classification → sanctions screening and returns a ReviewItem dict ready for human review. Args: file_path: Path to the PDF document to process. context: Optional context dict with overrides: - shipper (str): Override shipper name. - consignee (str): Override consignee name. - country_origin (str): Override country of origin. - country_destination (str): Override destination country. - invoice_number (str): Override invoice number. Returns: Dict representing a ReviewItem with all extracted and classified data. Raises: FileNotFoundError: If the file does not exist. PipelineValidationError: If input validation fails. PipelinePDFParsingError: If PDF parsing fails. PipelineClassificationError: If HS classification fails. PipelineSanctionsError: If sanctions screening fails. """ ctx: dict[str, Any] = context or {} path: Path = Path(file_path) if not path.exists(): raise FileNotFoundError(f"Document not found: {file_path}") if not path.suffix.lower() == ".pdf": raise PipelineValidationError( "file_path", f"Expected PDF file, got {path.suffix or 'no extension'}", ) # Step 1: Parse PDF parsed_doc = self._parse_pdf(file_path) # Step 2: Extract and classify items classifications: list[dict[str, Any]] = self._classify_items( parsed_doc, ctx ) # Step 3: Screen parties sanctions_results: dict[str, Any] = self._screen_parties(parsed_doc, ctx) # Step 4: Assemble ReviewItem review_item: dict[str, Any] = self._assemble_review_item( file_path=file_path, parsed_doc=parsed_doc, classifications=classifications, sanctions_results=sanctions_results, context=ctx, ) logger.info( "Pipeline processed %s → review item %s", file_path, review_item.get("item_id", "pending"), ) return review_item def _parse_pdf(self, file_path: str) -> Any: """Parse the PDF and extract document fields. Args: file_path: Path to the PDF file. Returns: ParsedDocument with extracted fields. Raises: PipelinePDFParsingError: If parsing fails. """ try: from hermes.tools.pdf_parser import ( PDFSizeLimitExceededError, PDFPageLimitExceededError, PDFNotValidError, PDFEmptyError, ) return self._parser.parse(file_path) except (PDFSizeLimitExceededError, PDFPageLimitExceededError) as exc: raise PipelinePDFParsingError(file_path, exc) from exc except (PDFNotValidError, PDFEmptyError) as exc: raise PipelinePDFParsingError(file_path, exc) from exc except FileNotFoundError: raise except Exception as exc: raise PipelinePDFParsingError(file_path, exc) from exc def _classify_items(self, parsed_doc: Any, ctx: dict[str, Any]) -> list[dict[str, Any]]: """Classify extracted items to HS codes. Args: parsed_doc: ParsedDocument from PDF parsing. ctx: Context dict with optional overrides. Returns: List of classification result dicts. """ from hermes.tools.hs_classifier import HSClassificationHallucinationError from hermes.tools.pdf_parser import InvoiceItem classifications: list[dict[str, Any]] = [] items: list[InvoiceItem] = getattr(parsed_doc, "items", []) if not items: # If no items extracted, classify based on document description description: str = self._build_description_from_doc(parsed_doc, ctx) if description: cls_result = self._classify_single(description, ctx) if cls_result: classifications.append(cls_result) return classifications country_origin: str = ctx.get( "country_origin", getattr( parsed_doc.country_origin, "value", "" ) if parsed_doc.country_origin else "", ) country_dest: str = ctx.get( "country_destination", getattr( parsed_doc.country_destination, "value", "" ) if parsed_doc.country_destination else "", ) for item in items: if not item.description: continue try: response = self._hs_engine.classify( description=item.description, country_origin=country_origin, country_destination=country_dest, use_llm=False, target_digits=6, ) alt_list: list[dict[str, Any]] = [] for alt in response.alternatives: alt_list.append({ "hs_code": alt.hs_code, "description": alt.description, "confidence": round(alt.confidence, 3), }) classifications.append({ "description": item.description, "hs_code": response.primary.hs_code, "hs_description": response.primary.description, "confidence": round(response.primary.confidence, 3), "source": response.primary.source, "needs_human_review": response.primary.needs_human_review, "alternatives": alt_list, "quantity": item.quantity, "unit_price": item.unit_price, "total_price": item.total_price, }) except HSClassificationHallucinationError as exc: logger.warning( "Classification hallucination detected for '%s': %s", item.description[:50], exc, ) classifications.append({ "description": item.description, "hs_code": "999999", "hs_description": "Classification failed - hallucination detected", "confidence": 0.0, "source": "error", "needs_human_review": True, "alternatives": [], "quantity": item.quantity, "unit_price": item.unit_price, "total_price": item.total_price, }) except Exception as exc: logger.warning( "Classification failed for '%s': %s", item.description[:50], exc, ) classifications.append({ "description": item.description, "hs_code": "", "hs_description": f"Classification error: {exc}", "confidence": 0.0, "source": "error", "needs_human_review": True, "alternatives": [], "quantity": item.quantity, "unit_price": item.unit_price, "total_price": item.total_price, }) return classifications def _classify_single( self, description: str, ctx: dict[str, Any] ) -> Optional[dict[str, Any]]: """Classify a single description. Args: description: Product description. ctx: Context dict with optional overrides. Returns: Classification result dict, or None if classification fails. """ try: country_origin: str = ctx.get("country_origin", "") country_dest: str = ctx.get("country_destination", "") response = self._hs_engine.classify( description=description, country_origin=country_origin, country_destination=country_dest, use_llm=False, target_digits=6, ) alt_list: list[dict[str, Any]] = [] for alt in response.alternatives: alt_list.append({ "hs_code": alt.hs_code, "description": alt.description, "confidence": round(alt.confidence, 3), }) return { "description": description, "hs_code": response.primary.hs_code, "hs_description": response.primary.description, "confidence": round(response.primary.confidence, 3), "source": response.primary.source, "needs_human_review": response.primary.needs_human_review, "alternatives": alt_list, "quantity": 0, "unit_price": 0, "total_price": 0, } except Exception as exc: logger.warning("Single classification failed: %s", exc) return None def _build_description_from_doc( self, parsed_doc: Any, ctx: dict[str, Any] ) -> str: """Build a product description from parsed document fields. Args: parsed_doc: ParsedDocument. ctx: Context dict. Returns: Description string for classification. """ parts: list[str] = [] if parsed_doc.document_type and parsed_doc.document_type != "unknown": parts.append(parsed_doc.document_type.replace("_", " ").title()) if ctx.get("description"): parts.append(ctx["description"]) if ctx.get("product_description"): parts.append(ctx["product_description"]) return " ".join(parts) if parts else "" def _screen_parties( self, parsed_doc: Any, ctx: dict[str, Any] ) -> dict[str, Any]: """Screen all parties against sanctions lists. Args: parsed_doc: ParsedDocument with party information. ctx: Context dict with optional overrides. Returns: Dict mapping party names to their screening results. """ from hermes.tools.sanctions import ScreeningResult parties: dict[str, str] = {} shipper: str = ctx.get("shipper", "") if not shipper and parsed_doc.shipper: shipper = getattr(parsed_doc.shipper, "name", "") if shipper: parties["shipper"] = shipper consignee: str = ctx.get("consignee", "") if not consignee and parsed_doc.consignee: consignee = getattr(parsed_doc.consignee, "name", "") if consignee: parties["consignee"] = consignee notify_party: str = ctx.get("notify_party", "") if not notify_party and parsed_doc.notify_party: notify_party = getattr(parsed_doc.notify_party, "name", "") if notify_party: parties["notify_party"] = notify_party results: dict[str, Any] = {} for role, name in parties.items(): try: result: ScreeningResult = self._sanctions_engine.screen(name) matches_data: list[dict[str, Any]] = [] for match in result.matches: matches_data.append({ "entity_name": match.entity.name, "confidence": round(match.confidence, 3), "match_type": match.match_type, "matched_name": match.matched_name, "source": match.source, }) results[role] = { "name": name, "risk_level": result.risk_level, "has_matches": result.has_matches, "matches": matches_data, "sources_checked": result.sources_checked, } except Exception as exc: logger.warning( "Sanctions screening failed for %s '%s': %s", role, name[:50], exc, ) results[role] = { "name": name, "risk_level": "unknown", "has_matches": False, "matches": [], "sources_checked": [], "error": str(exc), } return results def _assemble_review_item( self, file_path: str, parsed_doc: Any, classifications: list[dict[str, Any]], sanctions_results: dict[str, Any], context: dict[str, Any], ) -> dict[str, Any]: """Assemble a ReviewItem dict from pipeline results. Args: file_path: Original PDF file path. parsed_doc: ParsedDocument from PDF parsing. classifications: List of HS classification results. sanctions_results: Dict of sanctions screening results. context: Context dict with overrides. Returns: Dict representing a ReviewItem. """ invoice_number: str = context.get("invoice_number", "") if not invoice_number and parsed_doc.invoice_number: invoice_number = getattr(parsed_doc.invoice_number, "value", "") invoice_date: str = context.get("invoice_date", "") if not invoice_date and parsed_doc.invoice_date: invoice_date = getattr(parsed_doc.invoice_date, "value", "") total_amount: str = context.get("total_amount", "") if not total_amount and parsed_doc.total_amount: total_amount = getattr(parsed_doc.total_amount, "value", "") shipper: str = context.get("shipper", "") if not shipper and parsed_doc.shipper: shipper = getattr(parsed_doc.shipper, "name", "") consignee: str = context.get("consignee", "") if not consignee and parsed_doc.consignee: consignee = getattr(parsed_doc.consignee, "name", "") country_origin: str = context.get("country_origin", "") if not country_origin and parsed_doc.country_origin: country_origin = getattr(parsed_doc.country_origin, "value", "") country_dest: str = context.get("country_destination", "") if not country_dest and parsed_doc.country_destination: country_dest = getattr(parsed_doc.country_destination, "value", "") primary_hs: str = "" hs_description: str = "" hs_confidence: float = 0.0 hs_alternatives: list[dict[str, Any]] = [] needs_human_review: bool = False if classifications: best: dict[str, Any] = max( classifications, key=lambda c: c.get("confidence", 0) ) primary_hs = best.get("hs_code", "") hs_description = best.get("hs_description", "") hs_confidence = best.get("confidence", 0.0) needs_human_review = best.get("needs_human_review", False) hs_alternatives = best.get("alternatives", []) sanctions_risk: str = "clear" all_sanctions_matches: list[dict[str, Any]] = [] for role, result in sanctions_results.items(): level: str = result.get("risk_level", "clear") if level in ("high", "blocked"): sanctions_risk = level for match in result.get("matches", []): all_sanctions_matches.append({ "role": role, **match, }) if needs_human_review or sanctions_risk in ("high", "blocked"): priority: int = 2 if sanctions_risk == "blocked" else 1 else: priority = 0 return { "item_id": "", "document_path": file_path, "document_type": parsed_doc.document_type or "unknown", "invoice_number": invoice_number, "invoice_date": invoice_date, "total_amount": total_amount, "shipper": shipper, "consignee": consignee, "country_origin": country_origin, "country_destination": country_dest, "hs_code_suggested": primary_hs, "hs_code_description": hs_description, "hs_code_confidence": hs_confidence, "hs_code_alternatives": hs_alternatives, "sanctions_risk_level": sanctions_risk, "sanctions_matches": all_sanctions_matches, "status": "pending", "assigned_to": "", "priority": priority, "comments": [], "actions": [], "final_hs_code": "", "reviewer_notes": "", } # ── Singleton ───────────────────────────────────────────────────────── _pipeline: Optional[CompliancePipeline] = None _pipeline_lock: threading.Lock = threading.Lock() def get_pipeline() -> CompliancePipeline: """Get or create the global compliance pipeline (thread-safe). Returns: The singleton CompliancePipeline instance. """ global _pipeline if _pipeline is None: with _pipeline_lock: if _pipeline is None: _pipeline = CompliancePipeline() return _pipeline # ── Backward Compatibility ──────────────────────────────────────────── __all__ = [ "BaseTool", "tool_registry", "CompliancePipeline", "PipelineError", "PipelineValidationError", "PipelinePDFParsingError", "PipelineClassificationError", "PipelineSanctionsError", "get_pipeline", ]