Spaces:
Paused
Paused
| """ | |
| ECCN (Export Control Classification Number) classification engine. | |
| Classifies items under the US Bureau of Industry and Security (BIS) Export | |
| Administration Regulations (EAR). Uses the Commerce Control List (CCL) | |
| structured as ECCN codes (e.g., 3A001, 5D002). | |
| Classification pipeline: | |
| 1. Static ECCN database lookup (instant, FREE) | |
| 2. LLM classification via DeepSeek V4 Flash (requires API key) | |
| 3. Confidence scoring + human review flagging | |
| Data sources: | |
| - BIS Commerce Control List (CCL) β 10 categories Γ 5 product groups | |
| - Common ECCNs embedded as static dataset (public domain structure) | |
| - Country Chart / license requirements per reason for control | |
| Typing conventions: | |
| All public APIs use explicit type hints. Literal types enforce valid | |
| categories and product groups. Final constants prevent mutation. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import logging | |
| import os | |
| import re | |
| import threading | |
| import time as _time | |
| from dataclasses import dataclass, field | |
| from typing import Any, Final, Literal, Optional | |
| import httpx | |
| logger: Final = logging.getLogger(__name__) | |
| # ββ Domain Exceptions βββββββββββββββββββββββββββββββββββββββββββββββββ | |
| class ECCNClassificationError(Exception): | |
| """Base exception for all ECCN classification errors.""" | |
| class ECCNValidationError(ECCNClassificationError): | |
| """Raised when an ECCN code fails structural validation. | |
| Attributes: | |
| code: The invalid code. | |
| reason: Why validation failed. | |
| """ | |
| def __init__(self, code: str, reason: str) -> None: | |
| self.code = code | |
| self.reason = reason | |
| super().__init__(f"Invalid ECCN code '{code}': {reason}") | |
| class ECCNLookupError(ECCNClassificationError): | |
| """Raised when an ECCN code lookup fails in the database.""" | |
| class ECCNLLMClassificationError(ECCNClassificationError): | |
| """Raised when LLM-based classification fails. | |
| Attributes: | |
| provider: The LLM provider name. | |
| detail: Error detail. | |
| """ | |
| def __init__(self, provider: str, detail: str) -> None: | |
| self.provider = provider | |
| self.detail = detail | |
| super().__init__(f"LLM classification failed ({provider}): {detail}") | |
| # ββ Constants βββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # ECCN code pattern: letter + 4 digits (e.g., 3A001, 5D002) | |
| ECCN_PATTERN: Final[re.Pattern[str]] = re.compile(r"^[0-9][A-E]\d{3}$") | |
| # Confidence thresholds | |
| HIGH_CONFIDENCE_THRESHOLD: Final[float] = 0.85 | |
| MEDIUM_CONFIDENCE_THRESHOLD: Final[float] = 0.65 | |
| LLM_MAX_CONFIDENCE: Final[float] = 0.85 | |
| # Type aliases | |
| ProductGroup = Literal["A", "B", "C", "D", "E"] | |
| ControlReason = Literal[ | |
| "NS", "MT", "CB", "CC", "RS", "AT", "EI", "SL", "GB", | |
| ] | |
| # CCL Category names (EAR Part 774, Supplement No. 1) | |
| CCL_CATEGORIES: Final[dict[int, str]] = { | |
| 0: "Nuclear Materials, Facilities & Equipment", | |
| 1: "Materials, Chemicals, Microorganisms & Toxins", | |
| 2: "Materials Processing", | |
| 3: "Electronics", | |
| 4: "Computers", | |
| 5: "Telecommunications & Information Security", | |
| 6: "Sensors and Lasers", | |
| 7: "Navigation & Avionics", | |
| 8: "Marine", | |
| 9: "Aerospace & Propulsion", | |
| } | |
| # Product groups | |
| PRODUCT_GROUPS: Final[dict[str, str]] = { | |
| "A": "Equipment, Components, and Materials", | |
| "B": "Test, Inspection, and Production Equipment", | |
| "C": "Materials", | |
| "D": "Software", | |
| "E": "Technology", | |
| } | |
| # Reasons for control | |
| CONTROL_REASONS: Final[dict[str, str]] = { | |
| "NS": "National Security (NS applies under NSG, Wassenaar, etc.)", | |
| "MT": "Missile Technology (MTCR)", | |
| "CB": "Chemical & Biological Weapons (Australia Group)", | |
| "CC": "Chemical & Biological Weapons (CBW end-use)", | |
| "RS": "Regional Stability", | |
| "AT": "Anti-Terrorism (Country Group E:1/E:2)", | |
| "EI": "Encryption Items", | |
| "SL": "Surreptitious Listening", | |
| "GB": "Firearms / Guns & Ammunition (legacy)", | |
| } | |
| # Country Chart: reason for control β license requirement per country group | |
| # Based on EAR Part 738 Supplement No. 1 (simplified) | |
| COUNTRY_CHART: Final[dict[str, str]] = { | |
| "NS": "Yes β License required for NS Group D:1 countries", | |
| "MT": "Yes β License required for all destinations except Canada", | |
| "CB": "Yes β License required for AG Group D:1 countries", | |
| "CC": "Yes β License required for CBW-end-use countries", | |
| "RS": "Yes β License required for RS Group D:2 countries", | |
| "AT": "Yes β License required for E:1/E:2 countries", | |
| "EI": "Yes β License required for Country Group D:5", | |
| "SL": "Yes β License required for all destinations except Canada", | |
| "GB": "Yes β License required for E:1/E:2 countries", | |
| } | |
| # Common ECCN database (static, public domain structure) | |
| # Format: eccn_code β {description, items_controlled, control_reasons, product_group} | |
| _COMMON_ECCNS: Final[dict[str, dict[str, Any]]] = { | |
| # ββ Category 0: Nuclear Materials, Facilities & Equipment ββββββββββ | |
| "0A001": { | |
| "description": "Nuclear reactors and related equipment", | |
| "items_controlled": "Nuclear reactors, fuel elements, control rods, moderator assemblies", | |
| "control_reasons": ["NS"], | |
| "product_group": "A", | |
| }, | |
| "0A002": { | |
| "description": "Nuclear reactor internals and components", | |
| "items_controlled": "Pressure vessels, heat exchangers, gas turbine generators for nuclear", | |
| "control_reasons": ["NS"], | |
| "product_group": "A", | |
| }, | |
| "0B001": { | |
| "description": "Nuclear reactor equipment for fuel processing", | |
| "items_controlled": "Equipment for the production of heavy water, uranium enrichment", | |
| "control_reasons": ["NS"], | |
| "product_group": "B", | |
| }, | |
| "0C001": { | |
| "description": "Nuclear reactor fuels", | |
| "items_controlled": "Natural uranium, depleted uranium, thorium, fuel assemblies", | |
| "control_reasons": ["NS"], | |
| "product_group": "C", | |
| }, | |
| "0D001": { | |
| "description": "Software for nuclear reactors", | |
| "items_controlled": "Software for design, operation, or maintenance of nuclear reactors", | |
| "control_reasons": ["NS"], | |
| "product_group": "D", | |
| }, | |
| "0E001": { | |
| "description": "Technology for nuclear materials", | |
| "items_controlled": "Technology for development, production, or use of nuclear materials", | |
| "control_reasons": ["NS"], | |
| "product_group": "E", | |
| }, | |
| # ββ Category 1: Materials, Chemicals, Microorganisms & Toxins ββββββ | |
| "1A001": { | |
| "description": "Composite structures or laminates", | |
| "items_controlled": "Fibrous or filamentary materials, prepregs, composite structures", | |
| "control_reasons": ["NS", "CB"], | |
| "product_group": "A", | |
| }, | |
| "1A002": { | |
| "description": "Materials for the production of chemical weapons precursors", | |
| "items_controlled": "Equipment for the production of chemical weapons precursors", | |
| "control_reasons": ["CB"], | |
| "product_group": "A", | |
| }, | |
| "1A003": { | |
| "description": "Toxicological agents and related equipment", | |
| "items_controlled": "Chemical agents, biological agents, toxins, dissemination equipment", | |
| "control_reasons": ["CB"], | |
| "product_group": "A", | |
| }, | |
| "1B001": { | |
| "description": "Chemical manufacturing equipment", | |
| "items_controlled": "Chemical production equipment, bioprocessing equipment", | |
| "control_reasons": ["CB"], | |
| "product_group": "B", | |
| }, | |
| "1C001": { | |
| "description": "Chemical weapons precursors", | |
| "items_controlled": "Schedule 2 chemicals (thiodiglycol, phosphorus, etc.)", | |
| "control_reasons": ["CB"], | |
| "product_group": "C", | |
| }, | |
| "1C002": { | |
| "description": "Chemical weapon agents and precursors", | |
| "items_controlled": "Schedule 3 chemicals (phosgene, cyanogen chloride, etc.)", | |
| "control_reasons": ["CB"], | |
| "product_group": "C", | |
| }, | |
| "1C350": { | |
| "description": "Toxic chemicals and precursors", | |
| "items_controlled": "Schedule 1 chemicals (sarin, VX, etc.)", | |
| "control_reasons": ["CB"], | |
| "product_group": "C", | |
| }, | |
| "1D001": { | |
| "description": "Software for chemical weapons", | |
| "items_controlled": "Software for chemical weapon design, production, or delivery", | |
| "control_reasons": ["CB"], | |
| "product_group": "D", | |
| }, | |
| "1E001": { | |
| "description": "Technology for chemical weapons", | |
| "items_controlled": "Technology for chemical weapon development or production", | |
| "control_reasons": ["CB"], | |
| "product_group": "E", | |
| }, | |
| # ββ Category 2: Materials Processing βββββββββββββββββββββββββββββββ | |
| "2A001": { | |
| "description": "Specialized manufacturing equipment", | |
| "items_controlled": "Numerically controlled machine tools, precision forming machines", | |
| "control_reasons": ["NS"], | |
| "product_group": "A", | |
| }, | |
| "2A002": { | |
| "description": "Numerically controlled machine tools", | |
| "items_controlled": "CNC lathes, milling machines, grinding machines", | |
| "control_reasons": ["NS"], | |
| "product_group": "A", | |
| }, | |
| "2B001": { | |
| "description": "Numerically controlled machine tool equipment", | |
| "items_controlled": "Dimension inspection machines, coordinate measuring machines", | |
| "control_reasons": ["NS"], | |
| "product_group": "B", | |
| }, | |
| "2B002": { | |
| "description": "Vacuum or controlled atmosphere furnaces", | |
| "items_controlled": "Vacuum furnaces, inert atmosphere furnaces", | |
| "control_reasons": ["NS"], | |
| "product_group": "B", | |
| }, | |
| "2B003": { | |
| "description": "Numerically controlled machine tools (additional)", | |
| "items_controlled": "NC drilling machines, NC boring machines", | |
| "control_reasons": ["NS"], | |
| "product_group": "B", | |
| }, | |
| "2C001": { | |
| "description": "Materials associated with chemical weapon precursors", | |
| "items_controlled": "Chemical materials for CW precursors", | |
| "control_reasons": ["CB"], | |
| "product_group": "C", | |
| }, | |
| "2C350": { | |
| "description": "Chemical weapon precursors (Category 2)", | |
| "items_controlled": "Precursor chemicals for chemical weapons", | |
| "control_reasons": ["CB"], | |
| "product_group": "C", | |
| }, | |
| "2D001": { | |
| "description": "Software for numerical control of machine tools", | |
| "items_controlled": "Software for NC machines, adaptive control software", | |
| "control_reasons": ["NS"], | |
| "product_group": "D", | |
| }, | |
| "2E001": { | |
| "description": "Technology for numerical control machine tools", | |
| "items_controlled": "Technology for development or production of NC machine tools", | |
| "control_reasons": ["NS"], | |
| "product_group": "E", | |
| }, | |
| # ββ Category 3: Electronics ββββββββββββββββββββββββββββββββββββββββ | |
| "3A001": { | |
| "description": "Electronic components", | |
| "items_controlled": "Integrated circuits, microprocessor chips, semiconductor devices, transistors", | |
| "control_reasons": ["NS"], | |
| "product_group": "A", | |
| }, | |
| "3A002": { | |
| "description": "General-purpose electronic equipment", | |
| "items_controlled": "Signal analyzers, spectrum analyzers, frequency synthesizers", | |
| "control_reasons": ["NS"], | |
| "product_group": "A", | |
| }, | |
| "3A229": { | |
| "description": "Electronic components (Missile Technology)", | |
| "items_controlled": "Electronics for missile technology applications", | |
| "control_reasons": ["MT"], | |
| "product_group": "A", | |
| }, | |
| "3A991": { | |
| "description": "Electronic components (not controlled by 3A001)", | |
| "items_controlled": "Electronic components for non-military applications", | |
| "control_reasons": ["AT"], | |
| "product_group": "A", | |
| }, | |
| "3B001": { | |
| "description": "Test equipment for electronics", | |
| "items_controlled": "Semiconductor test equipment, wafer probers", | |
| "control_reasons": ["NS"], | |
| "product_group": "B", | |
| }, | |
| "3B002": { | |
| "description": "Test equipment for electron tubes, semiconductors", | |
| "items_controlled": "Equipment for testing integrated circuits", | |
| "control_reasons": ["NS"], | |
| "product_group": "B", | |
| }, | |
| "3C001": { | |
| "description": "Electronic materials", | |
| "items_controlled": "Epitaxial materials, semiconductor substrates, photoresists", | |
| "control_reasons": ["NS"], | |
| "product_group": "C", | |
| }, | |
| "3C002": { | |
| "description": "Precursor chemicals for semiconductors", | |
| "items_controlled": "Chemical precursors for semiconductor manufacturing", | |
| "control_reasons": ["NS"], | |
| "product_group": "C", | |
| }, | |
| "3D001": { | |
| "description": "Software for electronic design and development", | |
| "items_controlled": "EDA software, IC design software, simulation software", | |
| "control_reasons": ["NS"], | |
| "product_group": "D", | |
| }, | |
| "3E001": { | |
| "description": "Technology for electronics", | |
| "items_controlled": "Technology for development or production of electronic components", | |
| "control_reasons": ["NS"], | |
| "product_group": "E", | |
| }, | |
| # ββ Category 4: Computers ββββββββββββββββββββββββββββββββββββββββββ | |
| "4A001": { | |
| "description": "Digital computers and related equipment", | |
| "items_controlled": "Digital computers, electronic assemblies, related equipment", | |
| "control_reasons": ["NS", "AT"], | |
| "product_group": "A", | |
| }, | |
| "4A002": { | |
| "description": "Computers for 'signal processing'", | |
| "items_controlled": "Computers designed for signal processing applications", | |
| "control_reasons": ["NS"], | |
| "product_group": "A", | |
| }, | |
| "4A003": { | |
| "description": "Digital computers (advanced)", | |
| "items_controlled": "High-performance digital computers and digital computer systems", | |
| "control_reasons": ["NS"], | |
| "product_group": "A", | |
| }, | |
| "4A004": { | |
| "description": "Computers for 'signal processing' (advanced)", | |
| "items_controlled": "Advanced signal processing computers", | |
| "control_reasons": ["NS"], | |
| "product_group": "A", | |
| }, | |
| "4A994": { | |
| "description": "Computers not controlled by 4A001-4A004", | |
| "items_controlled": "General-purpose computers for non-military use", | |
| "control_reasons": ["AT"], | |
| "product_group": "A", | |
| }, | |
| "4B001": { | |
| "description": "Test equipment for computers", | |
| "items_controlled": "Equipment for testing digital computers", | |
| "control_reasons": ["NS"], | |
| "product_group": "B", | |
| }, | |
| "4C001": { | |
| "description": "Chemicals for computers", | |
| "items_controlled": "Chemicals for semiconductor manufacturing in computers", | |
| "control_reasons": ["NS"], | |
| "product_group": "C", | |
| }, | |
| "4D001": { | |
| "description": "Software for computers", | |
| "items_controlled": "Operating systems, application software for controlled computers", | |
| "control_reasons": ["NS"], | |
| "product_group": "D", | |
| }, | |
| "4D995": { | |
| "description": "Software not controlled by 4D001", | |
| "items_controlled": "General-purpose software for non-military computers", | |
| "control_reasons": ["AT"], | |
| "product_group": "D", | |
| }, | |
| "4E001": { | |
| "description": "Technology for computers", | |
| "items_controlled": "Technology for development, production, or use of computers", | |
| "control_reasons": ["NS"], | |
| "product_group": "E", | |
| }, | |
| # ββ Category 5: Telecommunications & Information Security ββββββββββ | |
| "5A001": { | |
| "description": "Telecommunications systems and equipment", | |
| "items_controlled": "Telecom systems, transmission equipment, digital switching", | |
| "control_reasons": ["NS"], | |
| "product_group": "A", | |
| }, | |
| "5A002": { | |
| "description": "Information security systems and equipment", | |
| "items_controlled": "Cryptographic systems, encryption devices, VPN appliances", | |
| "control_reasons": ["EI"], | |
| "product_group": "A", | |
| }, | |
| "5A003": { | |
| "description": "Defence-related telecom equipment", | |
| "items_controlled": "Military telecom equipment, secure communications", | |
| "control_reasons": ["NS", "AT"], | |
| "product_group": "A", | |
| }, | |
| "5A991": { | |
| "description": "Telecom equipment not controlled by 5A001-5A003", | |
| "items_controlled": "General-purpose telecom equipment", | |
| "control_reasons": ["AT"], | |
| "product_group": "A", | |
| }, | |
| "5B001": { | |
| "description": "Telecom test equipment", | |
| "items_controlled": "Equipment for testing telecom systems", | |
| "control_reasons": ["NS"], | |
| "product_group": "B", | |
| }, | |
| "5B002": { | |
| "description": "Information security test equipment", | |
| "items_controlled": "Equipment for testing information security systems", | |
| "control_reasons": ["EI"], | |
| "product_group": "B", | |
| }, | |
| "5C001": { | |
| "description": "Telecom materials", | |
| "items_controlled": "Materials for telecom equipment manufacturing", | |
| "control_reasons": ["NS"], | |
| "product_group": "C", | |
| }, | |
| "5D001": { | |
| "description": "Software for telecommunications", | |
| "items_controlled": "Telecom system software, network management software", | |
| "control_reasons": ["NS"], | |
| "product_group": "D", | |
| }, | |
| "5D002": { | |
| "description": "Information security software", | |
| "items_controlled": "Encryption software, cryptographic software, security software", | |
| "control_reasons": ["EI"], | |
| "product_group": "D", | |
| }, | |
| "5E001": { | |
| "description": "Technology for telecommunications", | |
| "items_controlled": "Technology for telecom equipment development or production", | |
| "control_reasons": ["NS"], | |
| "product_group": "E", | |
| }, | |
| "5E002": { | |
| "description": "Technology for information security", | |
| "items_controlled": "Technology for development or production of cryptographic systems", | |
| "control_reasons": ["EI"], | |
| "product_group": "E", | |
| }, | |
| # ββ Category 6: Sensors and Lasers βββββββββββββββββββββββββββββββββ | |
| "6A001": { | |
| "description": "Acoustics and sonar systems", | |
| "items_controlled": "Hydrophones, sonar systems, acoustic arrays", | |
| "control_reasons": ["NS"], | |
| "product_group": "A", | |
| }, | |
| "6A002": { | |
| "description": "Optical sensors and equipment", | |
| "items_controlled": "Thermal imaging cameras, infrared sensors, night vision equipment", | |
| "control_reasons": ["NS"], | |
| "product_group": "A", | |
| }, | |
| "6A003": { | |
| "description": "Radar systems and equipment", | |
| "items_controlled": "Radar systems, antenna assemblies, signal processing equipment", | |
| "control_reasons": ["NS"], | |
| "product_group": "A", | |
| }, | |
| "6A004": { | |
| "description": "Lasers and related equipment", | |
| "items_controlled": "High-energy lasers, laser amplifiers, laser optics", | |
| "control_reasons": ["NS"], | |
| "product_group": "A", | |
| }, | |
| "6A005": { | |
| "description": "Magnetometers and related equipment", | |
| "items_controlled": "Magnetometers, magnetic field sensors, related equipment", | |
| "control_reasons": ["NS"], | |
| "product_group": "A", | |
| }, | |
| "6B001": { | |
| "description": "Acoustic test and inspection equipment", | |
| "items_controlled": "Equipment for testing acoustic systems", | |
| "control_reasons": ["NS"], | |
| "product_group": "B", | |
| }, | |
| "6B002": { | |
| "description": "Optical test equipment", | |
| "items_controlled": "Equipment for testing optical sensors", | |
| "control_reasons": ["NS"], | |
| "product_group": "B", | |
| }, | |
| "6C001": { | |
| "description": "Optical materials", | |
| "items_controlled": "Optical crystals, infrared optical materials", | |
| "control_reasons": ["NS"], | |
| "product_group": "C", | |
| }, | |
| "6D001": { | |
| "description": "Software for sensors", | |
| "items_controlled": "Software for design, development, or operation of sensors", | |
| "control_reasons": ["NS"], | |
| "product_group": "D", | |
| }, | |
| "6E001": { | |
| "description": "Technology for sensors", | |
| "items_controlled": "Technology for development or production of sensors", | |
| "control_reasons": ["NS"], | |
| "product_group": "E", | |
| }, | |
| # ββ Category 7: Navigation & Avionics ββββββββββββββββββββββββββββββ | |
| "7A001": { | |
| "description": "Accelerometers and gyroscopes", | |
| "items_controlled": "Accelerometers, gyroscopes, inertial measurement units", | |
| "control_reasons": ["NS", "MT"], | |
| "product_group": "A", | |
| }, | |
| "7A002": { | |
| "description": "Gravity meters and related equipment", | |
| "items_controlled": "Gravity meters, gravimeters, related equipment", | |
| "control_reasons": ["NS"], | |
| "product_group": "A", | |
| }, | |
| "7A003": { | |
| "description": "Navigation and avionics equipment", | |
| "items_controlled": "GPS receivers, INS, flight control systems", | |
| "control_reasons": ["NS", "AT"], | |
| "product_group": "A", | |
| }, | |
| "7B001": { | |
| "description": "Navigation test equipment", | |
| "items_controlled": "Equipment for testing navigation systems", | |
| "control_reasons": ["NS"], | |
| "product_group": "B", | |
| }, | |
| "7C001": { | |
| "description": "Navigation materials", | |
| "items_controlled": "Materials for navigation equipment manufacturing", | |
| "control_reasons": ["NS"], | |
| "product_group": "C", | |
| }, | |
| "7D001": { | |
| "description": "Software for navigation and avionics", | |
| "items_controlled": "Navigation software, flight control software", | |
| "control_reasons": ["NS"], | |
| "product_group": "D", | |
| }, | |
| "7E001": { | |
| "description": "Technology for navigation and avionics", | |
| "items_controlled": "Technology for navigation equipment development or production", | |
| "control_reasons": ["NS"], | |
| "product_group": "E", | |
| }, | |
| # ββ Category 8: Marine βββββββββββββββββββββββββββββββββββββββββββββ | |
| "8A001": { | |
| "description": "Marine vessels and related equipment", | |
| "items_controlled": "Submersibles, marine engines, related equipment", | |
| "control_reasons": ["NS", "AT"], | |
| "product_group": "A", | |
| }, | |
| "8A002": { | |
| "description": "Underwater vehicles and related equipment", | |
| "items_controlled": "Remotely operated vehicles, autonomous underwater vehicles", | |
| "control_reasons": ["NS"], | |
| "product_group": "A", | |
| }, | |
| "8B001": { | |
| "description": "Marine test equipment", | |
| "items_controlled": "Equipment for testing marine systems", | |
| "control_reasons": ["NS"], | |
| "product_group": "B", | |
| }, | |
| "8C001": { | |
| "description": "Marine materials", | |
| "items_controlled": "Materials for marine equipment manufacturing", | |
| "control_reasons": ["NS"], | |
| "product_group": "C", | |
| }, | |
| "8D001": { | |
| "description": "Software for marine systems", | |
| "items_controlled": "Software for design, development, or operation of marine systems", | |
| "control_reasons": ["NS"], | |
| "product_group": "D", | |
| }, | |
| "8E001": { | |
| "description": "Technology for marine systems", | |
| "items_controlled": "Technology for development or production of marine systems", | |
| "control_reasons": ["NS"], | |
| "product_group": "E", | |
| }, | |
| # ββ Category 9: Aerospace & Propulsion βββββββββββββββββββββββββββββ | |
| "9A001": { | |
| "description": "Aero gas turbine engines", | |
| "items_controlled": "Gas turbine engines, turbofan engines, turboprop engines", | |
| "control_reasons": ["NS", "AT"], | |
| "product_group": "A", | |
| }, | |
| "9A002": { | |
| "description": "Missile technology (rockets, missiles, drones)", | |
| "items_controlled": "Rockets, missiles, unmanned aerial vehicles, cruise missiles", | |
| "control_reasons": ["MT", "NS"], | |
| "product_group": "A", | |
| }, | |
| "9A003": { | |
| "description": "Space launch vehicles and sounding rockets", | |
| "items_controlled": "Space launch vehicles, sounding rockets, related equipment", | |
| "control_reasons": ["MT", "NS"], | |
| "product_group": "A", | |
| }, | |
| "9A004": { | |
| "description": "Space launch vehicles and sounding rockets", | |
| "items_controlled": "Space launch vehicles, sounding rockets, ground equipment", | |
| "control_reasons": ["MT", "NS"], | |
| "product_group": "A", | |
| }, | |
| "9A005": { | |
| "description": "Unmanned aerial vehicles (UAVs)", | |
| "items_controlled": "Unmanned aerial vehicles, drones, related equipment", | |
| "control_reasons": ["MT", "NS"], | |
| "product_group": "A", | |
| }, | |
| "9A007": { | |
| "description": "Aircraft and related equipment", | |
| "items_controlled": "Military aircraft, stealth aircraft, related equipment", | |
| "control_reasons": ["NS", "AT"], | |
| "product_group": "A", | |
| }, | |
| "9A008": { | |
| "description": "Aerospace propulsion systems", | |
| "items_controlled": "Liquid/solid rocket propulsion systems, ramjet/scramjet engines", | |
| "control_reasons": ["MT", "NS"], | |
| "product_group": "A", | |
| }, | |
| "9A990": { | |
| "description": "Aircraft and related equipment (non-controlled)", | |
| "items_controlled": "General aviation aircraft, non-military aircraft", | |
| "control_reasons": ["AT"], | |
| "product_group": "A", | |
| }, | |
| "9B001": { | |
| "description": "Aerospace test equipment", | |
| "items_controlled": "Equipment for testing gas turbine engines, wind tunnels", | |
| "control_reasons": ["NS"], | |
| "product_group": "B", | |
| }, | |
| "9B002": { | |
| "description": "Aero gas turbine engine test equipment", | |
| "items_controlled": "Equipment for testing aero gas turbine engines", | |
| "control_reasons": ["NS"], | |
| "product_group": "B", | |
| }, | |
| "9C001": { | |
| "description": "Aerospace materials", | |
| "items_controlled": "Materials for aerospace equipment manufacturing", | |
| "control_reasons": ["NS", "MT"], | |
| "product_group": "C", | |
| }, | |
| "9C002": { | |
| "description": "Composite materials for aerospace", | |
| "items_controlled": "Fibrous or filamentary materials for aerospace applications", | |
| "control_reasons": ["NS", "MT"], | |
| "product_group": "C", | |
| }, | |
| "9D001": { | |
| "description": "Software for aerospace", | |
| "items_controlled": "Software for design, development, or production of aerospace equipment", | |
| "control_reasons": ["NS", "MT"], | |
| "product_group": "D", | |
| }, | |
| "9E001": { | |
| "description": "Technology for aerospace", | |
| "items_controlled": "Technology for development or production of aerospace equipment", | |
| "control_reasons": ["NS", "MT"], | |
| "product_group": "E", | |
| }, | |
| "9E002": { | |
| "description": "Technology for aerospace propulsion", | |
| "items_controlled": "Technology for development or production of aerospace propulsion systems", | |
| "control_reasons": ["NS", "MT"], | |
| "product_group": "E", | |
| }, | |
| "9E003": { | |
| "description": "Technology for missile technology", | |
| "items_controlled": "Technology for development or production of missiles", | |
| "control_reasons": ["MT"], | |
| "product_group": "E", | |
| }, | |
| # ββ Legacy / Miscellaneous βββββββββββββββββββββββββββββββββββββββββ | |
| "6A008": { | |
| "description": "Radar systems (specialized)", | |
| "items_controlled": "Specialized radar systems for military applications", | |
| "control_reasons": ["NS"], | |
| "product_group": "A", | |
| }, | |
| "6A006": { | |
| "description": "Magnetometers (specialized)", | |
| "items_controlled": "Specialized magnetometers for military applications", | |
| "control_reasons": ["NS"], | |
| "product_group": "A", | |
| }, | |
| } | |
| # ββ Data Models βββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| class ECCNResult: | |
| """Represents an ECCN classification result. | |
| Attributes: | |
| eccn_code: The classified ECCN code (e.g., "3A001"). | |
| confidence: Classification confidence from 0.0 to 1.0. | |
| category: CCL category number (0-9). | |
| product_group: Product group letter (A-E). | |
| description: Human-readable description of the ECCN. | |
| control_reasons: List of reasons for control (NS, MT, CB, etc.). | |
| license_requirement: License requirement string. | |
| """ | |
| eccn_code: str | |
| confidence: float | |
| category: int | |
| product_group: str | |
| description: str | |
| control_reasons: list[str] | |
| license_requirement: str | |
| def __post_init__(self) -> None: | |
| """Validate and normalize the result after creation.""" | |
| self.eccn_code = self.eccn_code.upper().strip() | |
| if len(self.eccn_code) >= 1 and self.eccn_code[0].isdigit(): | |
| self.category = int(self.eccn_code[0]) | |
| if len(self.eccn_code) >= 2: | |
| self.product_group = self.eccn_code[1].upper() | |
| def needs_review(self) -> bool: | |
| """Whether human review is recommended (low confidence or LLM-only).""" | |
| return self.confidence < MEDIUM_CONFIDENCE_THRESHOLD | |
| class ECCNInfo: | |
| """Detailed information about an ECCN code. | |
| Attributes: | |
| eccn_code: The ECCN code. | |
| category: CCL category number (0-9). | |
| product_group: Product group letter (A-E). | |
| description: Human-readable description. | |
| items_controlled: Specific items controlled under this ECCN. | |
| control_reasons: List of reasons for control. | |
| country_chart: Reason β license requirement per country group. | |
| """ | |
| eccn_code: str | |
| category: int | |
| product_group: str | |
| description: str | |
| items_controlled: str | |
| control_reasons: list[str] | |
| country_chart: dict[str, str] | |
| def __post_init__(self) -> None: | |
| """Validate and normalize the info after creation.""" | |
| self.eccn_code = self.eccn_code.upper().strip() | |
| if len(self.eccn_code) >= 1 and self.eccn_code[0].isdigit(): | |
| self.category = int(self.eccn_code[0]) | |
| if len(self.eccn_code) >= 2: | |
| self.product_group = self.eccn_code[1].upper() | |
| # ββ ECCN Validator ββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| class ECCNValidator: | |
| """Validates ECCN code structure. | |
| Provides static methods for structural validation of ECCN codes | |
| according to EAR Part 774 rules. | |
| """ | |
| def is_valid_eccn(code: str) -> bool: | |
| """Validate ECCN code structure. | |
| Args: | |
| code: The code to validate. | |
| Returns: | |
| True if code is a valid 5-character ECCN code. | |
| """ | |
| normalized: str = code.upper().strip() | |
| if len(normalized) != 5: | |
| return False | |
| if not normalized[0].isdigit(): | |
| return False | |
| category: int = int(normalized[0]) | |
| if category < 0 or category > 9: | |
| return False | |
| if normalized[1] not in "ABCDE": | |
| return False | |
| if not normalized[2:5].isdigit(): | |
| return False | |
| return True | |
| def parse(code: str) -> dict[str, Any]: | |
| """Parse ECCN code into components. | |
| Args: | |
| code: The ECCN code to parse. | |
| Returns: | |
| Dict with keys: category, product_group, group_name, category_name. | |
| Raises: | |
| ECCNValidationError: If code is not a valid ECCN. | |
| """ | |
| if not ECCNValidator.is_valid_eccn(code): | |
| raise ECCNValidationError( | |
| code=code, | |
| reason="Must be 5 characters: digit (0-9) + letter (A-E) + 3 digits", | |
| ) | |
| normalized: str = code.upper().strip() | |
| category: int = int(normalized[0]) | |
| product_group: str = normalized[1] | |
| return { | |
| "category": category, | |
| "product_group": product_group, | |
| "category_name": CCL_CATEGORIES.get(category, "Unknown"), | |
| "group_name": PRODUCT_GROUPS.get(product_group, "Unknown"), | |
| } | |
| # ββ Main Classification Engine ββββββββββββββββββββββββββββββββββββββββ | |
| class ECCNClassifier: | |
| """Main ECCN classification interface. | |
| Combines a static ECCN database with LLM-based classification for | |
| optimal accuracy. Static lookup handles known items; LLM handles | |
| novel or ambiguous items. | |
| Attributes: | |
| _eccn_db: Static ECCN database (embedded). | |
| """ | |
| def __init__(self) -> None: | |
| """Initialize the ECCN classifier with the static database.""" | |
| self._eccn_db: dict[str, dict[str, Any]] = dict(_COMMON_ECCNS) | |
| self._llm_provider: str = os.environ.get("MODEL_PROVIDER", "gemini") | |
| self._llm_model: str = os.environ.get("MODEL_NAME", "gemini-2.5-flash") | |
| self._llm_api_key: str = ( | |
| os.environ.get("GEMINI_API_KEY") | |
| or os.environ.get("GOOGLE_API_KEY") | |
| or os.environ.get("MODEL_API_KEY", "") | |
| ) | |
| def classify(self, product_description: str) -> ECCNResult: | |
| """Classify a product description to an ECCN code. | |
| Pipeline: | |
| 1. Check static ECCN database for exact/description match | |
| 2. Use LLM classification if no static match found | |
| 3. Return result with confidence score | |
| Args: | |
| product_description: Product description to classify. | |
| Returns: | |
| ECCNResult with classification details and confidence. | |
| """ | |
| start_time: float = _time.time() | |
| # Step 1: Static database lookup (keyword matching) | |
| static_result: Optional[ECCNResult] = self._static_lookup(product_description) | |
| if static_result is not None and static_result.confidence >= HIGH_CONFIDENCE_THRESHOLD: | |
| logger.info( | |
| "Static ECCN lookup HIT β %s (%.0f%%)", | |
| static_result.eccn_code, | |
| static_result.confidence * 100, | |
| ) | |
| return static_result | |
| # Step 2: LLM classification | |
| llm_result: Optional[ECCNResult] = self._classify_with_llm(product_description) | |
| if llm_result is not None: | |
| logger.info( | |
| "LLM ECCN classification β %s (%.0f%%, %.1fms)", | |
| llm_result.eccn_code, | |
| llm_result.confidence * 100, | |
| (_time.time() - start_time) * 1000, | |
| ) | |
| return llm_result | |
| # Step 3: Return best available result | |
| if static_result is not None: | |
| return static_result | |
| # Fallback: unknown classification | |
| return ECCNResult( | |
| eccn_code="0A999", | |
| confidence=0.0, | |
| category=0, | |
| product_group="A", | |
| description="Unable to classify β manual review required", | |
| control_reasons=[], | |
| license_requirement="Unknown β consult BIS", | |
| ) | |
| def lookup(self, eccn_code: str) -> ECCNInfo: | |
| """Look up an ECCN code for detailed information. | |
| Args: | |
| eccn_code: The ECCN code to look up (e.g., "3A001"). | |
| Returns: | |
| ECCNInfo with full details about the ECCN. | |
| Raises: | |
| ECCNValidationError: If code structure is invalid. | |
| ECCNLookupError: If code is not found in the database. | |
| """ | |
| normalized: str = eccn_code.upper().strip() | |
| if not ECCNValidator.is_valid_eccn(normalized): | |
| raise ECCNValidationError( | |
| code=eccn_code, | |
| reason="Must be 5 characters: digit (0-9) + letter (A-E) + 3 digits", | |
| ) | |
| info: Optional[dict[str, Any]] = self._eccn_db.get(normalized) | |
| if info is None: | |
| raise ECCNLookupError(f"ECCN code '{normalized}' not found in database") | |
| category: int = int(normalized[0]) | |
| product_group: str = normalized[1] | |
| control_reasons: list[str] = info.get("control_reasons", []) | |
| country_chart: dict[str, str] = {} | |
| for reason in control_reasons: | |
| country_chart[reason] = COUNTRY_CHART.get(reason, "License requirement not specified") | |
| return ECCNInfo( | |
| eccn_code=normalized, | |
| category=category, | |
| product_group=product_group, | |
| description=info.get("description", ""), | |
| items_controlled=info.get("items_controlled", ""), | |
| control_reasons=control_reasons, | |
| country_chart=country_chart, | |
| ) | |
| def get_control_reason(self, eccn_code: str) -> list[str]: | |
| """Get reasons for control for an ECCN code. | |
| Args: | |
| eccn_code: The ECCN code to look up. | |
| Returns: | |
| List of control reason codes (e.g., ["NS", "AT"]). | |
| Raises: | |
| ECCNValidationError: If code structure is invalid. | |
| ECCNLookupError: If code is not found in the database. | |
| """ | |
| info: ECCNInfo = self.lookup(eccn_code) | |
| return info.control_reasons | |
| def _static_lookup(self, description: str) -> Optional[ECCNResult]: | |
| """Search the static ECCN database for a matching description. | |
| Uses keyword matching against descriptions and items_controlled | |
| fields in the embedded database. | |
| Args: | |
| description: Product description to search for. | |
| Returns: | |
| ECCNResult if a match is found, None otherwise. | |
| """ | |
| desc_lower: str = description.lower().strip() | |
| words: set[str] = set(re.findall(r"[a-zA-Z]{3,}", desc_lower)) | |
| best_match: Optional[tuple[str, float, dict[str, Any]]] = None | |
| for eccn_code, info in self._eccn_db.items(): | |
| desc_text: str = info.get("description", "").lower() | |
| items_text: str = info.get("items_controlled", "").lower() | |
| combined: str = f"{desc_text} {items_text}" | |
| # Exact substring match | |
| if desc_lower in combined or combined in desc_lower: | |
| score: float = 0.95 | |
| if best_match is None or score > best_match[1]: | |
| best_match = (eccn_code, score, info) | |
| continue | |
| # Token overlap matching | |
| combined_words: set[str] = set(re.findall(r"[a-zA-Z]{3,}", combined)) | |
| if not combined_words: | |
| continue | |
| overlap: set[str] = words & combined_words | |
| if not overlap: | |
| continue | |
| overlap_ratio: float = len(overlap) / len(words) if words else 0.0 | |
| score = 0.6 + (overlap_ratio * 0.3) | |
| score = min(score, 0.90) | |
| if best_match is None or score > best_match[1]: | |
| best_match = (eccn_code, score, info) | |
| if best_match is None: | |
| return None | |
| eccn_code, confidence, info = best_match | |
| category: int = int(eccn_code[0]) | |
| product_group: str = eccn_code[1] | |
| control_reasons: list[str] = info.get("control_reasons", []) | |
| license_parts: list[str] = [] | |
| for reason in control_reasons: | |
| req: str = COUNTRY_CHART.get(reason, "License requirement not specified") | |
| license_parts.append(req) | |
| return ECCNResult( | |
| eccn_code=eccn_code, | |
| confidence=confidence, | |
| category=category, | |
| product_group=product_group, | |
| description=info.get("description", ""), | |
| control_reasons=control_reasons, | |
| license_requirement="; ".join(license_parts) if license_parts else "No license required", | |
| ) | |
| def _classify_with_llm(self, description: str) -> Optional[ECCNResult]: | |
| """Use LLM for ECCN classification. | |
| Sends a prompt to Gemini 2.5 Flash via Google AI API | |
| and parses the structured response. | |
| Args: | |
| description: Product description to classify. | |
| Returns: | |
| ECCNResult from LLM, or None if unavailable/failed. | |
| """ | |
| prompt: str = self._build_classification_prompt(description) | |
| if not self._llm_api_key: | |
| logger.warning("No LLM API key configured β skipping LLM classification") | |
| return None | |
| try: | |
| url = ( | |
| f"https://generativelanguage.googleapis.com/v1beta/" | |
| f"models/{self._llm_model}:generateContent?key={self._llm_api_key}" | |
| ) | |
| payload = { | |
| "contents": [{"parts": [{"text": prompt}]}], | |
| "generationConfig": { | |
| "temperature": 0.1, | |
| "maxOutputTokens": 4096, | |
| }, | |
| } | |
| with httpx.Client(timeout=30.0) as client: | |
| resp = client.post(url, json=payload) | |
| resp.raise_for_status() | |
| data = resp.json() | |
| content = data["candidates"][0]["content"]["parts"][0]["text"] | |
| return self._parse_llm_response(content, description) | |
| except Exception as exc: | |
| logger.error("LLM ECCN classification failed: %s", exc) | |
| return None | |
| def _build_classification_prompt(self, description: str) -> str: | |
| """Build prompt for LLM ECCN classification. | |
| Args: | |
| description: Product description to classify. | |
| Returns: | |
| Formatted prompt string. | |
| """ | |
| categories_text: str = "\n".join( | |
| f" Category {cat_id}: {cat_name}" | |
| for cat_id, cat_name in sorted(CCL_CATEGORIES.items()) | |
| ) | |
| groups_text: str = "\n".join( | |
| f" Group {grp_id}: {grp_name}" | |
| for grp_id, grp_name in PRODUCT_GROUPS.items() | |
| ) | |
| return ( | |
| "You are an expert export control compliance officer specializing in " | |
| "ECCN (Export Control Classification Number) classification under the " | |
| "US Bureau of Industry and Security (BIS) Export Administration " | |
| "Regulations (EAR).\n\n" | |
| "The Commerce Control List (CCL) has 10 categories (0-9) and " | |
| "5 product groups (A-E):\n\n" | |
| f"CCL Categories:\n{categories_text}\n\n" | |
| f"Product Groups:\n{groups_text}\n\n" | |
| f"Classify the following product to an ECCN code:\n\n" | |
| f"Product Description: {description}\n\n" | |
| "Provide your classification in JSON format:\n" | |
| "{\n" | |
| ' "eccn_code": "XYYYY (e.g., 3A001)",\n' | |
| ' "confidence": 0.0-1.0,\n' | |
| ' "category": 0-9,\n' | |
| ' "product_group": "A-E",\n' | |
| ' "description": "Brief description of the ECCN classification",\n' | |
| ' "control_reasons": ["NS", "AT", ...],\n' | |
| ' "license_requirement": "License requirement summary"\n' | |
| "}\n\n" | |
| "Common control reasons:\n" | |
| " NS = National Security\n" | |
| " MT = Missile Technology\n" | |
| " CB = Chemical & Biological Weapons\n" | |
| " CC = Chemical & Biological Weapons (end-use)\n" | |
| " RS = Regional Stability\n" | |
| " AT = Anti-Terrorism\n" | |
| " EI = Encryption Items\n" | |
| " SL = Surreptitious Listening\n\n" | |
| "Rules:\n" | |
| "1. Use the 5-character ECCN format: digit (category) + letter (group) + 3 digits\n" | |
| "2. If the item is not controlled under any ECCN, use EAR99\n" | |
| "3. Provide the most specific ECCN that applies\n" | |
| "4. Include applicable control reasons\n" | |
| "5. Return ONLY the JSON object, no other text" | |
| ) | |
| def _parse_llm_response( | |
| self, content: str, original_description: str | |
| ) -> Optional[ECCNResult]: | |
| """Parse LLM response into ECCNResult. | |
| LLM self-reported confidence is capped at LLM_MAX_CONFIDENCE (0.85) | |
| because LLMs may output inflated confidence for hallucinated codes. | |
| Args: | |
| content: Raw LLM response text. | |
| original_description: Original product description. | |
| Returns: | |
| ECCNResult if parsing succeeds, None otherwise. | |
| """ | |
| try: | |
| # Strip markdown code blocks if present | |
| cleaned: str = content.strip() | |
| if cleaned.startswith("```"): | |
| first_newline: int = cleaned.find("\n") | |
| if first_newline != -1: | |
| cleaned = cleaned[first_newline + 1:] | |
| if cleaned.rstrip().endswith("```"): | |
| cleaned = cleaned.rstrip()[:-len("```")].rstrip() | |
| # Extract outermost JSON block | |
| start: int = cleaned.find("{") | |
| if start == -1: | |
| logger.error("No JSON found in LLM response") | |
| return None | |
| depth: int = 0 | |
| end: int = -1 | |
| for i in range(start, len(cleaned)): | |
| if cleaned[i] == "{": | |
| depth += 1 | |
| elif cleaned[i] == "}": | |
| depth -= 1 | |
| if depth == 0: | |
| end = i | |
| break | |
| if end == -1: | |
| logger.error("Unbalanced JSON braces in LLM response") | |
| return None | |
| data: dict[str, Any] = json.loads(cleaned[start:end + 1]) | |
| eccn_code: str = data.get("eccn_code", "") | |
| # Normalize: strip dots, dashes, spaces | |
| eccn_code = re.sub(r"[.\-\s]", "", eccn_code).upper() | |
| # Validate ECCN format | |
| if not ECCNValidator.is_valid_eccn(eccn_code): | |
| logger.warning("LLM returned invalid ECCN: %s", eccn_code) | |
| return None | |
| raw_confidence: float = data.get("confidence", 0.5) | |
| capped_confidence: float = min(raw_confidence, LLM_MAX_CONFIDENCE) | |
| if raw_confidence > LLM_MAX_CONFIDENCE: | |
| logger.warning( | |
| "LLM self-reported confidence %.2f capped to %.2f", | |
| raw_confidence, | |
| LLM_MAX_CONFIDENCE, | |
| ) | |
| category: int = int(eccn_code[0]) | |
| product_group: str = eccn_code[1] | |
| control_reasons: list[str] = data.get("control_reasons", []) | |
| license_req: str = data.get("license_requirement", "") | |
| # Build license requirement from control reasons if not provided | |
| if not license_req and control_reasons: | |
| license_parts: list[str] = [] | |
| for reason in control_reasons: | |
| req: str = COUNTRY_CHART.get(reason, "License requirement not specified") | |
| license_parts.append(req) | |
| license_req = "; ".join(license_parts) | |
| return ECCNResult( | |
| eccn_code=eccn_code, | |
| confidence=capped_confidence, | |
| category=category, | |
| product_group=product_group, | |
| description=data.get("description", ""), | |
| control_reasons=control_reasons, | |
| license_requirement=license_req, | |
| ) | |
| except Exception as exc: | |
| logger.error("Failed to parse LLM ECCN response: %s", exc) | |
| return None | |
| # ββ Convenience Functions βββββββββββββββββββββββββββββββββββββββββββββ | |
| _engine: Optional[ECCNClassifier] = None | |
| _engine_lock: threading.Lock = threading.Lock() | |
| def get_engine() -> ECCNClassifier: | |
| """Get or create the global ECCN classifier (thread-safe). | |
| Returns: | |
| The singleton ECCNClassifier instance. | |
| """ | |
| global _engine | |
| if _engine is None: | |
| with _engine_lock: | |
| if _engine is None: | |
| _engine = ECCNClassifier() | |
| return _engine | |
| def classify_product(description: str) -> ECCNResult: | |
| """Classify a product to an ECCN code. | |
| Args: | |
| description: Product description. | |
| Returns: | |
| ECCNResult with classification details and confidence. | |
| """ | |
| engine: ECCNClassifier = get_engine() | |
| return engine.classify(description) | |
| def lookup_eccn(eccn_code: str) -> ECCNInfo: | |
| """Look up an ECCN code for detailed information. | |
| Args: | |
| eccn_code: The ECCN code to look up. | |
| Returns: | |
| ECCNInfo with full details about the ECCN. | |
| """ | |
| engine: ECCNClassifier = get_engine() | |
| return engine.lookup(eccn_code) | |
| def get_control_reasons(eccn_code: str) -> list[str]: | |
| """Get reasons for control for an ECCN code. | |
| Args: | |
| eccn_code: The ECCN code to look up. | |
| Returns: | |
| List of control reason codes. | |
| """ | |
| engine: ECCNClassifier = get_engine() | |
| return engine.get_control_reason(eccn_code) | |