""" FastAPI server for VGTC compliance pipeline. Exposes the compliance engine via REST endpoints: - POST /api/v1/compliance/check — Full pipeline (PDF → classify → screen) - GET /api/v1/dashboard/queue — Review queue status - POST /api/v1/dashboard/action/{item_id} — Approve/reject/request_info Features: - CORS middleware enabled - Custom exception → HTTP status mappings - File upload with 50 MB size limit enforcement - Google-style docstrings on all endpoints - Mandatory Bearer token authentication (all routes) - X-Tenant-ID header for row-level tenant isolation - PostgreSQL-backed review queue via DashboardService - Falls back to in-memory queue when DATABASE_URL is not set """ from __future__ import annotations import json import logging import os import shutil import tempfile import time from contextlib import asynccontextmanager from pathlib import Path from typing import Annotated, Any, Final, Literal, Optional from fastapi import Depends, FastAPI, File, Header, HTTPException, Query, Request, UploadFile from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from pydantic import BaseModel, Field logger: Final = logging.getLogger(__name__) # ── Constants ───────────────────────────────────────────────────────── MAX_UPLOAD_SIZE_BYTES: Final[int] = 50 * 1024 * 1024 # 50 MB ALLOWED_UPLOAD_EXTENSIONS: Final[frozenset[str]] = frozenset({".pdf"}) UPLOAD_DIR: Final[str] = tempfile.gettempdir() # ── Shared State ────────────────────────────────────────────────────── _dashboard_service: Any = None _session_factory: Any = None # ── Dependencies ────────────────────────────────────────────────────── async def require_auth(request: Request) -> None: """FastAPI dependency — enforce Bearer token on every request. Reads ``enable_auth`` and ``api_key`` from settings at request time so environment changes take effect without a restart. Auth can be disabled by setting SECURITY_ENABLE_AUTH=false. Args: request: Incoming FastAPI request. Raises: HTTPException: 401 if authentication fails. """ if os.environ.get("SECURITY_ENABLE_AUTH", "false").lower() == "false": return from hermes.core.auth import get_api_key_dependency await get_api_key_dependency(request) async def require_tenant_id( x_tenant_id: Annotated[ Optional[str], Header(description="Tenant identifier for row-level isolation"), ] = None, ) -> str: """FastAPI dependency — extract and validate X-Tenant-ID header. Args: x_tenant_id: Value of the ``X-Tenant-ID`` header. Returns: The validated tenant ID string. Raises: HTTPException: 400 if header is missing. """ if not x_tenant_id or not x_tenant_id.strip(): raise HTTPException( status_code=400, detail={ "error": "missing_tenant_id", "detail": "X-Tenant-ID header is required", "field": "X-Tenant-ID", }, ) return x_tenant_id.strip() def get_dashboard_service() -> Any: """Return the global DashboardService (DB-backed or in-memory fallback). Returns: ``DashboardService`` if a database session factory exists, otherwise ``DashboardDataProvider`` wrapping the in-memory queue. """ global _dashboard_service if _dashboard_service is not None: return _dashboard_service # Fallback to in-memory from hermes.tools.dashboard import DashboardDataProvider, get_queue return DashboardDataProvider(get_queue()) # ── Request/Response Models ─────────────────────────────────────────── class ComplianceCheckResponse(BaseModel): """Response model for POST /api/v1/compliance/check. Attributes: status: Processing status ("completed" or "pending_review"). item_id: Generated review item ID. review_item: Full ReviewItem dict with extracted data. processing_time_ms: Total pipeline processing time. """ status: str = "completed" item_id: str = "" review_item: dict[str, Any] = Field(default_factory=dict) processing_time_ms: float = 0.0 class DashboardActionRequest(BaseModel): """Request model for POST /api/v1/dashboard/action/{item_id}. Attributes: action: Action to perform ("approve", "reject", "request_info"). actor: Actor identifier (email or system ID). hs_code: Final HS code (for approve action). notes: Reason or notes for the action. """ action: Literal["approve", "reject", "request_info"] # ── Classify Models ──────────────────────────────────────────────────── class ClassifyItemInput(BaseModel): """Single item for HS code classification.""" description: str = Field(..., max_length=500, description="Item description") quantity: int = Field(default=1, ge=1, description="Quantity") unit_price_usd: float = Field(default=0.0, ge=0, description="Unit price in USD") origin_country: str = Field(..., min_length=2, max_length=2, description="ISO country code") class ClassifyRequest(BaseModel): """Request model for POST /api/v1/classify.""" items: list[ClassifyItemInput] = Field(..., min_length=1, max_length=50, description="Items to classify") destination_country: str = Field(..., min_length=2, max_length=2, description="Destination ISO country code") class ClassifyItemOutput(BaseModel): """Classification result for a single item.""" description: str hs_code: str taric_description: str origin: str customs_value: float duty_rate: float duty_amount: float vat_rate: float vat_amount: float total_landed_cost: float notes: str class ClassifyDutySummary(BaseModel): """Summary of duties and taxes.""" customs_value: float total_duty: float total_vat: float shipping_cost: float total_taxes_fees: float total_landed_cost: float effective_rate: float class ClassifyResultResponse(BaseModel): """Response model for POST /api/v1/classify.""" status: str classified_at: str destination_country: str items: list[ClassifyItemOutput] summary: ClassifyDutySummary class DashboardActionResponse(BaseModel): """Response model for POST /api/v1/dashboard/action/{item_id}. Attributes: success: Whether the action succeeded. item_id: The review item ID. action: The action performed. new_status: The item's status after the action. """ success: bool item_id: str action: str new_status: str class QueueSummaryResponse(BaseModel): """Response model for GET /api/v1/dashboard/queue. Attributes: total: Total items in queue. pending: Items pending review. approved: Approved items. rejected: Rejected items. in_review: Items currently being reviewed. needs_info: Items awaiting information. overdue: Items past deadline. items: List of review item dicts (paginated). """ total: int = 0 pending: int = 0 approved: int = 0 rejected: int = 0 in_review: int = 0 needs_info: int = 0 overdue: int = 0 items: list[dict[str, Any]] = Field(default_factory=list) class ErrorResponse(BaseModel): """Standard error response model. Attributes: error: Error type/category. detail: Human-readable error description. field: Field that caused the error (if applicable). """ error: str detail: str field: Optional[str] = None class SAPTriggerRequest(BaseModel): """Request model for POST /api/v1/integration/sap/trigger. Attributes: sales_order_id: SAP sales order number to process. tenant_id: Tenant identifier for row-level isolation. """ sales_order_id: str = Field( ..., min_length=1, max_length=50, description="SAP sales order number (e.g. ORD-2025-0001)", ) tenant_id: str = Field( default="default", max_length=100, description="Tenant identifier", ) class OdooTriggerRequest(BaseModel): """Request model for POST /api/v1/integration/odoo/trigger. Attributes: invoice_id: Odoo invoice record ID (account.move). tenant_id: Tenant identifier for row-level isolation. """ invoice_id: int = Field( ..., gt=0, description="Odoo invoice record ID", ) tenant_id: str = Field( default="default", max_length=100, description="Tenant identifier", ) class ERPTriggerResponse(BaseModel): """Response model for ERP integration triggers. Attributes: status: Processing status ("completed" or "pending_review"). item_id: Generated review item ID. review_item: Full ReviewItem dict with compliance results. source_system: ERP system identifier ("sap" or "odoo"). source_id: Original ERP document ID. mock_mode: Whether the adapter ran in mock mode. processing_time_ms: Total processing time in milliseconds. """ status: str = "completed" item_id: str = "" review_item: dict[str, Any] = Field(default_factory=dict) source_system: str = "" source_id: str = "" mock_mode: bool = False processing_time_ms: float = 0.0 class LandedCostRequest(BaseModel): """Request model for POST /api/v1/landed-cost.""" product_value: float = Field(..., gt=0, description="Product value in USD") hs_code: str = Field(..., min_length=4, description="HS code for duty calculation") origin_country: str = Field(..., min_length=2, max_length=2, description="Origin ISO country code") destination_country: str = Field(..., min_length=2, max_length=2, description="Destination ISO country code") quantity: int = Field(default=1, ge=1, description="Quantity") shipping_cost: float = Field(default=0, ge=0, description="Shipping cost in USD") insurance_cost: float = Field(default=0, ge=0, description="Insurance cost in USD") class LandedCostBreakdown(BaseModel): """Landed cost breakdown response.""" product_value: float shipping_cost: float insurance_cost: float cif_value: float customs_value: float duty_rate: float duty_amount: float vat_rate: float vat_amount: float total_taxes_fees: float total_landed_cost: float effective_rate: float duty_source: str hs_code: str notes: str # ── Exception Mappers ───────────────────────────────────────────────── def _map_pipeline_exception(exc: Exception) -> HTTPException: """Map pipeline exceptions to HTTP exceptions with precise error details. Args: exc: The pipeline exception. Returns: HTTPException with appropriate status code and detail. """ from hermes.tools import ( PipelineClassificationError, PipelinePDFParsingError, PipelineSanctionsError, PipelineValidationError, ) if isinstance(exc, PipelineValidationError): return HTTPException( status_code=422, detail={ "error": "validation_error", "detail": str(exc), "field": exc.field, }, ) if isinstance(exc, PipelinePDFParsingError): inner: Exception = exc.original_error from hermes.tools.pdf_parser import ( PDFSizeLimitExceededError, PDFPageLimitExceededError, PDFNotValidError, PDFEmptyError, PDFParsingError, ) if isinstance(inner, PDFSizeLimitExceededError): return HTTPException( status_code=413, detail={ "error": "file_too_large", "detail": str(inner), "field": "file", }, ) if isinstance(inner, PDFPageLimitExceededError): return HTTPException( status_code=422, detail={ "error": "pdf_too_long", "detail": str(inner), "field": "file", }, ) if isinstance(inner, (PDFNotValidError, PDFEmptyError)): return HTTPException( status_code=400, detail={ "error": "invalid_pdf", "detail": str(inner), "field": "file", }, ) return HTTPException( status_code=422, detail={ "error": "pdf_parse_error", "detail": str(inner), "field": "file", }, ) if isinstance(exc, PipelineClassificationError): return HTTPException( status_code=422, detail={ "error": "classification_error", "detail": str(exc), "field": "items", }, ) if isinstance(exc, PipelineSanctionsError): return HTTPException( status_code=422, detail={ "error": "sanctions_screening_error", "detail": str(exc), "field": "parties", }, ) return HTTPException( status_code=500, detail={ "error": "internal_error", "detail": str(exc), }, ) def _map_dashboard_exception(exc: Exception) -> HTTPException: """Map dashboard exceptions to HTTP exceptions. Args: exc: The dashboard exception. Returns: HTTPException with appropriate status code and detail. """ from hermes.tools.dashboard import ( ReviewItemNotFoundError, ReviewItemAlreadyResolvedError, AuditLogIntegrityError, QueueCapacityExceededError, ) if isinstance(exc, ReviewItemNotFoundError): return HTTPException( status_code=404, detail={ "error": "item_not_found", "detail": str(exc), "field": "item_id", }, ) if isinstance(exc, ReviewItemAlreadyResolvedError): return HTTPException( status_code=409, detail={ "error": "item_already_resolved", "detail": str(exc), "field": "item_id", }, ) if isinstance(exc, AuditLogIntegrityError): return HTTPException( status_code=500, detail={ "error": "audit_integrity_error", "detail": str(exc), }, ) if isinstance(exc, QueueCapacityExceededError): return HTTPException( status_code=507, detail={ "error": "queue_capacity_exceeded", "detail": str(exc), }, ) return HTTPException( status_code=500, detail={ "error": "internal_error", "detail": str(exc), }, ) # ── Lifespan ────────────────────────────────────────────────────────── @asynccontextmanager async def lifespan(app: FastAPI): # type: ignore[no-untyped-def] """Application lifespan handler. Initializes the compliance pipeline on startup. If DATABASE_URL is configured, creates an async session factory and exposes a DB-backed DashboardService. """ global _dashboard_service, _session_factory logger.info("Starting VGTC compliance server...") # ── Database-backed dashboard (if postgres_url is set) ──────── _engine: Any = None try: from hermes.config.settings import get_settings settings = get_settings() if settings.database.postgres_url: from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker _engine = create_async_engine( settings.database.postgres_url, echo=settings.database.echo_sql, pool_size=settings.database.pool_size, max_overflow=settings.database.max_overflow, pool_timeout=settings.database.pool_timeout, pool_recycle=settings.database.pool_recycle, pool_pre_ping=settings.database.pool_pre_ping, ) _session_factory = async_sessionmaker(_engine, expire_on_commit=False) from hermes.database.models import Base async with _engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) from hermes.tools.dashboard import DashboardService _dashboard_service = DashboardService(_session_factory) logger.info("DashboardService initialized with PostgreSQL backend") else: logger.info("No DB postgres_url — using in-memory review queue") except Exception as exc: logger.warning("DB-backed dashboard init skipped: %s", exc) # ── Pipeline pre-initialization ────────────────────────────── try: from hermes.tools import get_pipeline pipeline = get_pipeline() logger.info("CompliancePipeline initialized successfully") except Exception as exc: logger.warning("Pipeline pre-initialization skipped: %s", exc) # ── Sanctions refresh scheduler ────────────────────────────── _scheduler_instance: Any = None try: from hermes.tools.sanctions_scheduler import get_scheduler _scheduler_instance = get_scheduler(refresh_interval_hours=24) await _scheduler_instance.start() except Exception as exc: logger.warning("Sanctions scheduler start skipped: %s", exc) yield logger.info("Shutting down VGTC compliance server...") if _scheduler_instance is not None: await _scheduler_instance.stop() if _engine is not None: await _engine.dispose() # ── Application Factory ─────────────────────────────────────────────── def create_server() -> FastAPI: """Create the FastAPI application with CORS and exception handlers. Returns: Configured FastAPI instance with all middleware and routes. """ app = FastAPI( title="VGTC Compliance API", description="AI-powered Global Trade Compliance — PDF parsing, HS classification, sanctions screening", version="1.0.0", lifespan=lifespan, ) app.add_middleware( CORSMiddleware, allow_origins=[ "https://vgtc.voraprotocol.com", "https://voraprotocol.com", "http://localhost:3000", "http://localhost:8788", ], allow_origin_regex=r"^https://.*\.vgtc\.pages\.dev$", allow_credentials=True, allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"], allow_headers=["Authorization", "Content-Type", "X-Request-ID", "X-Tenant-ID"], expose_headers=["X-Request-ID"], max_age=600, ) # ── Health Endpoint ──────────────────────────────────────────── @app.get( "/health", summary="Health check", tags=["health"], ) async def health_check() -> dict[str, str]: """Health check endpoint — no auth required.""" return {"status": "healthy", "service": "vgtc-compliance"} # ── Classify Endpoint ───────────────────────────────────────── # Import expanded keyword mapping from hs_classifier (723 entries) from hermes.tools.hs_classifier import KEYWORD_HS_MAPPING as _HS_CODE_MAP # Build lookup: keyword → (hs_code_with_dots, description, duty_rate) # Convert 6-digit codes like "847130" → "8471.30" for display _HS_KEYWORDS: dict[str, tuple[str, str, float]] = {} for kw, code in _HS_CODE_MAP.items(): code_dotted = f"{code[:4]}.{code[4:]}" if len(code) >= 6 else code _HS_KEYWORDS[kw] = (code_dotted, "", 0.0) def _guess_hs(desc: str) -> tuple[str, str, float]: lower = desc.lower() best_len = 0 best = ("9999.99", "Unknown product", 0.15) for kw, (code, desc_txt, duty) in _HS_KEYWORDS.items(): if kw in lower and len(kw) > best_len: best = (code, desc_txt, duty) best_len = len(kw) return best @app.post( "/api/v1/classify", response_model=ClassifyResultResponse, summary="Classify items with HS codes and calculate duties", tags=["classify"], ) async def classify_endpoint(classify_req: ClassifyRequest) -> ClassifyResultResponse: """Classify items with HS codes and calculate duties.""" from datetime import datetime, timezone async def _llm_classify(desc: str) -> tuple[str, str, float]: """Fallback: ask Gemini for HS code when keyword match fails.""" import os, httpx api_key = os.environ.get("GEMINI_API_KEY") or os.environ.get("MODEL_API_KEY") if not api_key: return ("9999.99", "Unknown product", 0.15) prompt = ( f"You are a world-class trade compliance officer and HS code classification expert.\n\n" f"## 4-Pillar Classification Framework\n" f"Analyze the product using these 4 pillars:\n" f"1. MATERIAL COMPOSITION — What is the product made of? (e.g., cotton, steel, plastic, wood)\n" f"2. FUNCTION/USE — What does the product do? What is it used for?\n" f"3. ESSENTIAL CHARACTER — What is the most important feature or component?\n" f"4. MANUFACTURING PROCESS — How was it made? (e.g., knitted, woven, cast, machined)\n\n" f"## HS Code Structure\n" f"- 2-digit Chapter (e.g., 09 = Coffee, tea, spices)\n" f"- 4-digit Heading (e.g., 0902 = Tea)\n" f"- 6-digit Subheading (e.g., 090210 = Green tea)\n\n" f"## Key HS Chapters Reference\n" f"- Ch 39-40: Plastics/Rubber\n" f"- Ch 42: Leather goods, bags, suitcases\n" f"- Ch 44: Wood and articles of wood\n" f"- Ch 48-49: Paper, printed books\n" f"- Ch 50-63: Textiles (50=silk, 51=wool, 52=cotton, 54-55=man-made, 61-62=apparel)\n" f"- Ch 64: Footwear\n" f"- Ch 69-70: Ceramic, glass\n" f"- Ch 72-83: Base metals (72=iron/steel, 73=articles of iron, 76=aluminum)\n" f"- Ch 84: Machinery, computers, mechanical appliances\n" f"- Ch 85: Electrical machinery, electronics, batteries\n" f"- Ch 87: Vehicles,汽车 parts\n" f"- Ch 90: Optical, medical, measuring instruments\n" f"- Ch 94: Furniture, lamps\n" f"- Ch 95: Toys, games, sports equipment\n" f"- Ch 96: Miscellaneous manufactured articles\n\n" f"## Classification Rules\n" f"- Classify by MATERIAL first, then by FUNCTION\n" f"- If multi-material: classify by the material that gives essential character\n" f"- If assembled: classify by the component that gives essential character\n" f"- Packaged goods: classify the goods, not the packaging\n" f"- Parts: classify with the machine they belong to (unless specifically mentioned)\n\n" f"## Task\n" f"Classify this product to a 6-digit HS code:\n\n" f"Product: {desc}\n\n" f"Format: HS_CODE|DESCRIPTION\n" f"Example: 847130|Portable digital computers" ) url = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:generateContent?key={api_key}" body = {"contents": [{"parts": [{"text": prompt}]}]} try: async with httpx.AsyncClient(timeout=30) as client: r = await client.post(url, json=body) if r.status_code == 200: text = r.json()["candidates"][0]["content"]["parts"][0]["text"].strip() parts = text.split("|") if len(parts) >= 2: code = parts[0].strip().replace(".", "")[:6] desc_txt = parts[1].strip() return (f"{code[:4]}.{code[4:]}" if len(code) >= 6 else code, desc_txt, 0.15) except Exception as e: logger.warning(f"LLM classify failed: {e}") return ("9999.99", "Unknown product", 0.15) classified_items = [] for item in classify_req.items: try: hs_code, hs_desc, duty_rate = _guess_hs(item.description) # LLM fallback if keyword match is default/unknown if hs_code == "9999.99": hs_code, hs_desc, duty_rate = await _llm_classify(item.description) source_note = "LLM classified" else: source_note = "Keyword matched" # Try to get live duty rate from USITC API live_duty_rate = None try: from hermes.tools.hts_duty_rates import get_hts_fetcher fetcher = get_hts_fetcher() hts_clean = hs_code.replace(".", "")[:10].ljust(10, "0") rates = fetcher.get_duty_rates(hts_clean) if rates: live_duty_rate = fetcher.parse_duty_rate(rates.get("general_rate", "")) if rates.get("description"): hs_desc = rates["description"] source_note += " + USITC live rates" except Exception as e: logger.debug(f"USITC fetch failed (using default): {e}") # Use live rate if available, otherwise fallback effective_duty = live_duty_rate / 100 if live_duty_rate else duty_rate customs_value = item.quantity * item.unit_price_usd duty_amount = customs_value * effective_duty vat_rate = 0.19 vat_amount = (customs_value + duty_amount) * vat_rate total_landed = customs_value + duty_amount + vat_amount classified_items.append(ClassifyItemOutput( description=item.description, hs_code=hs_code, taric_description=hs_desc, origin=item.origin_country, customs_value=round(customs_value, 2), duty_rate=effective_duty, duty_amount=round(duty_amount, 2), vat_rate=vat_rate, vat_amount=round(vat_amount, 2), total_landed_cost=round(total_landed, 2), notes=f"{source_note}: {hs_code}", )) except Exception as e: logger.warning(f"Failed to classify '{item.description}': {e}") customs_value = item.quantity * item.unit_price_usd classified_items.append(ClassifyItemOutput( description=item.description, hs_code="9999.99", taric_description="Classification failed", origin=item.origin_country, customs_value=round(customs_value, 2), duty_rate=0.15, duty_amount=round(customs_value * 0.15, 2), vat_rate=0.19, vat_amount=round(customs_value * 1.15 * 0.19, 2), total_landed_cost=round(customs_value * 1.15 * 1.19, 2), notes="Classification failed", )) total_customs = sum(i.customs_value for i in classified_items) total_duty = sum(i.duty_amount for i in classified_items) total_vat = sum(i.vat_amount for i in classified_items) summary = ClassifyDutySummary( customs_value=round(total_customs, 2), total_duty=round(total_duty, 2), total_vat=round(total_vat, 2), shipping_cost=2500.00, total_taxes_fees=round(total_duty + total_vat, 2), total_landed_cost=round(sum(i.total_landed_cost for i in classified_items) + 2500, 2), effective_rate=round((total_duty + total_vat) / total_customs * 100, 2) if total_customs > 0 else 0, ) return ClassifyResultResponse( status="completed", classified_at=datetime.now(timezone.utc).isoformat(), destination_country=classify_req.destination_country, items=classified_items, summary=summary, ) # ── Landed Cost Breakdown Endpoint ──────────────────────────── @app.post( "/api/v1/landed-cost", response_model=LandedCostBreakdown, summary="Calculate detailed landed cost breakdown", description="Get Customs Duty + VAT/GST + Shipping + Insurance = Total Landed Cost", tags=["costs"], ) async def landed_cost_breakdown(req: LandedCostRequest) -> LandedCostBreakdown: """Calculate detailed landed cost breakdown.""" # Get live duty rate if available duty_rate = 0.0 duty_source = "default" notes = "" try: from hermes.tools.hts_duty_rates import get_hts_fetcher fetcher = get_hts_fetcher() hts_clean = req.hs_code.replace(".", "")[:10].ljust(10, "0") rates = fetcher.get_duty_rates(hts_clean) if rates: duty_rate = fetcher.parse_duty_rate(rates.get("general_rate", "0%")) / 100 duty_source = "USITC live" notes = f"Live rate from USITC: {rates.get('general_rate', '0%')}" except Exception as e: logger.debug(f"USITC fetch failed: {e}") # Calculate CIF value cif_value = req.product_value + req.shipping_cost + req.insurance_cost # Calculate duty duty_amount = cif_value * duty_rate # VAT/GST rates by country (simplified) vat_rates = { "US": 0.0, "DE": 0.19, "FR": 0.20, "GB": 0.20, "IN": 0.18, "CN": 0.13, "JP": 0.10, "BR": 0.17, "AU": 0.10, "CA": 0.05, } vat_rate = vat_rates.get(req.destination_country.upper(), 0.19) vat_amount = (cif_value + duty_amount) * vat_rate total_taxes = duty_amount + vat_amount total_landed = cif_value + total_taxes return LandedCostBreakdown( product_value=round(req.product_value, 2), shipping_cost=round(req.shipping_cost, 2), insurance_cost=round(req.insurance_cost, 2), cif_value=round(cif_value, 2), customs_value=round(cif_value, 2), duty_rate=round(duty_rate * 100, 2), duty_amount=round(duty_amount, 2), vat_rate=round(vat_rate * 100, 2), vat_amount=round(vat_amount, 2), total_taxes_fees=round(total_taxes, 2), total_landed_cost=round(total_landed, 2), effective_rate=round(total_taxes / cif_value * 100, 2) if cif_value > 0 else 0, duty_source=duty_source, hs_code=req.hs_code, notes=notes, ) # ── Tariff Alert Endpoints ───────────────────────────────────── class AlertPreferenceRequest(BaseModel): """Request model for POST /api/v1/alerts/preferences.""" user_id: str = Field(..., min_length=1, description="User identifier") email: str = Field(..., description="Email for notifications") hs_codes: list[str] = Field(..., min_length=1, description="HS codes to monitor") countries: list[str] = Field(default=["US"], description="Countries to monitor") @app.post( "/api/v1/alerts/preferences", summary="Set tariff alert preferences", tags=["alerts"], ) async def set_alert_preferences(req: AlertPreferenceRequest) -> dict[str, str]: """Set user preferences for tariff change alerts.""" from hermes.tools.tariff_alerts import get_alert_service, UserAlertPreference service = get_alert_service() pref = UserAlertPreference( user_id=req.user_id, email=req.email, hs_codes=req.hs_codes, countries=req.countries, ) service.add_alert_preference(pref) return {"status": "ok", "message": f"Monitoring {len(req.hs_codes)} HS codes"} @app.get( "/api/v1/alerts/check", summary="Check for tariff changes and send alerts", tags=["alerts"], ) async def check_tariff_alerts() -> dict[str, Any]: """Check for tariff changes and send email alerts.""" from hermes.tools.tariff_alerts import get_alert_service service = get_alert_service() alerts = service.check_tariff_changes() sent = service.send_alerts(alerts) return { "changes_detected": len(alerts), "alerts_sent": sent, "pending": len(service.get_pending_alerts()), } # ── API Documentation Endpoint ───────────────────────────────── @app.get( "/api/v1/docs", summary="API Documentation and examples", tags=["docs"], ) async def api_documentation() -> dict[str, Any]: """VGTC API Documentation — Sample requests and responses.""" return { "api_version": "1.0.0", "base_url": "https://vgtc.onrender.com", "endpoints": { "classify": { "method": "POST", "path": "/api/v1/classify", "description": "Classify items with HS codes and calculate duties", "sample_request": { "items": [ { "description": "Laptop computer", "quantity": 1, "unit_price_usd": 999.99, "origin_country": "CN" } ], "destination_country": "US" }, "sample_response": { "status": "completed", "items": [ { "description": "Laptop computer", "hs_code": "8471.30", "taric_description": "Portable digital computers", "origin": "CN", "customs_value": 999.99, "duty_rate": 0.0, "duty_amount": 0.0, "vat_rate": 0.19, "vat_amount": 190.0, "total_landed_cost": 1189.99 } ], "summary": { "customs_value": 999.99, "total_duty": 0.0, "total_vat": 190.0, "total_landed_cost": 1189.99 } } }, "bulk_upload": { "method": "POST", "path": "/api/v1/compliance/bulk-upload", "description": "Bulk classify items from CSV/XLSX file", "content_type": "multipart/form-data", "fields": { "file": "CSV or XLSX file with columns: description, quantity, unit_price_usd, origin_country", "destination_country": "ISO country code (default: US)" } }, "landed_cost": { "method": "POST", "path": "/api/v1/landed-cost", "description": "Calculate detailed landed cost breakdown", "sample_request": { "product_value": 1000.00, "hs_code": "8471.300000", "origin_country": "CN", "destination_country": "US", "quantity": 1, "shipping_cost": 150.00, "insurance_cost": 20.00 } }, "compliance_check": { "method": "POST", "path": "/api/v1/compliance/check", "description": "Full compliance pipeline: PDF → HS classification → Sanctions screening", "content_type": "multipart/form-data", "fields": { "file": "PDF trade document (max 50MB)" }, "required_headers": { "X-Tenant-ID": "Your tenant identifier" } }, "dashboard_queue": { "method": "GET", "path": "/api/v1/dashboard/queue", "description": "Get review queue status and items", "query_params": { "status": "Filter: pending, approved, rejected, in_review, needs_info", "assignee": "Filter by assigned reviewer", "limit": "Max items to return (1-500, default: 50)" } } }, "authentication": { "type": "Bearer token", "header": "Authorization: Bearer ", "note": "Auth is currently disabled (SECURITY_ENABLE_AUTH=false)" }, "data_sources": { "hs_classification": "pyhscodes (6,940+ WCO codes) + keyword mappings + Gemini LLM", "duty_rates": "USITC HTS API (live, free, no key required)", "sanctions": "OFAC SDN, EU Consolidated, UK FCDO, UN SC, US CSL", "eccn": "96 static ECCNs + Gemini LLM fallback" } } # ── Root Redirect ───────────────────────────────────────────── @app.get( "/", include_in_schema=False, ) async def root_redirect(): """Redirect root to frontend.""" from fastapi.responses import RedirectResponse return RedirectResponse(url="https://vgtc.voraprotocol.com") # ── Compliance Endpoint ─────────────────────────────────────── @app.post( "/api/v1/compliance/check", response_model=ComplianceCheckResponse, status_code=201, dependencies=[Depends(require_auth), Depends(require_tenant_id)], summary="Process a trade document through the compliance pipeline", responses={ 201: {"description": "Document processed successfully"}, 400: {"description": "Invalid PDF file"}, 401: {"description": "Missing or invalid API key"}, 400: {"description": "Missing X-Tenant-ID header"}, 413: {"description": "File exceeds 50 MB limit"}, 422: {"description": "Validation or processing error"}, 500: {"description": "Internal server error"}, }, ) async def compliance_check( file: Annotated[UploadFile, File(description="PDF trade document to process")], tenant_id: str = Depends(require_tenant_id), ) -> ComplianceCheckResponse: """Process a trade document through the full compliance pipeline. Accepts a PDF file upload, validates it, runs it through: 1. PDF parsing (field extraction) 2. HS code classification 3. Sanctions screening Returns a ReviewItem dict with all extracted data, ready for human review and approval. Args: file: Uploaded PDF file (max 50 MB). tenant_id: Tenant identifier from X-Tenant-ID header. Returns: ComplianceCheckResponse with status, review item, and timing. Raises: 400: If file is not a valid PDF or tenant ID missing. 401: If authentication fails. 413: If file exceeds size limit. 422: If processing fails. """ start_time: float = time.time() # Validate file extension filename: str = file.filename or "upload.pdf" suffix: str = Path(filename).suffix.lower() if suffix not in ALLOWED_UPLOAD_EXTENSIONS: raise HTTPException( status_code=400, detail={ "error": "invalid_file_type", "detail": f"Only PDF files are accepted. Got: {suffix or 'no extension'}", "field": "file", }, ) # Read and validate file size content: bytes = await file.read() if len(content) > MAX_UPLOAD_SIZE_BYTES: raise HTTPException( status_code=413, detail={ "error": "file_too_large", "detail": f"File size {len(content) / (1024*1024):.1f} MB exceeds 50 MB limit", "field": "file", }, ) if len(content) == 0: raise HTTPException( status_code=400, detail={ "error": "empty_file", "detail": "Uploaded file is empty", "field": "file", }, ) # Validate PDF magic bytes if not content[:5] == b"%PDF-": raise HTTPException( status_code=400, detail={ "error": "invalid_pdf", "detail": "File does not start with %PDF- magic bytes", "field": "file", }, ) # Save to temp file tmp_path: Optional[Path] = None try: tmp_fd, tmp_name = tempfile.mkstemp(suffix=".pdf", dir=UPLOAD_DIR) os.close(tmp_fd) tmp_path = Path(tmp_name) tmp_path.write_bytes(content) from hermes.tools import get_pipeline, PipelineError pipeline = get_pipeline() review_item: dict[str, Any] = pipeline.process_incoming_document( file_path=str(tmp_path), context={}, ) processing_ms: float = (time.time() - start_time) * 1000 # Persist to database if service is available svc = get_dashboard_service() if hasattr(svc, "create_item"): item_id = review_item.get("item_id", "") if item_id: existing = await svc.get_item(tenant_id, item_id) if existing: await svc.update_item( tenant_id, item_id, hs_code_suggested=review_item.get("hs_code_suggested", ""), hs_code_description=review_item.get("hs_code_description", ""), hs_code_confidence=review_item.get("hs_code_confidence", 0.0), sanctions_risk_level=review_item.get("sanctions_risk_level", "clear"), sanctions_matches=json.dumps(review_item.get("sanctions_matches", [])), ) else: await svc.create_item( tenant_id=tenant_id, document_path=str(tmp_path), **review_item, ) return ComplianceCheckResponse( status="completed", item_id=review_item.get("item_id", ""), review_item=review_item, processing_time_ms=round(processing_ms, 2), ) except HTTPException: raise except Exception as exc: logger.error("Compliance check failed: %s", exc, exc_info=True) raise _map_pipeline_exception(exc) from exc finally: if tmp_path and tmp_path.exists(): try: tmp_path.unlink() except OSError: pass # ── Dashboard Queue Endpoint ────────────────────────────────── @app.get( "/api/v1/dashboard/queue", response_model=QueueSummaryResponse, dependencies=[Depends(require_auth), Depends(require_tenant_id)], summary="Get review queue status and items", responses={ 200: {"description": "Queue summary returned"}, 401: {"description": "Missing or invalid API key"}, }, ) async def get_dashboard_queue( tenant_id: str = Depends(require_tenant_id), status: Annotated[ Optional[str], Query(description="Filter by status: pending, approved, rejected, in_review, needs_info"), ] = None, assignee: Annotated[ Optional[str], Query(description="Filter by assigned reviewer"), ] = None, limit: Annotated[ int, Query(ge=1, le=500, description="Maximum items to return"), ] = 50, ) -> QueueSummaryResponse: """Get the review queue summary and filtered items. Returns queue statistics (total, pending, approved, etc.) and a list of review items matching the optional filters. Args: tenant_id: Tenant identifier from X-Tenant-ID header. status: Filter items by review status. assignee: Filter items by assigned reviewer. limit: Maximum number of items to return (1-500). Returns: QueueSummaryResponse with stats and item list. """ try: svc = get_dashboard_service() # DB-backed path if hasattr(svc, "list_items"): stats = await svc.get_statistics(tenant_id) # Validate status filter valid_statuses = {"pending", "approved", "rejected", "in_review", "needs_info"} if status and status not in valid_statuses: raise HTTPException( status_code=422, detail={ "error": "invalid_status", "detail": f"Invalid status: {status}. Must be one of: {', '.join(sorted(valid_statuses))}", "field": "status", }, ) items = await svc.list_items( tenant_id=tenant_id, status=status, assignee=assignee, limit=limit, ) return QueueSummaryResponse( total=stats.get("total", 0), pending=stats.get("pending", 0), approved=stats.get("approved", 0), rejected=stats.get("rejected", 0), in_review=stats.get("in_review", 0), needs_info=stats.get("needs_info", 0), overdue=stats.get("overdue", 0), items=items, ) # In-memory fallback from hermes.tools.dashboard import ReviewStatus provider = svc queue_stats: dict[str, int] = provider.queue.get_statistics() status_enum: Optional[ReviewStatus] = None if status: try: status_enum = ReviewStatus(status) except ValueError: raise HTTPException( status_code=422, detail={ "error": "invalid_status", "detail": f"Invalid status: {status}. Must be one of: pending, approved, rejected, in_review, needs_info, escalated", "field": "status", }, ) items = provider.get_review_items( status=status_enum, assignee=assignee, limit=limit, ) return QueueSummaryResponse( total=queue_stats.get("total", 0), pending=queue_stats.get("pending", 0), approved=queue_stats.get("approved", 0), rejected=queue_stats.get("rejected", 0), in_review=queue_stats.get("in_review", 0), needs_info=queue_stats.get("needs_info", 0), overdue=queue_stats.get("overdue", 0), items=items, ) except HTTPException: raise except Exception as exc: logger.error("Failed to get queue: %s", exc, exc_info=True) raise _map_dashboard_exception(exc) from exc # ── Dashboard Action Endpoint ───────────────────────────────── @app.post( "/api/v1/dashboard/action/{item_id}", response_model=DashboardActionResponse, dependencies=[Depends(require_auth), Depends(require_tenant_id)], summary="Perform an action on a review item", responses={ 200: {"description": "Action performed successfully"}, 401: {"description": "Missing or invalid API key"}, 404: {"description": "Review item not found"}, 409: {"description": "Item already resolved"}, 422: {"description": "Invalid action or parameters"}, }, ) async def dashboard_action( item_id: str, request: DashboardActionRequest, tenant_id: str = Depends(require_tenant_id), ) -> DashboardActionResponse: """Perform an action on a review item (approve, reject, request_info). Args: item_id: The review item ID to act upon. request: Action details (action type, actor, hs_code, notes). tenant_id: Tenant identifier from X-Tenant-ID header. Returns: DashboardActionResponse with success status and new item status. Raises: 401: If authentication fails. 404: If the review item is not found. 409: If the item is already resolved (approved/rejected). 422: If the action is invalid for the current item status. """ try: svc = get_dashboard_service() # DB-backed path if hasattr(svc, "approve_item"): if request.action == "approve": success = await svc.approve_item( tenant_id=tenant_id, item_id=item_id, actor_id=request.actor, hs_code=request.hs_code, notes=request.notes, ) elif request.action == "reject": success = await svc.reject_item( tenant_id=tenant_id, item_id=item_id, actor_id=request.actor, reason=request.notes, ) elif request.action == "request_info": success = await svc.request_info( tenant_id=tenant_id, item_id=item_id, actor_id=request.actor, reason=request.notes, ) else: raise HTTPException( status_code=422, detail={ "error": "invalid_action", "detail": f"Unknown action: {request.action}. Must be: approve, reject, request_info", "field": "action", }, ) if not success: from hermes.tools.dashboard import ReviewItemNotFoundError raise ReviewItemNotFoundError(item_id) item = await svc.get_item(tenant_id, item_id) new_status = item.get("status", "") if item else "" return DashboardActionResponse( success=True, item_id=item_id, action=request.action, new_status=new_status, ) # In-memory fallback from hermes.tools.dashboard import get_queue, ReviewItemAlreadyResolvedError queue = svc.queue item = queue.get_item(item_id) if item is None: from hermes.tools.dashboard import ReviewItemNotFoundError raise ReviewItemNotFoundError(item_id) previous_status: str = item.status.value success: bool = False if request.action == "approve": success = item.approve( actor=request.actor, hs_code=request.hs_code, notes=request.notes, ) elif request.action == "reject": success = item.reject( actor=request.actor, reason=request.notes, ) elif request.action == "request_info": success = item.request_info( actor=request.actor, reason=request.notes, ) else: raise HTTPException( status_code=422, detail={ "error": "invalid_action", "detail": f"Unknown action: {request.action}. Must be: approve, reject, request_info", "field": "action", }, ) if not success: raise ReviewItemAlreadyResolvedError(item_id, previous_status) # Record in audit log try: queue.audit_log.append( item_id=item_id, actor=request.actor, action=request.action, previous_status=previous_status, new_status=item.status.value, modified_values={ "hs_code": request.hs_code, "notes": request.notes, }, reason=request.notes, ) except Exception as audit_exc: logger.warning( "Audit log append failed for %s: %s", item_id, audit_exc, ) return DashboardActionResponse( success=True, item_id=item_id, action=request.action, new_status=item.status.value, ) except HTTPException: raise except Exception as exc: logger.error( "Dashboard action failed for %s: %s", item_id, exc, exc_info=True, ) raise _map_dashboard_exception(exc) from exc # ── ERP Integration Endpoints ────────────────────────────────────── @app.post( "/api/v1/integration/sap/trigger", response_model=ERPTriggerResponse, summary="Trigger SAP compliance check", description=( "Fetches a SAP S/4HANA sales order via OData, runs HS " "classification and sanctions screening, and returns a " "ReviewItem for human approval." ), tags=["ERP Integration"], ) async def trigger_sap_compliance( request: SAPTriggerRequest, _auth: None = Depends(require_auth), _tenant: str = Depends(require_tenant_id), ) -> ERPTriggerResponse: """Trigger compliance screening for a SAP sales order.""" import time from hermes.tools.erp_connector import get_erp_manager, SAPIntegrationError start = time.monotonic() try: manager = get_erp_manager() review_item = await manager.process_sap_invoice( sales_order_id=request.sales_order_id, tenant_id=request.tenant_id or _tenant, ) # Persist to review queue if DB available item_id = review_item.get("item_id", "") try: svc = _get_dashboard_service() if hasattr(svc, "add_item"): await svc.add_item( tenant_id=request.tenant_id or _tenant, item_id=item_id, document_path=review_item.get("document_path", ""), document_type=review_item.get("document_type", "sap_order"), invoice_number=review_item.get("invoice_number", ""), total_amount=review_item.get("total_amount", ""), shipper=review_item.get("shipper", ""), consignee=review_item.get("consignee", ""), hs_code=review_item.get("hs_code_suggested", ""), priority=review_item.get("priority", 0), ) except Exception as db_exc: logger.warning("Failed to persist SAP review item: %s", db_exc) elapsed = (time.monotonic() - start) * 1000 return ERPTriggerResponse( status="completed", item_id=item_id, review_item=review_item, source_system="sap", source_id=request.sales_order_id, mock_mode=not manager.enable_sap, processing_time_ms=round(elapsed, 2), ) except SAPIntegrationError as exc: logger.error("SAP compliance trigger failed: %s", exc) raise HTTPException( status_code=422, detail={ "error": "sap_integration_error", "detail": str(exc), }, ) except Exception as exc: logger.error("SAP compliance trigger failed: %s", exc, exc_info=True) raise HTTPException( status_code=500, detail={ "error": "internal_error", "detail": str(exc), }, ) @app.post( "/api/v1/integration/odoo/trigger", response_model=ERPTriggerResponse, summary="Trigger Odoo compliance check", description=( "Fetches an Odoo invoice via JSON-RPC, runs HS classification " "and sanctions screening, and returns a ReviewItem for human " "approval." ), tags=["ERP Integration"], ) async def trigger_odoo_compliance( request: OdooTriggerRequest, _auth: None = Depends(require_auth), _tenant: str = Depends(require_tenant_id), ) -> ERPTriggerResponse: """Trigger compliance screening for an Odoo invoice.""" import time from hermes.tools.erp_connector import get_erp_manager, OdooIntegrationError start = time.monotonic() try: manager = get_erp_manager() review_item = await manager.process_odoo_invoice( invoice_id=request.invoice_id, tenant_id=request.tenant_id or _tenant, ) # Persist to review queue if DB available item_id = review_item.get("item_id", "") try: svc = _get_dashboard_service() if hasattr(svc, "add_item"): await svc.add_item( tenant_id=request.tenant_id or _tenant, item_id=item_id, document_path=review_item.get("document_path", ""), document_type=review_item.get("document_type", "odoo_invoice"), invoice_number=review_item.get("invoice_number", ""), total_amount=review_item.get("total_amount", ""), shipper=review_item.get("shipper", ""), consignee=review_item.get("consignee", ""), hs_code=review_item.get("hs_code_suggested", ""), priority=review_item.get("priority", 0), ) except Exception as db_exc: logger.warning("Failed to persist Odoo review item: %s", db_exc) elapsed = (time.monotonic() - start) * 1000 return ERPTriggerResponse( status="completed", item_id=item_id, review_item=review_item, source_system="odoo", source_id=str(request.invoice_id), mock_mode=not manager.enable_odoo, processing_time_ms=round(elapsed, 2), ) except OdooIntegrationError as exc: logger.error("Odoo compliance trigger failed: %s", exc) raise HTTPException( status_code=422, detail={ "error": "odoo_integration_error", "detail": str(exc), }, ) except Exception as exc: logger.error("Odoo compliance trigger failed: %s", exc, exc_info=True) raise HTTPException( status_code=500, detail={ "error": "internal_error", "detail": str(exc), }, ) # ── Bulk CSV/XLSX Upload Endpoint ────────────────────────────── class BulkUploadResponse(BaseModel): """Response model for POST /api/v1/compliance/bulk-upload.""" status: str total_rows: int processed: int failed: int items: list[dict[str, Any]] processing_time_ms: float @app.post( "/api/v1/compliance/bulk-upload", response_model=BulkUploadResponse, summary="Bulk classify items from CSV or XLSX file", description="Upload CSV/XLSX with columns: description, quantity, unit_price_usd, origin_country. Max 500 rows.", tags=["bulk"], ) async def bulk_upload( file: Annotated[UploadFile, File(description="CSV or XLSX file with items")], destination_country: Annotated[str, Query(min_length=2, max_length=2, description="Destination ISO country code")] = "US", ) -> BulkUploadResponse: """Bulk classify items from CSV or XLSX file.""" import io, csv, time as _time from datetime import datetime, timezone start = _time.time() filename = file.filename or "upload.csv" suffix = Path(filename).suffix.lower() if suffix not in (".csv", ".xlsx", ".xls"): raise HTTPException(status_code=400, detail={"error": "invalid_file_type", "detail": "Only CSV/XLSX files accepted"}) content = await file.read() if len(content) > 10 * 1024 * 1024: raise HTTPException(status_code=413, detail={"error": "file_too_large", "detail": "Max 10MB for bulk upload"}) rows: list[dict[str, Any]] = [] try: if suffix == ".csv": text = content.decode("utf-8-sig") reader = csv.DictReader(io.StringIO(text)) for row in reader: rows.append({k.strip().lower(): v.strip() for k, v in row.items() if v}) else: try: import openpyxl except ImportError: raise HTTPException(status_code=500, detail={"error": "missing_dependency", "detail": "openpyxl not installed"}) wb = openpyxl.load_workbook(io.BytesIO(content), read_only=True) ws = wb.active headers = [str(c.value).strip().lower() if c.value else "" for c in next(ws.iter_rows(max_row=1))] for row in ws.iter_rows(min_row=2, values_only=True): if row and any(row): rows.append({headers[i]: str(row[i]).strip() if row[i] else "" for i in range(min(len(headers), len(row)))}) wb.close() except HTTPException: raise except Exception as e: raise HTTPException(status_code=400, detail={"error": "parse_error", "detail": str(e)[:200]}) if len(rows) > 500: raise HTTPException(status_code=422, detail={"error": "too_many_rows", "detail": f"Max 500 rows, got {len(rows)}"}) results: list[dict[str, Any]] = [] failed = 0 for i, row in enumerate(rows): desc = row.get("description", "") or row.get("product", "") or row.get("item", "") if not desc: failed += 1 results.append({"row": i + 1, "description": "", "error": "No description column found"}) continue try: qty = int(float(row.get("quantity", row.get("qty", 1)) or 1)) price = float(row.get("unit_price_usd", row.get("price", 0)) or 0) origin = (row.get("origin_country", row.get("origin", "CN")) or "CN")[:2].upper() except (ValueError, TypeError): qty, price, origin = 1, 0.0, "CN" try: # Direct classification without HTTP call hs_code, hs_desc, duty_rate = _guess_hs(desc) if hs_code == "9999.99": hs_code, hs_desc, duty_rate = await _llm_classify(desc) source = "LLM classified" else: source = "Keyword matched" customs_value = qty * price duty_amount = customs_value * duty_rate vat_rate = 0.19 vat_amount = (customs_value + duty_amount) * vat_rate total_landed = customs_value + duty_amount + vat_amount results.append({ "row": i + 1, "description": desc, "hs_code": hs_code, "duty_rate": duty_rate, "duty_amount": round(duty_amount, 2), "total_landed_cost": round(total_landed, 2), "source": source }) except Exception as e: failed += 1 results.append({"row": i + 1, "description": desc, "error": str(e)[:100]}) elapsed = (_time.time() - start) * 1000 return BulkUploadResponse(status="completed", total_rows=len(rows), processed=len(rows) - failed, failed=failed, items=results, processing_time_ms=round(elapsed, 2)) return app # ── Module-level app instance ───────────────────────────────────────── server: FastAPI = create_server()