Spaces:
Paused
Paused
| """ | |
| HS Code Classification Engine with pyhscodes + LLM hybrid approach. | |
| Classification pipeline: | |
| 1. Local semantic cache → instant response for repeated products (FREE, zero-latency) | |
| 2. pyhscodes fuzzy search → 6-digit WCO base code (FREE, local) | |
| 3. LLM refinement → 10-digit HTSUS / 12-digit TARIC (requires API key) | |
| 4. Confidence scoring + human review flagging | |
| Data sources: | |
| - ComplianceReviewItem PostgreSQL table → cached classifications (ILIKE match) | |
| - pyhscodes: 6,940+ WCO HS codes (LGPL-2.1, FREE) | |
| - US HTSUS: 10-digit codes (public domain structure) | |
| - EU TARIC: 12-digit codes (public domain structure) | |
| Enterprise features: | |
| - Async batch classification with asyncio.gather + Semaphore(5) | |
| - Local semantic cache via PostgreSQL ILIKE fuzzy matching | |
| - Thread-safe singleton engine | |
| Cost savings: cache + pyhscodes handles 80%+ of classification, LLM only for edge cases. | |
| Typing conventions: | |
| All public APIs use explicit type hints. Literal types enforce valid | |
| code levels and sources. Final constants prevent mutation. | |
| """ | |
| from __future__ import annotations | |
| import asyncio | |
| import hashlib | |
| import json | |
| import logging | |
| import re | |
| import threading | |
| import time as _time | |
| from dataclasses import dataclass, field | |
| from typing import Annotated, Any, Final, Literal, Optional | |
| import jellyfish | |
| from difflib import SequenceMatcher | |
| logger: Final = logging.getLogger(__name__) | |
| try: | |
| from pyhscodes import hscodes as _pyhscodes | |
| PYHSCODES_AVAILABLE: Final[bool] = True | |
| except ImportError: | |
| _pyhscodes = None | |
| PYHSCODES_AVAILABLE = False | |
| logger.warning("pyhscodes not installed. Install with: pip install pyhscodes") | |
| # ── Domain Exceptions ───────────────────────────────────────────────── | |
| class HSClassificationError(Exception): | |
| """Base exception for all HS classification errors.""" | |
| class HSCodeValidationError(HSClassificationError): | |
| """Raised when an HS 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 HS code '{code}': {reason}") | |
| class HSClassificationHallucinationError(HSClassificationError): | |
| """Raised when LLM returns a code that contradicts pyhscodes database. | |
| Attributes: | |
| llm_code: The LLM-generated code. | |
| llm_chapter: The chapter from LLM code. | |
| pyh_chapter: The chapter from pyhscodes. | |
| """ | |
| def __init__(self, llm_code: str, llm_chapter: str, pyh_chapter: str) -> None: | |
| self.llm_code = llm_code | |
| self.llm_chapter = llm_chapter | |
| self.pyh_chapter = pyh_chapter | |
| super().__init__( | |
| f"LLM chapter {llm_chapter} contradicts pyhscodes chapter {pyh_chapter} " | |
| f"(code: {llm_code})" | |
| ) | |
| class HSCodeLookupError(HSClassificationError): | |
| """Raised when an HS code lookup fails in the database.""" | |
| class LLMClassificationError(HSClassificationError): | |
| """Raised when LLM-based classification fails. | |
| Attributes: | |
| provider: The LLM provider (openai, anthropic). | |
| 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 ───────────────────────────────────────────────────────── | |
| # Confidence thresholds | |
| HIGH_CONFIDENCE_THRESHOLD: Final[float] = 0.85 | |
| MEDIUM_CONFIDENCE_THRESHOLD: Final[float] = 0.65 | |
| LOW_CONFIDENCE_THRESHOLD: Final[float] = 0.45 | |
| NEEDS_REVIEW_THRESHOLD: Final[float] = 0.7 | |
| LLM_MAX_CONFIDENCE: Final[float] = 0.85 | |
| # HS code structure patterns | |
| HS2_PATTERN: Final[re.Pattern[str]] = re.compile(r"^\d{2}$") | |
| HS4_PATTERN: Final[re.Pattern[str]] = re.compile(r"^\d{4}$") | |
| HS6_PATTERN: Final[re.Pattern[str]] = re.compile(r"^\d{6}$") | |
| HTSUS_PATTERN: Final[re.Pattern[str]] = re.compile(r"^\d{10}$") | |
| TARIC_PATTERN: Final[re.Pattern[str]] = re.compile(r"^\d{12}$") | |
| # Valid HS code digit lengths | |
| VALID_CODE_LEVELS: Final[frozenset[int]] = frozenset({2, 4, 6, 10, 12}) | |
| # Type aliases | |
| CodeLevel = Literal[2, 4, 6, 10, 12] | |
| ClassificationSource = Literal["pyhscodes", "llm-openai", "llm-anthropic", "llm-opencode", "llm-gemini", "manual", "none"] | |
| TargetDigits = Literal[6, 10, 12] | |
| # Ambiguous terms that always trigger human review | |
| _AMBIGUOUS_TERMS: Final[frozenset[str]] = frozenset({ | |
| "various", "mixed", "assorted", "multiple", "miscellaneous", | |
| "general", "sundry", | |
| }) | |
| # LLM stop words | |
| _LLM_STOP_WORDS: Final[frozenset[str]] = frozenset({ | |
| "the", "a", "an", "and", "or", "of", "for", "in", "on", "at", "to", | |
| "with", "by", "from", "is", "are", "was", "were", "be", "been", | |
| }) | |
| # HS Section names | |
| SECTION_NAMES: Final[dict[str, str]] = { | |
| "I": "Live Animals", | |
| "II": "Vegetable Products", | |
| "III": "Animal/Vegetable Fats", | |
| "IV": "Prepared Foodstuffs", | |
| "V": "Mineral Products", | |
| "VI": "Chemical Products", | |
| "VII": "Plastics/Rubber", | |
| "VIII": "Raw Hides/Skins/Leather", | |
| "IX": "Wood/Cork/Straw", | |
| "X": "Pulp/Paper", | |
| "XI": "Textiles", | |
| "XII": "Footwear/Headgear", | |
| "XIII": "Stone/Ceramic/Glass", | |
| "XIV": "Precious Stones/Metals", | |
| "XV": "Base Metals", | |
| "XVI": "Machinery", | |
| "XVII": "Vehicles/Aircraft", | |
| "XVIII": "Optical/Photographic/Medical", | |
| "XIX": "Arms/Ammunition", | |
| "XX": "Miscellaneous Manufactured", | |
| "XXI": "Works of Art/Antiques", | |
| } | |
| # HS Chapter descriptions for LLM context | |
| CHAPTER_DESCRIPTIONS: Final[dict[str, str]] = { | |
| "01": "Live animals", | |
| "02": "Meat and edible meat offal", | |
| "03": "Fish and crustaceans, molluscs", | |
| "04": "Dairy produce, birds' eggs, honey", | |
| "05": "Products of animal origin, n.e.s.", | |
| "06": "Live trees and other plants, bulbs, roots", | |
| "07": "Edible vegetables and certain roots and tubers", | |
| "08": "Edible fruit and nuts, peel of citrus fruit or melons", | |
| "09": "Coffee, tea, maté and spices", | |
| "10": "Cereals", | |
| "11": "Products of the milling industry, malt, starches", | |
| "12": "Oil seeds and oleaginous fruits", | |
| "13": "Lac, gums, resins", | |
| "14": "Vegetable plaiting materials", | |
| "15": "Animal or vegetable fats and oils, cleavage products", | |
| "16": "Preparations of meat, fish or crustaceans", | |
| "17": "Sugars and sugar confectionery", | |
| "18": "Cocoa and cocoa preparations", | |
| "19": "Preparations of cereals, flour, starch or milk", | |
| "20": "Preparations of vegetables, fruit, nuts", | |
| "21": "Miscellaneous edible preparations", | |
| "22": "Beverages, spirits and vinegar", | |
| "23": "Residues and waste from the food industries, animal fodder", | |
| "24": "Tobacco and manufactured tobacco substitutes", | |
| "25": "Salt, sulphur, earths, stone, plaster, lime and cement", | |
| "26": "Ores, slag and ash", | |
| "27": "Mineral fuels, mineral oils, bituminous substances", | |
| "28": "Inorganic chemicals, organic/inorganic compounds", | |
| "29": "Organic chemicals", | |
| "30": "Pharmaceutical products", | |
| "31": "Fertilisers", | |
| "32": "Tanning or dyeing extracts, paints, putties", | |
| "33": "Essential oils, perfumery, cosmetic preparations", | |
| "34": "Soap, organic surface-active agents, waxes", | |
| "35": "Albuminoidal substances, modified starches, glues", | |
| "36": "Explosives, pyrotechnic products, matches", | |
| "37": "Photographic or cinematographic goods", | |
| "38": "Miscellaneous chemical products", | |
| "39": "Plastics and articles thereof", | |
| "40": "Rubber and articles thereof", | |
| "41": "Raw hides and skins (other than furskins) and leather", | |
| "42": "Articles of leather, saddlery and harness, travel goods", | |
| "43": "Furskins and artificial fur, manufactures thereof", | |
| "44": "Wood and articles of wood, wood charcoal", | |
| "45": "Cork and articles of cork", | |
| "46": "Basketwork, wickerwork", | |
| "47": "Pulp of wood, fibrous cellulosic material", | |
| "48": "Paper and paperboard, articles thereof", | |
| "49": "Printed books, newspapers, pictures", | |
| "50": "Silk", | |
| "51": "Wool, fine or coarse animal hair, horsehair yarn", | |
| "52": "Cotton", | |
| "53": "Other vegetable textile fibres, paper yarn", | |
| "54": "Man-made filaments, strips of man-made textile materials", | |
| "55": "Man-made staple fibres", | |
| "56": "Wadding, felt and nonwovens, special yarns", | |
| "57": "Carpets and other textile floor coverings", | |
| "58": "Special woven fabrics, tufted textile fabrics, lace", | |
| "59": "Impregnated, coated, covered or laminated textile fabrics", | |
| "60": "Knitted or crocheted fabrics", | |
| "61": "Articles of apparel and clothing accessories, knitted or crocheted", | |
| "62": "Articles of apparel and clothing accessories, not knitted or crocheted", | |
| "63": "Other made up textile articles, worn clothing", | |
| "64": "Footwear, gaiters and the like, parts thereof", | |
| "65": "Headgear and parts thereof", | |
| "66": "Umbrellas, sun umbrellas, walking-sticks, whips", | |
| "67": "Prepared feathers and down, artificial flowers", | |
| "68": "Articles of stone, plaster, cement, asbestos, mica", | |
| "69": "Ceramic products", | |
| "70": "Glass and glassware", | |
| "71": "Natural or cultured pearls, precious stones, precious metals, coin", | |
| "72": "Iron and steel", | |
| "73": "Articles of iron or steel", | |
| "74": "Copper and articles thereof", | |
| "75": "Nickel and articles thereof", | |
| "76": "Aluminium and articles thereof", | |
| "77": "Reserved", | |
| "78": "Lead and articles thereof", | |
| "79": "Zinc and articles thereof", | |
| "80": "Tin and articles thereof", | |
| "81": "Other base metals, cermets, articles thereof", | |
| "82": "Tools, implements, cutlery, spoons and forks, base metal", | |
| "83": "Miscellaneous articles of base metal", | |
| "84": "Nuclear reactors, boilers, machinery and mechanical appliances", | |
| "85": "Electrical machinery and equipment, sound recorders, television", | |
| "86": "Railway or tramway locomotives, rolling-stock", | |
| "87": "Vehicles other than railway or tramway rolling-stock", | |
| "88": "Aircraft, spacecraft, and parts thereof", | |
| "89": "Ships, boats and floating structures", | |
| "90": "Optical, photographic, measuring, checking, precision instruments", | |
| "91": "Clocks and watches and parts thereof", | |
| "92": "Musical instruments, parts and accessories thereof", | |
| "93": "Arms and ammunition, parts and accessories thereof", | |
| "94": "Furniture, bedding, lamps and lighting, prefabricated buildings", | |
| "95": "Toys, games and sports requisites, parts and accessories thereof", | |
| "96": "Miscellaneous manufactured articles", | |
| "97": "Works of art, collectors' pieces and antiques", | |
| } | |
| # Keyword → HS code mapping for pre-filtering (common product terms) | |
| KEYWORD_HS_MAPPING: Final[dict[str, str]] = { | |
| # ── Compound keywords (longest match priority) ── | |
| "organic cotton t-shirt": "610910", "silk evening dress": "620449", | |
| "wool blend coat": "620110", "leather handbag": "420221", | |
| "stainless steel pipe": "730630", "aluminum sheet": "760612", | |
| "copper wire": "740811", "steel beam": "721633", | |
| "plastic water bottle": "392330", "glass jar": "701090", | |
| "ceramic tile": "690721", "wooden floor": "441112", | |
| "cardboard box": "481910", "rubber hose": "400932", | |
| "cotton bed linen": "630231", "silk scarf": "611710", | |
| "wool blanket": "630140", "nylon backpack": "420292", | |
| "leather wallet": "420231", "canvas bag": "420292", | |
| "suede boots": "640391", "rubber sole shoe": "640411", | |
| "velvet curtain": "630391", "linen tablecloth": "630260", | |
| "cashmere sweater": "611020", "polyester shirt": "620520", | |
| "denim pants": "620342", "corduroy jacket": "620193", | |
| "terry towel": "630260", "fleece jacket": "611020", | |
| "down sleeping bag": "630790", "insulated jacket": "620193", | |
| "safety helmet": "650610", "work gloves": "420330", | |
| "steel toe boot": "640340", "reflective vest": "621139", | |
| "fire extinguisher": "842410", "smoke detector": "853110", | |
| "cctv camera": "852589", "security camera": "852589", | |
| "wireless speaker": "851822", "bluetooth speaker": "851822", | |
| "sound bar": "851822", "subwoofer": "851822", | |
| "usb hub": "851762", "power bank": "850760", | |
| "lithium ion battery": "850760", "lithium battery": "850760", | |
| "rechargeable battery": "850760", "alkaline battery": "850650", | |
| "car battery": "850710", "battery pack": "850760", | |
| "polyester jacket": "620193", "leather jacket": "620193", | |
| "down jacket": "620110", "cotton t-shirt": "610910", | |
| "cotton shirt": "620520", "cotton dress": "620442", | |
| "cotton bed sheet": "630221", "cotton pillowcase": "630231", | |
| "cotton towel": "630260", "cotton blanket": "630221", | |
| "cotton yarn": "520511", "cotton thread": "520411", | |
| "cotton fabric": "520899", "cotton cloth": "520899", | |
| "wool carpet": "570110", "wool rug": "570110", | |
| "wool sweater": "611020", "wool coat": "620110", | |
| "wool suit": "620331", "wool trousers": "620342", | |
| "silk dress": "620449", "silk scarf": "611710", | |
| "silk tie": "621520", "silk shirt": "620560", | |
| "leather boots": "640391", "leather shoes": "640399", | |
| "leather gloves": "420330", "leather bag": "420221", | |
| "rubber shoes": "640299", "rubber boots": "640299", | |
| "rubber sole": "640411", "rubber tire": "401110", | |
| "nylon bag": "420292", "nylon backpack": "420292", | |
| "denim jeans": "620342", "denim jacket": "620193", | |
| "satin dress": "620449", | |
| "coffee beans": "090111", "coffee roasted": "090121", | |
| "green tea": "090210", "black tea": "090230", | |
| "olive oil virgin": "150910", "sunflower seed oil": "151219", | |
| "natural gas": "271111", "crude oil": "270900", | |
| "diesel fuel": "271019", "gasoline": "271012", | |
| "kerosene": "271019", "jet fuel": "271019", | |
| "sodium hydroxide": "281511", "sulfuric acid": "280700", | |
| "hydrochloric acid": "280610", "nitric acid": "280800", | |
| "phosphoric acid": "280920", | |
| "usb type c cable": "854442", "usb cable": "854442", | |
| "hdmi cable": "854442", "ethernet cable": "854442", | |
| "power cable": "854442", | |
| "led headlight": "851220", "led lamp": "940542", | |
| "led strip": "940542", "led panel": "940542", | |
| "brake pad": "870830", "brake disc": "870830", | |
| "spark plug": "851110", "ignition coil": "851130", | |
| "tractor": "870110", "combine harvester": "843351", | |
| "plough": "843010", "harvester": "843351", | |
| "blender kitchen": "850940", "food processor": "850940", | |
| "wheat flour": "110100", "wheat bread": "190510", | |
| "white sugar": "170114", "raw sugar": "170111", | |
| "brown sugar": "170112", "powdered sugar": "170199", | |
| "milk powder": "040210", "condensed milk": "040219", | |
| "salmon fillet": "030481", "salmon fresh": "030214", | |
| "cheddar cheese": "040610", "mozzarella cheese": "040610", | |
| "pasta spaghetti": "190211", "pasta penne": "190211", | |
| "work overalls": "620343", "overall": "620343", | |
| "hydraulic press": "846291", "grinding machine": "846024", | |
| "crop sprayer": "842482", "irrigation system": "842482", | |
| "pharmaceutical": "300692", "medicament": "300490", | |
| "sludge": "271099", "waste oil": "271099", | |
| "wooden crate": "441520", "wooden pallet": "441520", | |
| "server": "847150", "workstation": "847130", | |
| "blender": "850940", "mixer": "850940", | |
| "vaccine": "300241", "antiserum": "300241", | |
| "steering wheel": "870894", "seat belt": "870893", | |
| "rough diamond": "710231", "diamond rough": "710231", | |
| "granite slab": "680223", "granite tile": "680223", | |
| "crushed gravel": "251710", "gravel": "251710", | |
| "bituminous coal": "270112", "coal": "270112", | |
| "lpg": "271119", "propane": "271119", "butane": "271119", | |
| "biofuel": "150120", "biodiesel": "382600", | |
| "fiber optic cable": "854470", "optical cable": "854470", | |
| "satellite dish": "852910", "antenna": "852910", | |
| "acrylic paint": "320890", "latex paint": "320890", | |
| "watercolor": "321310", "oil paint": "320890", | |
| "shower head": "650691", "shower": "650691", | |
| "power socket": "853669", "electrical outlet": "853669", | |
| "power outlet": "853669", "wall socket": "853669", | |
| "ethylene": "271114", "polyethylene resin": "390110", | |
| "benzene": "290220", "toluene": "290230", | |
| "formaldehyde": "291211", "methanol formaldehyde": "291211", | |
| "ammonia": "281410", "ammonium": "310520", | |
| "viscose rayon": "540331", "rayon": "540331", | |
| "kevlar": "550111", "aramid fiber": "550111", | |
| "cork": "450110", "cork board": "450490", | |
| "rattan": "140120", "wicker": "940159", | |
| "seagrass": "940389", | |
| "automotive parts": "851771", "car parts": "870899", | |
| "auto parts": "870899", "vehicle parts": "870899", | |
| "heat exchanger": "841950", "radiator": "841950", | |
| "alarm system": "853110", "security alarm": "853110", | |
| "burglar alarm": "853110", "fire alarm": "853110", | |
| "vault safe": "830300", "safe": "830300", | |
| "helicopter parts": "880730", "helicopter": "880240", | |
| "rocket": "880790", "satellite": "880260", | |
| "anchor": "731600", "chain anchor": "731600", | |
| "diamond ring": "711319", "engagement ring": "711319", | |
| "mechanical watch": "910129", "pocket watch": "910119", | |
| "lawn mower": "843311", "mower": "843311", | |
| "garden hose": "400932", "hose pipe": "400932", | |
| "dog food": "230910", "cat food": "230910", | |
| "pet food": "230910", "animal feed": "230990", | |
| # ── IT & Electronics ── | |
| "laptop": "847130", "computer": "847130", "notebook computer": "847130", | |
| "desktop computer": "847130", "server computer": "847150", | |
| "tablet": "847130", "ipad": "847130", | |
| "phone": "851713", "smartphone": "851712", "mobile phone": "851712", | |
| "cell phone": "851712", "telephone": "851718", | |
| "television": "852872", "tv": "852872", "monitor": "852852", | |
| "computer monitor": "852852", "display": "852852", | |
| "camera": "900653", "digital camera": "900659", "video camera": "852589", | |
| "webcam": "852589", "cctv camera": "852589", | |
| "headphones": "851830", "earbuds": "851830", "headset": "851830", | |
| "speaker": "851822", "microphone": "851810", | |
| "printer": "844332", "scanner": "844331", "copier": "844331", | |
| "router": "851762", "modem": "851762", "switch": "851762", | |
| "network switch": "851762", "network hub": "851762", | |
| "usb flash drive": "852351", "memory card": "852351", "ssd": "852351", | |
| "hard drive": "852351", "hdd": "852351", "solid state drive": "852351", | |
| "storage device": "852351", | |
| "battery": "850650", "charger": "850440", "power adapter": "850440", | |
| "power supply": "850440", "power inverter": "850440", | |
| "led light": "940542", "solar panel": "854140", | |
| "circuit board": "853400", "pcb": "853400", "printed circuit board": "853400", | |
| "motherboard": "847150", "ram": "854231", "memory module": "854231", | |
| "ram memory module": "854231", "dimm": "854231", | |
| "graphics card": "847180", "gpu": "847180", "video card": "847180", | |
| "semiconductor": "854110", "chip": "854110", "microchip": "854110", | |
| "processor": "854231", "cpu": "854231", | |
| "transformer": "850421", "electric motor": "850110", | |
| "led display": "852872", "lcd": "852852", "oled": "852872", | |
| "smartwatch": "910212", "wearable": "910212", | |
| "drone": "880260", "uav": "880260", | |
| "speaker system": "851822", "sound bar": "851822", | |
| "keyboard": "847160", "mouse": "847160", "trackpad": "847160", | |
| "external hard drive": "852351", "usb hub": "851762", | |
| "graphics processor": "847180", "co-processor": "847180", | |
| # ── Vehicles ── | |
| "car": "870323", "automobile": "870323", "vehicle": "870323", | |
| "truck": "870422", "motorcycle": "871120", "bicycle": "871200", | |
| "bus": "870240", "aircraft": "880240", "airplane": "880240", | |
| "ship": "890120", "boat": "890310", "train": "860110", | |
| "trailer": "871640", "semi-trailer": "871631", | |
| "suv": "870323", "van": "870323", "pickup truck": "870422", | |
| "scooter": "871120", "moped": "871120", | |
| "locomotive": "860110", "railway car": "860210", | |
| # ── Apparel & Textiles ── | |
| "shirt": "620520", "t-shirt": "610910", "blouse": "620630", | |
| "polo shirt": "610510", "dress shirt": "620520", | |
| "pants": "620342", "trousers": "620342", "jeans": "620342", | |
| "shorts": "620342", "chinos": "620342", | |
| "dress": "620442", "skirt": "620452", "jumpsuit": "621149", | |
| "jacket": "620193", "coat": "620193", "blazer": "620331", | |
| "raincoat": "620213", "windbreaker": "620193", | |
| "sweater": "611030", "pullover": "611030", "cardigan": "611030", | |
| "hoodie": "611020", "sweatshirt": "611020", "wool sweater": "611020", | |
| "socks": "611595", "underwear": "610821", "boxers": "610711", | |
| "gloves": "621600", "hat": "650500", "cap": "650699", | |
| "scarf": "611710", "tie": "621520", "belt": "420330", | |
| "leather belt": "420330", | |
| "shoe": "640399", "shoes": "640399", "sneakers": "640411", | |
| "boot": "640399", "boots": "640399", "sandal": "640299", | |
| "slipper": "640419", "flip-flop": "640299", | |
| "cotton t-shirt": "610910", "cotton tshirt": "610910", | |
| "cotton shirt": "620520", "cotton dress": "620442", | |
| "silk dress": "620442", "linen shirt": "620560", | |
| "sports uniform": "611241", "jersey": "611241", "tracksuit": "611241", | |
| "baby clothes": "611120", "infant clothing": "611120", | |
| "swimsuit": "611231", "bikini": "611231", "swimming trunks": "611231", | |
| "nightwear": "610721", "pajamas": "610721", "nightgown": "610721", | |
| "uniform": "611241", "workwear": "621133", | |
| # ── Food & Agriculture ── | |
| "apple": "080810", "orange": "080510", "banana": "080390", | |
| "grape": "080610", "strawberry": "081010", "blueberry": "081040", | |
| "mango": "081060", "pineapple": "080430", "peach": "080930", | |
| "cherry": "080930", "pear": "080830", "kiwi": "081050", | |
| "lemon": "080520", "lime": "080520", "grapefruit": "080540", | |
| "watermelon": "080710", "melon": "080710", | |
| "coffee": "090121", "tea": "090210", "cocoa": "180100", | |
| "rice": "100630", "wheat": "100119", "corn": "100590", | |
| "barley": "100390", "oat": "100410", "rye": "100210", | |
| "bread": "190590", "pasta": "190219", "noodles": "190219", | |
| "cereal": "190410", "flour": "110100", "starch": "110819", | |
| "cheese": "040690", "milk": "040110", "butter": "040510", | |
| "yogurt": "040310", "cream": "040390", | |
| "egg": "040711", "chicken": "020714", "beef": "020130", | |
| "pork": "020319", "lamb": "020430", "turkey": "020727", | |
| "fish": "030289", "salmon": "030214", "tuna": "030487", | |
| "shrimp": "030617", "lobster": "030632", "crab": "030614", | |
| "oyster": "030629", "squid": "030752", "octopus": "030759", | |
| "wine": "220421", "beer": "220300", "whisky": "220830", | |
| "vodka": "220860", "rum": "220840", "gin": "220850", | |
| "chocolate": "180632", "candy": "170490", "sugar": "170199", | |
| "salt": "250100", "pepper": "090411", "cinnamon": "090611", | |
| "turmeric": "091030", "ginger": "091011", | |
| "olive oil": "150910", "sunflower oil": "151219", | |
| "palm oil": "151190", "coconut oil": "151319", | |
| "honey": "040900", "maple syrup": "170220", | |
| "tomato": "070200", "potato": "070190", "onion": "070310", | |
| "garlic": "070320", "carrot": "070610", "cabbage": "070490", | |
| "lettuce": "070511", "cucumber": "070700", "pepper": "070960", | |
| "mushroom": "070951", "spinach": "070960", "broccoli": "070690", | |
| "corn kernels": "071040", "peas": "070810", "beans": "071333", | |
| "lentils": "071340", "soybeans": "120190", | |
| "peanut": "120241", "almond": "080211", "walnut": "080231", | |
| "cashew": "080390", "hazelnut": "080221", "pecan": "080232", | |
| "coconut": "080119", "pistachio": "080212", | |
| "avocado": "080440", "olive": "070990", | |
| # ── Metals & Minerals ── | |
| "steel": "720899", "iron": "720899", "aluminum": "760310", | |
| "aluminium": "760310", "copper": "740311", "brass": "740321", | |
| "zinc": "790111", "tin": "800110", "lead": "780110", | |
| "nickel": "750210", "titanium": "810820", "tungsten": "810110", | |
| "platinum": "711011", "palladium": "711021", | |
| "gold": "710812", "silver": "710691", "copper wire": "740811", | |
| "steel pipe": "730630", "steel tube": "730630", | |
| "aluminum sheet": "760612", "aluminum foil": "760711", | |
| "steel plate": "720851", "steel sheet": "720851", | |
| "stainless steel": "721990", "alloy steel": "721990", | |
| "iron ore": "260112", "bauxite": "260600", | |
| # ── Wood & Paper ── | |
| "wood": "440799", "timber": "440799", "lumber": "440799", | |
| "plywood": "441231", "veneer": "440890", | |
| "mdf": "441112", "particle board": "441011", | |
| "furniture": "940390", "furniture board": "940390", | |
| "paper": "481910", "cardboard": "481910", "paperboard": "481910", | |
| "tissue paper": "481810", "toilet paper": "481810", | |
| "paper towel": "481890", "napkin": "481890", | |
| "book": "490199", "newspaper": "490290", "magazine": "490210", | |
| "notebook": "482010", "diary": "482010", | |
| "envelope": "481710", "file folder": "482010", "ring binder": "482010", | |
| "paper bag": "481930", "shopping bag": "481930", | |
| "gift wrap": "481940", "wrapping paper": "481940", | |
| "label": "482110", "sticker": "482110", "stamp": "490700", | |
| "poster": "491191", "calendar": "491000", | |
| # ── Plastics & Rubber ── | |
| "plastic": "392690", "rubber": "401590", | |
| "plastic bottle": "392330", "plastic bag": "392321", | |
| "plastic container": "392330", "plastic cup": "392410", | |
| "plastic plate": "392410", "plastic toy": "950300", | |
| "rubber tire": "401110", "rubber tube": "400942", | |
| "silicone": "391000", "teflon": "390461", | |
| "foam": "391723", "styrofoam": "391723", | |
| "nylon": "540771", "polyester": "540761", | |
| "acrylic": "392051", "polycarbonate": "392061", | |
| "polyethylene": "390110", "polypropylene": "390210", | |
| "pvc": "390410", "vinyl": "390410", | |
| # ── Furniture ── | |
| "chair": "940161", "sofa": "940161", "couch": "940161", | |
| "table": "940360", "desk": "940360", "coffee table": "940360", | |
| "bed": "940350", "mattress": "940421", "pillow": "940490", | |
| "cabinet": "940330", "shelf": "940390", "bookshelf": "940390", | |
| "wardrobe": "940350", "dresser": "940350", "nightstand": "940350", | |
| "bookcase": "940390", "cupboard": "940330", | |
| "kitchen cabinet": "940330", "office desk": "940360", | |
| "office chair": "940161", "filing cabinet": "940330", | |
| # ── Beauty & Personal Care ── | |
| "shampoo": "330510", "conditioner": "330510", | |
| "soap": "340111", "detergent": "340220", | |
| "toothpaste": "330610", "toothbrush": "960321", | |
| "perfume": "330300", "cologne": "330300", | |
| "cosmetics": "330499", "makeup": "330499", | |
| "lipstick": "330410", "mascara": "330420", | |
| "foundation": "330499", "powder": "330499", | |
| "lotion": "330499", "moisturizer": "330499", | |
| "sunscreen": "330499", "spf": "330499", | |
| "deodorant": "330790", "antiperspirant": "330790", | |
| "razor": "821210", "shaving cream": "330790", | |
| "nail polish": "330300", "nail remover": "330790", | |
| "hair dye": "330510", "hair color": "330510", | |
| "facial cleanser": "330499", "face wash": "330499", | |
| "eye cream": "330499", "serum": "330499", | |
| # ── Tools & Hardware ── | |
| "drill": "846721", "saw": "846722", "hammer": "820520", | |
| "screwdriver": "820540", "wrench": "820411", | |
| "pliers": "820320", "pliers set": "820320", | |
| "tape measure": "901780", "level": "901720", | |
| "sandpaper": "680520", "grinder": "846781", | |
| "welder": "851511", "soldering iron": "851511", | |
| "paint brush": "960340", "roller": "960340", | |
| "spray gun": "842420", "air compressor": "841480", | |
| "generator": "850211", "inverter": "850440", | |
| "battery charger": "850440", "power bank": "850760", | |
| "extension cord": "854442", "power strip": "853690", | |
| "electrical panel": "853710", "circuit breaker": "853630", | |
| # ── Medical & Health ── | |
| "medicine": "300490", "drug": "300490", "pill": "300490", | |
| "tablet": "300490", "capsule": "300490", | |
| "syringe": "901890", "bandage": "300590", | |
| "thermometer": "902580", "stethoscope": "901819", | |
| "mask": "630790", "surgical mask": "630790", | |
| "glove": "401519", "surgical glove": "401519", | |
| "goggles": "900490", "safety glasses": "900490", | |
| "wheelchair": "871310", "crutch": "902110", | |
| "hearing aid": "902110", "pacemaker": "902150", | |
| "x-ray": "902212", "ultrasound": "901819", | |
| "mri": "902219", "ct scanner": "902219", | |
| "defibrillator": "902180", "ventilator": "901920", | |
| "bandage": "300590", "gauze": "300590", | |
| "antiseptic": "300650", "disinfectant": "300650", | |
| "vitamin": "210690", "supplement": "210690", | |
| "probiotic": "210690", "protein powder": "210690", | |
| # ── Sports & Recreation ── | |
| "ball": "950662", "football": "950662", "soccer": "950662", | |
| "basketball": "950662", "baseball": "950662", "volleyball": "950662", | |
| "golf club": "950639", "tennis racket": "950659", | |
| "badminton racket": "950659", "table tennis": "950640", | |
| "ski": "950611", "snowboard": "950699", "ski boots": "950612", | |
| "skates": "950670", "ice skates": "950670", "roller skates": "950670", | |
| "yoga mat": "950691", "exercise mat": "950691", | |
| "dumbbell": "950691", "barbell": "950691", "weight": "950691", | |
| "treadmill": "950691", "bicycle": "871200", | |
| "camping tent": "630622", "sleeping bag": "630790", | |
| "backpack": "420292", "hiking boots": "640399", | |
| "fishing rod": "950710", "fishing net": "950790", | |
| "hunting rifle": "930520", "shotgun": "930520", | |
| # ── Music & Entertainment ── | |
| "guitar": "920210", "piano": "920120", "violin": "920210", | |
| "drum": "920600", "flute": "920510", "saxophone": "920510", | |
| "trumpet": "920510", "ukulele": "920210", | |
| "microphone": "851810", "amplifier": "851840", | |
| "audio mixer": "851840", "turntable": "851981", | |
| "record player": "851981", "cd player": "851981", | |
| "dvd": "852349", "blu-ray": "852349", "cd": "852349", | |
| "video game": "950450", "game console": "950450", | |
| "board game": "950490", "card game": "950490", | |
| "toy": "950300", "doll": "950391", "puzzle": "950300", | |
| "teddy bear": "950300", "stuffed animal": "950300", | |
| "building blocks": "950300", "lego": "950300", | |
| "rc car": "950300", "remote control toy": "950300", | |
| # ── Construction Materials ── | |
| "cement": "252329", "concrete": "252329", "brick": "690490", | |
| "tile": "690721", "ceramic tile": "690721", "porcelain tile": "690721", | |
| "glass": "700529", "mirror": "700910", "window": "700800", | |
| "door": "441820", "wooden door": "441820", | |
| "insulation": "680610", "foam insulation": "391723", | |
| "drywall": "480210", "gypsum board": "480210", | |
| "plywood": "441231", "osb": "441011", | |
| "concrete block": "690100", "paver": "690410", | |
| # ── Household Items ── | |
| "box": "481910", "carton": "481910", | |
| "bag": "481930", "bottle": "701090", "jar": "701090", | |
| "cup": "691200", "mug": "691200", "plate": "691110", | |
| "bowl": "691200", "spoon": "821599", "fork": "821599", | |
| "knife": "821191", "kitchen knife": "821191", "scissors": "821300", | |
| "pot": "732393", "pan": "732393", "wok": "732393", | |
| "frying pan": "732393", "saucepan": "732393", | |
| "kettle": "851679", "teapot": "732393", | |
| "trash can": "732399", "recycling bin": "732399", | |
| "umbrella": "660110", "walking stick": "660210", | |
| "broom": "960310", "mop": "960310", | |
| "vacuum cleaner": "850811", "washing machine": "845011", | |
| "dishwasher": "842211", "refrigerator": "841810", | |
| "freezer": "841821", "oven": "851410", "stove": "851610", | |
| "microwave": "851650", "toaster": "851672", | |
| "iron": "851640", "hair dryer": "851631", | |
| "air conditioner": "841510", "fan": "841451", | |
| "heater": "851629", "lamp": "940520", | |
| "bulb": "853950", "led bulb": "853950", | |
| "curtain": "630391", "blinds": "630392", | |
| "carpet": "570330", "rug": "570330", | |
| "doormat": "570330", "mat": "570330", | |
| # ── Chemicals ── | |
| "paint": "320890", "varnish": "320910", "lacquer": "320910", | |
| "ink": "321519", "dye": "320411", "pigment": "320610", | |
| "glue": "350610", "adhesive": "350610", "epoxy": "390730", | |
| "resin": "390710", "silicone": "391000", | |
| "fertilizer": "310590", "pesticide": "380891", | |
| "herbicide": "380891", "insecticide": "380891", | |
| "cleaning product": "340220", "bleach": "281111", | |
| "acetone": "291411", "methanol": "290511", | |
| "ethanol": "220710", "isopropanol": "290512", | |
| "sulfuric acid": "280700", "hydrochloric acid": "280610", | |
| # ── Energy & Power ── | |
| "solar panel": "854140", "solar cell": "854140", | |
| "wind turbine": "850240", "generator": "850211", | |
| "ups": "850440", "uninterruptible power supply": "850440", | |
| "fuel cell": "850790", "lithium battery": "850760", | |
| "alkaline battery": "850650", "rechargeable battery": "850760", | |
| # ── Office Supplies ── | |
| "pen": "960810", "pencil": "960910", "crayon": "960990", | |
| "marker": "960910", "highlighter": "960910", | |
| "eraser": "960900", "sharpener": "961400", | |
| "ruler": "901780", "protractor": "901720", | |
| "stapler": "847290", "paper clip": "830510", | |
| "rubber band": "402700", "tape": "391990", | |
| "correction fluid": "382490", "whiteout": "382490", | |
| "calculator": "847010", "cash register": "847050", | |
| # ── Industrial Equipment ── | |
| "valve": "848180", "pump": "841370", "compressor": "841430", | |
| "filter": "842131", "bearing": "848210", "gear": "848340", | |
| "spring": "732020", "chain": "731582", | |
| "pipe": "730300", "fitting": "730719", "flange": "730791", | |
| "coupling": "848360", "belt": "401011", | |
| "conveyor": "842820", "crane": "842611", "forklift": "842710", | |
| "hoist": "842511", "winch": "842519", | |
| "lathe": "845811", "milling machine": "845710", | |
| "drill press": "845921", "grinder": "846781", | |
| "cnc machine": "845811", "3d printer": "847740", | |
| "robot": "847950", "industrial robot": "847950", | |
| "concrete mixer": "847431", "excavator": "843049", | |
| "bulldozer": "843049", "loader": "842952", | |
| # ── Fasteners & Small Hardware ── | |
| "nail": "731700", "screw": "731815", "bolt": "731815", | |
| "nut": "731816", "washer": "731822", "rivet": "731823", | |
| "clip": "731824", "pin": "731829", "cotter pin": "731829", | |
| "hinge": "830210", "lock": "830110", "padlock": "830110", | |
| "handle": "830240", "knob": "830240", | |
| "hook": "830810", "eyelet": "830820", | |
| # ── Textiles & Fabrics ── | |
| "cotton": "520899", "polyester": "540761", "nylon": "540771", | |
| "silk": "500720", "wool": "511130", "linen": "530921", | |
| "denim": "520833", "canvas": "520852", | |
| "velvet": "580121", "satin": "500720", | |
| "lace": "580410", "embroidery": "581010", | |
| "fabric": "520899", "textile": "520899", | |
| "thread": "520411", "yarn": "520411", | |
| "rope": "560750", "cord": "560750", | |
| "towel": "630260", "blanket": "630140", | |
| "sheet": "630231", "pillowcase": "630231", | |
| "tablecloth": "630260", "napkin": "630260", | |
| "curtain": "630391", "drape": "630391", | |
| # ── Leather Goods ── | |
| "leather": "420221", "suede": "420330", | |
| "wallet": "420231", "purse": "420221", | |
| "handbag": "420221", "briefcase": "420212", | |
| "backpack": "420292", "suitcase": "420212", | |
| "luggage": "420212", "travel bag": "420212", | |
| "glove": "420330", "jacket": "420330", | |
| "shoe upper": "640610", | |
| # ── Glass & Ceramics ── | |
| "glass": "700529", "mirror": "700910", | |
| "window": "700800", "door": "700800", | |
| "glass bottle": "701090", "glass jar": "701090", | |
| "glassware": "701339", "drinking glass": "701339", | |
| "ceramic": "690721", "porcelain": "690721", | |
| "vase": "691310", "flowerpot": "691390", | |
| "tile": "690721", "brick": "690490", | |
| # ── Sports Equipment (specific) ── | |
| "tennis ball": "950662", "golf ball": "950662", | |
| "baseball bat": "950639", "cricket bat": "950639", | |
| "hockey stick": "950639", "lacrosse stick": "950639", | |
| "surfboard": "950699", "paddleboard": "950699", | |
| "kayak": "890310", "canoe": "890310", | |
| "snorkel": "950619", "scuba gear": "950619", | |
| "ski pole": "950619", "ski boot": "950612", | |
| "snowshoe": "950619", | |
| } | |
| # Multi-word keyword matching order (longest first for greedy match) | |
| _KEYWORD_MATCH_ORDER: Final[tuple[str, ...]] = tuple( | |
| sorted(KEYWORD_HS_MAPPING.keys(), key=len, reverse=True) | |
| ) | |
| # ── Data Models ─────────────────────────────────────────────────────── | |
| class HSClassification: | |
| """Represents an HS code classification result. | |
| Attributes: | |
| hs_code: The classified code (6-digit WCO, 10-digit HTSUS, or 12-digit TARIC). | |
| description: Human-readable description of the HS code. | |
| confidence: Classification confidence from 0.0 to 1.0. | |
| source: Classification source (pyhscodes, llm-openai, etc.). | |
| section: HS section letter (I-XXI). | |
| chapter: 2-digit chapter code. | |
| heading: 4-digit heading code. | |
| subheading: 6-digit subheading code. | |
| parent_code: Parent HS code in the hierarchy. | |
| level: Code digit length (2, 4, 6, 10, or 12). | |
| reasoning: Explanation of classification rationale. | |
| alternatives: Alternative classifications considered. | |
| needs_human_review: Whether human review is required. | |
| """ | |
| hs_code: str | |
| description: str | |
| confidence: float | |
| source: str | |
| section: str = "" | |
| chapter: str = "" | |
| heading: str = "" | |
| subheading: str = "" | |
| parent_code: str = "" | |
| level: int = 0 | |
| reasoning: str = "" | |
| alternatives: list[dict[str, Any]] = field(default_factory=list) | |
| needs_human_review: bool = False | |
| def __post_init__(self) -> None: | |
| """Parse code structure into hierarchical components.""" | |
| self.level = len(self.hs_code) | |
| if self.level >= 2: | |
| self.chapter = self.hs_code[:2] | |
| if self.level >= 4: | |
| self.heading = self.hs_code[:4] | |
| if self.level >= 6: | |
| self.subheading = self.hs_code[:6] | |
| def is_valid(self) -> bool: | |
| """Whether the code has a valid structural length.""" | |
| return self.level in VALID_CODE_LEVELS | |
| def needs_review(self) -> bool: | |
| """Whether human review is needed (low confidence or flagged).""" | |
| return self.needs_human_review or self.confidence < MEDIUM_CONFIDENCE_THRESHOLD | |
| class ClassificationRequest: | |
| """Request for HS code classification. | |
| Attributes: | |
| description: Product description to classify. | |
| country_origin: Country of origin (ISO 3166-1 alpha-2). | |
| country_destination: Destination country (ISO 3166-1 alpha-2). | |
| quantity: Quantity of goods. | |
| unit: Unit of measurement. | |
| value: Monetary value. | |
| material: Primary material composition. | |
| additional_info: Additional context for classification. | |
| """ | |
| description: str | |
| country_origin: str = "" | |
| country_destination: str = "" | |
| quantity: float = 0.0 | |
| unit: str = "" | |
| value: float = 0.0 | |
| material: str = "" | |
| additional_info: str = "" | |
| class ClassificationResponse: | |
| """Response from HS code classification. | |
| Attributes: | |
| request: The original classification request. | |
| primary: The primary (best) classification result. | |
| alternatives: Alternative classifications in descending confidence. | |
| processing_time_ms: Total processing time in milliseconds. | |
| model_used: Which model/source was used. | |
| """ | |
| request: ClassificationRequest | |
| primary: HSClassification | |
| alternatives: list[HSClassification] = field(default_factory=list) | |
| processing_time_ms: float = 0.0 | |
| model_used: str = "" | |
| class ClassificationCacheEntry: | |
| """In-memory cache entry for a classification result. | |
| Attributes: | |
| description_hash: SHA-256 hash of the normalized description. | |
| description: Original product description. | |
| response: The cached ClassificationResponse. | |
| created_at: Timestamp when the cache entry was created. | |
| hit_count: Number of times this cache entry was accessed. | |
| """ | |
| description_hash: str | |
| description: str | |
| response: ClassificationResponse | |
| created_at: float = field(default_factory=_time.time) | |
| hit_count: int = 0 | |
| # ── Semantic Cache ──────────────────────────────────────────────────── | |
| class ClassificationCache: | |
| """Local semantic cache for HS classifications. | |
| Provides two-tier caching: | |
| 1. In-memory LRU dict for hot paths (O(1) lookup) | |
| 2. PostgreSQL ILIKE fallback for cross-restart persistence | |
| Thread-safe via threading.Lock. | |
| """ | |
| _MAX_MEMORY_ENTRIES: Final[int] = 2048 | |
| _CACHE_TTL_SECONDS: Final[float] = 86400.0 # 24 hours | |
| def __init__(self) -> None: | |
| self._lock: threading.Lock = threading.Lock() | |
| self._memory_cache: dict[str, ClassificationCacheEntry] = {} | |
| def _normalize(description: str) -> str: | |
| """Normalize description for deterministic hashing.""" | |
| normalized: str = description.lower().strip() | |
| normalized = re.sub(r"\s+", " ", normalized) | |
| return normalized | |
| def _hash(description: str) -> str: | |
| """SHA-256 hash of a normalized description.""" | |
| return hashlib.sha256(description.encode("utf-8")).hexdigest() | |
| def get(self, description: str) -> Optional[ClassificationResponse]: | |
| """Look up a description in the in-memory cache.""" | |
| normalized: str = self._normalize(description) | |
| desc_hash: str = self._hash(normalized) | |
| with self._lock: | |
| entry = self._memory_cache.get(desc_hash) | |
| if entry is None: | |
| return None | |
| if (_time.time() - entry.created_at) > self._CACHE_TTL_SECONDS: | |
| del self._memory_cache[desc_hash] | |
| return None | |
| entry.hit_count += 1 | |
| return entry.response | |
| def put(self, description: str, response: ClassificationResponse) -> None: | |
| """Store a classification result in the in-memory cache.""" | |
| normalized: str = self._normalize(description) | |
| desc_hash: str = self._hash(normalized) | |
| with self._lock: | |
| if len(self._memory_cache) >= self._MAX_MEMORY_ENTRIES: | |
| if self._memory_cache: | |
| lru_key: str = min( | |
| self._memory_cache, | |
| key=lambda k: ( | |
| self._memory_cache[k].hit_count, | |
| self._memory_cache[k].created_at, | |
| ), | |
| ) | |
| del self._memory_cache[lru_key] | |
| self._memory_cache[desc_hash] = ClassificationCacheEntry( | |
| description_hash=desc_hash, | |
| description=normalized, | |
| response=response, | |
| ) | |
| async def get_from_db(self, description: str) -> Optional[ClassificationResponse]: | |
| """Look up a description in the PostgreSQL compliance_review_items table. | |
| Uses ILIKE for fuzzy matching against previously classified products. | |
| """ | |
| try: | |
| from hermes.database.connection import get_database_manager | |
| db_manager = get_database_manager() | |
| if not db_manager or not db_manager.async_engine: | |
| return None | |
| from sqlalchemy import text | |
| normalized: str = self._normalize(description) | |
| pattern: str = f"%{normalized}%" | |
| async with db_manager.async_engine.connect() as conn: | |
| result = await conn.execute( | |
| text( | |
| "SELECT hs_code_suggested, hs_code_description, " | |
| "hs_code_confidence, hs_code_alternatives " | |
| "FROM compliance_review_items " | |
| "WHERE hs_code_suggested != '' " | |
| "AND (" | |
| " LOWER(shipper) LIKE LOWER(:pattern) " | |
| " OR LOWER(consignee) LIKE LOWER(:pattern) " | |
| " OR LOWER(invoice_number) LIKE LOWER(:pattern) " | |
| ") " | |
| "ORDER BY created_at DESC LIMIT 1" | |
| ), | |
| {"pattern": pattern}, | |
| ) | |
| row = result.fetchone() | |
| if row is None: | |
| return None | |
| hs_code: str = row[0] or "" | |
| hs_desc: str = row[1] or "" | |
| hs_conf: float = float(row[2] or 0.0) | |
| hs_alts_raw: str = row[3] or "[]" | |
| if not hs_code: | |
| return None | |
| try: | |
| alts_data: list[dict[str, Any]] = json.loads(hs_alts_raw) | |
| except (json.JSONDecodeError, TypeError): | |
| alts_data = [] | |
| alternatives: list[HSClassification] = [] | |
| for alt in alts_data: | |
| alternatives.append( | |
| HSClassification( | |
| hs_code=alt.get("hs_code", ""), | |
| description=alt.get("description", ""), | |
| confidence=float(alt.get("confidence", 0.0)), | |
| source="cache-db", | |
| ) | |
| ) | |
| primary = HSClassification( | |
| hs_code=hs_code, | |
| description=hs_desc, | |
| confidence=hs_conf, | |
| source="cache-db", | |
| reasoning="Loaded from compliance review cache", | |
| ) | |
| logger.info( | |
| "Cache HIT (DB) → HS %s (%.0f%%)", | |
| hs_code, | |
| hs_conf * 100, | |
| ) | |
| return ClassificationResponse( | |
| request=ClassificationRequest(description=description), | |
| primary=primary, | |
| alternatives=alternatives, | |
| model_used="cache-db", | |
| ) | |
| except Exception as exc: | |
| logger.debug("DB cache lookup failed: %s", exc) | |
| return None | |
| def stats(self) -> dict[str, Any]: | |
| """Return cache statistics.""" | |
| with self._lock: | |
| total_hits: int = sum(e.hit_count for e in self._memory_cache.values()) | |
| return { | |
| "memory_entries": len(self._memory_cache), | |
| "max_entries": self._MAX_MEMORY_ENTRIES, | |
| "total_hits": total_hits, | |
| "ttl_seconds": self._CACHE_TTL_SECONDS, | |
| } | |
| # ── HS Code Validator ──────────────────────────────────────────────── | |
| class HSCodeValidator: | |
| """Validates HS code structure and consistency. | |
| Provides static methods for structural validation of 2/4/6/10/12-digit | |
| HS codes according to WCO nomenclature rules. | |
| """ | |
| def is_valid_hs6(code: str) -> bool: | |
| """Validate 6-digit HS code structure. | |
| Args: | |
| code: The code to validate. | |
| Returns: | |
| True if code is a valid 6-digit HS code (chapter 01-99, etc.). | |
| """ | |
| if not HS6_PATTERN.match(code): | |
| return False | |
| chapter: int = int(code[:2]) | |
| if chapter < 1 or chapter > 99: | |
| return False | |
| heading: int = int(code[2:4]) | |
| if heading < 1 or heading > 99: | |
| return False | |
| subheading: int = int(code[4:6]) | |
| if subheading < 1 or subheading > 99: | |
| return False | |
| return True | |
| def is_valid_htsus(code: str) -> bool: | |
| """Validate 10-digit HTSUS code structure. | |
| Args: | |
| code: The code to validate. | |
| Returns: | |
| True if code is a valid 10-digit HTSUS code. | |
| """ | |
| if not HTSUS_PATTERN.match(code): | |
| return False | |
| return HSCodeValidator.is_valid_hs6(code[:6]) | |
| def is_valid_taric(code: str) -> bool: | |
| """Validate 12-digit TARIC code structure. | |
| Args: | |
| code: The code to validate. | |
| Returns: | |
| True if code is a valid 12-digit TARIC code. | |
| """ | |
| if not TARIC_PATTERN.match(code): | |
| return False | |
| return HSCodeValidator.is_valid_hs6(code[:6]) | |
| def get_hierarchy(code: str) -> dict[str, str]: | |
| """Get hierarchical breakdown of HS code. | |
| Args: | |
| code: The HS code to decompose. | |
| Returns: | |
| Dict with keys: section, chapter, heading, subheading, national. | |
| """ | |
| hierarchy: dict[str, str] = { | |
| "section": "", | |
| "chapter": "", | |
| "heading": "", | |
| "subheading": "", | |
| "national": "", | |
| } | |
| if len(code) >= 2: | |
| hierarchy["chapter"] = code[:2] | |
| if len(code) >= 4: | |
| hierarchy["heading"] = code[:4] | |
| if len(code) >= 6: | |
| hierarchy["subheading"] = code[:6] | |
| if len(code) >= 10: | |
| hierarchy["national"] = code[6:10] | |
| if len(code) >= 12: | |
| hierarchy["national"] = code[6:12] | |
| return hierarchy | |
| # ── pyhscodes Wrapper ──────────────────────────────────────────────── | |
| class PyHSCodesClassifier: | |
| """Wrapper around pyhscodes for local HS code classification. | |
| Provides search, lookup, and hierarchy traversal using the pyhscodes | |
| database of 6,940+ WCO HS codes. | |
| Raises: | |
| ImportError: If pyhscodes is not installed. | |
| """ | |
| def __init__(self) -> None: | |
| """Initialize the classifier. | |
| Raises: | |
| ImportError: If pyhscodes is not installed. | |
| """ | |
| if not PYHSCODES_AVAILABLE: | |
| raise ImportError("pyhscodes not installed") | |
| self._hs: Any = _pyhscodes | |
| def search( | |
| self, query: str, max_results: int = 5 | |
| ) -> list[HSClassification]: | |
| """Search for HS codes matching the query. | |
| Args: | |
| query: Product description to search for. | |
| max_results: Maximum number of results to return. | |
| Returns: | |
| List of HSClassification objects sorted by relevance. | |
| """ | |
| try: | |
| all_results: list[Any] = self._hs.search_fuzzy(query) | |
| results: list[Any] = all_results[:max_results] if all_results else [] | |
| except LookupError: | |
| results = self._search_by_keywords(query, max_results) | |
| except Exception as exc: | |
| logger.error("pyhscodes search failed: %s", exc) | |
| results = [] | |
| classifications: list[HSClassification] = [] | |
| for result in results: | |
| confidence: float = self._calculate_confidence(query, result) | |
| classification = HSClassification( | |
| hs_code=result.hscode, | |
| description=result.description, | |
| confidence=confidence, | |
| source="pyhscodes", | |
| section=result.section, | |
| parent_code=result.parent, | |
| level=int(result.level) if result.level else len(result.hscode), | |
| ) | |
| classifications.append(classification) | |
| return classifications | |
| def _search_by_keywords(self, query: str, max_results: int) -> list[Any]: | |
| """Search by individual keywords when full query fails. | |
| Args: | |
| query: Original product query. | |
| max_results: Maximum results per keyword. | |
| Returns: | |
| List of raw pyhscodes results. | |
| """ | |
| words: list[str] = re.findall(r"[a-zA-Z]+", query.lower()) | |
| keywords: list[str] = [ | |
| w for w in words if w not in _LLM_STOP_WORDS and len(w) > 2 | |
| ] | |
| for keyword in keywords: | |
| try: | |
| results: list[Any] = self._hs.search_fuzzy(keyword) | |
| if results: | |
| return results[:max_results] | |
| except (LookupError, Exception): | |
| continue | |
| return [] | |
| def lookup(self, code: str) -> Optional[HSClassification]: | |
| """Lookup a specific HS code. | |
| Args: | |
| code: The HS code to lookup. | |
| Returns: | |
| HSClassification if found, None otherwise. | |
| """ | |
| try: | |
| result: Any = self._hs.lookup(code) | |
| if result is None: | |
| return None | |
| return HSClassification( | |
| hs_code=result.hscode, | |
| description=result.description, | |
| confidence=1.0, | |
| source="pyhscodes", | |
| section=result.section, | |
| parent_code=result.parent, | |
| level=int(result.level) if result.level else len(result.hscode), | |
| ) | |
| except Exception as exc: | |
| logger.error("pyhscodes lookup failed: %s", exc) | |
| return None | |
| def get_children(self, code: str) -> list[HSClassification]: | |
| """Get child codes of a parent code. | |
| Args: | |
| code: The parent HS code. | |
| Returns: | |
| List of child HSClassification objects. | |
| """ | |
| try: | |
| results: list[Any] = self._hs.get_children(code) | |
| classifications: list[HSClassification] = [] | |
| for result in results: | |
| classification = HSClassification( | |
| hs_code=result.hscode, | |
| description=result.description, | |
| confidence=0.9, | |
| source="pyhscodes", | |
| section=result.section, | |
| parent_code=result.parent, | |
| level=int(result.level) if result.level else len(result.hscode), | |
| ) | |
| classifications.append(classification) | |
| return classifications | |
| except Exception as exc: | |
| logger.error("pyhscodes get_children failed: %s", exc) | |
| return [] | |
| def _calculate_confidence(self, query: str, result: Any) -> float: | |
| """Calculate confidence score for a search result. | |
| Args: | |
| query: Original search query. | |
| result: Raw pyhscodes result object. | |
| Returns: | |
| Confidence score between 0.5 and 0.95. | |
| """ | |
| query_lower: str = query.lower().strip() | |
| desc_lower: str = result.description.lower() if result.description else "" | |
| commodity_lower: str = result.commodity.lower() if result.commodity else "" | |
| if query_lower in desc_lower: | |
| return 0.95 | |
| if query_lower in commodity_lower: | |
| return 0.95 | |
| query_words: set[str] = set(query_lower.split()) | |
| desc_words: set[str] = set(desc_lower.split()) | |
| commodity_words: set[str] = set(commodity_lower.split()) | |
| all_entity_words: set[str] = desc_words | commodity_words | |
| if not query_words or not all_entity_words: | |
| return 0.5 | |
| overlap: set[str] = query_words & all_entity_words | |
| overlap_ratio: float = len(overlap) / len(query_words) | |
| base_confidence: float = 0.6 + (overlap_ratio * 0.3) | |
| return min(base_confidence, 0.95) | |
| # ── Main Classification Engine ──────────────────────────────────────── | |
| class HSClassificationEngine: | |
| """Main HS code classification interface with hybrid approach. | |
| Combines pyhscodes database search with LLM refinement for optimal | |
| accuracy. pyhscodes handles 80% of cases locally; LLM is used only | |
| for edge cases requiring deeper reasoning. | |
| Attributes: | |
| _pyhscodes: The pyhscodes classifier instance, or None if unavailable. | |
| """ | |
| def __init__(self) -> None: | |
| """Initialize the engine with pyhscodes and semantic cache.""" | |
| self._pyhscodes: Optional[PyHSCodesClassifier] = None | |
| self._cache: ClassificationCache = ClassificationCache() | |
| self._semaphore: Optional[asyncio.Semaphore] = None | |
| try: | |
| self._pyhscodes = PyHSCodesClassifier() | |
| except ImportError: | |
| logger.warning("pyhscodes not available, LLM-only mode") | |
| def _get_semaphore(self) -> asyncio.Semaphore: | |
| """Get or create the async semaphore (lazy, thread-safe).""" | |
| if self._semaphore is None: | |
| self._semaphore = asyncio.Semaphore(5) | |
| return self._semaphore | |
| def classify( | |
| self, | |
| description: str, | |
| country_origin: str = "", | |
| country_destination: str = "", | |
| use_llm: bool = False, | |
| target_digits: TargetDigits = 6, | |
| ) -> ClassificationResponse: | |
| """Classify a product description to HS code. | |
| Args: | |
| description: Product description to classify. | |
| country_origin: Country of origin (ISO 3166-1 alpha-2). | |
| country_destination: Destination country (ISO 3166-1 alpha-2). | |
| use_llm: Whether to use LLM for refinement. | |
| target_digits: Target code length (6, 10, or 12). | |
| Returns: | |
| ClassificationResponse with primary code and alternatives. | |
| """ | |
| start_time: float = _time.time() | |
| # ── Cache check (in-memory first, zero-latency) ──────────────── | |
| cached: Optional[ClassificationResponse] = self._cache.get(description) | |
| if cached is not None: | |
| logger.info( | |
| "Cache HIT (memory) for '%s' → HS %s (%.0f%%, %.1fms)", | |
| description[:60], | |
| cached.primary.hs_code, | |
| cached.primary.confidence * 100, | |
| (_time.time() - start_time) * 1000, | |
| ) | |
| cached.processing_time_ms = (_time.time() - start_time) * 1000 | |
| return cached | |
| request = ClassificationRequest( | |
| description=description, | |
| country_origin=country_origin, | |
| country_destination=country_destination, | |
| ) | |
| # Step 0: Keyword pre-filter (fastest path for common products) | |
| keyword_match: Optional[HSClassification] = self._keyword_pre_filter(description) | |
| if keyword_match and keyword_match.confidence >= 0.90: | |
| logger.info( | |
| "Keyword pre-filter match for '%s' → HS %s (95%%)", | |
| description[:60], | |
| keyword_match.hs_code, | |
| ) | |
| processing_time = (_time.time() - start_time) * 1000 | |
| response = ClassificationResponse( | |
| request=request, | |
| primary=keyword_match, | |
| alternatives=[], | |
| processing_time_ms=processing_time, | |
| model_used="keyword-match", | |
| ) | |
| self._cache.put(description, response) | |
| return response | |
| # Step 1: pyhscodes classification (always try first) | |
| pyhscodes_results: list[HSClassification] = [] | |
| if self._pyhscodes: | |
| pyhscodes_results = self._pyhscodes.search(description, max_results=5) | |
| # Step 2: If LLM requested and pyhscodes didn't give high confidence | |
| llm_result: Optional[HSClassification] = None | |
| if use_llm and ( | |
| not pyhscodes_results | |
| or pyhscodes_results[0].confidence < HIGH_CONFIDENCE_THRESHOLD | |
| ): | |
| llm_result = self._classify_with_llm( | |
| description, country_origin, country_destination, target_digits | |
| ) | |
| # Step 3: Select best result with cross-validation | |
| primary: Optional[HSClassification] = None | |
| alternatives: list[HSClassification] = [] | |
| if llm_result and ( | |
| not pyhscodes_results | |
| or llm_result.confidence > pyhscodes_results[0].confidence | |
| ): | |
| if pyhscodes_results: | |
| llm_chapter: str = ( | |
| llm_result.hs_code[:2] if len(llm_result.hs_code) >= 2 else "" | |
| ) | |
| pyh_chapter: str = ( | |
| pyhscodes_results[0].hs_code[:2] | |
| if len(pyhscodes_results[0].hs_code) >= 2 | |
| else "" | |
| ) | |
| if llm_chapter and pyh_chapter and llm_chapter != pyh_chapter: | |
| # LLM wins if its confidence is significantly higher | |
| confidence_gap: float = ( | |
| llm_result.confidence - pyhscodes_results[0].confidence | |
| ) | |
| if confidence_gap > 0.15: | |
| # LLM is much more confident — trust it | |
| llm_result.needs_human_review = True | |
| llm_result.reasoning += ( | |
| f" [CROSS-VALIDATION: LLM chapter {llm_chapter} differs from" | |
| f" pyhscodes chapter {pyh_chapter} — LLM preferred due to higher" | |
| f" confidence ({llm_result.confidence:.0%} vs" | |
| f" {pyhscodes_results[0].confidence:.0%})]" | |
| ) | |
| primary = llm_result | |
| alternatives = pyhscodes_results[:3] | |
| else: | |
| # Close confidence — prefer pyhscodes, flag for review | |
| llm_result.needs_human_review = True | |
| llm_result.reasoning += ( | |
| f" [CROSS-VALIDATION WARNING: LLM chapter {llm_chapter} differs from" | |
| f" pyhscodes chapter {pyh_chapter} — human review required]" | |
| ) | |
| primary = pyhscodes_results[0] | |
| alternatives = [llm_result] + pyhscodes_results[1:3] | |
| else: | |
| primary = llm_result | |
| alternatives = pyhscodes_results[:3] | |
| else: | |
| primary = llm_result | |
| elif pyhscodes_results: | |
| primary = pyhscodes_results[0] | |
| alternatives = pyhscodes_results[1:3] | |
| if llm_result: | |
| alternatives.append(llm_result) | |
| else: | |
| primary = HSClassification( | |
| hs_code="000000", | |
| description="Unable to classify", | |
| confidence=0.0, | |
| source="none", | |
| needs_human_review=True, | |
| reasoning="No matching HS codes found", | |
| ) | |
| # Step 4: Validate and flag for review | |
| if primary: | |
| primary = self._post_process_validation(primary) | |
| primary.needs_human_review = self._needs_human_review( | |
| primary, description | |
| ) | |
| # Additional check: if confidence below NEEDS_REVIEW_THRESHOLD, flag it | |
| if primary.confidence < NEEDS_REVIEW_THRESHOLD: | |
| primary.needs_human_review = True | |
| if primary.needs_human_review: | |
| primary.reasoning += " [Flagged for human review]" | |
| processing_time: float = (_time.time() - start_time) * 1000 | |
| response = ClassificationResponse( | |
| request=request, | |
| primary=primary, | |
| alternatives=alternatives, | |
| processing_time_ms=processing_time, | |
| model_used="pyhscodes" if not llm_result else "pyhscodes+llm", | |
| ) | |
| # ── Cache store ──────────────────────────────────────────────── | |
| try: | |
| self._cache.put(description, response) | |
| except Exception as cache_exc: | |
| logger.debug("Cache store failed: %s", cache_exc) | |
| return response | |
| async def classify_batch( | |
| self, | |
| items: list[dict[str, str]], | |
| use_llm: bool = False, | |
| target_digits: TargetDigits = 6, | |
| ) -> list[ClassificationResponse]: | |
| """Classify a batch of products concurrently with semaphore control. | |
| Uses asyncio.gather with a Semaphore(5) to limit concurrent LLM | |
| API calls. Checks in-memory cache first, then DB cache, then | |
| fires the LLM. | |
| Args: | |
| items: List of dicts with keys: description, country_origin (opt), | |
| country_destination (opt). | |
| use_llm: Whether to use LLM for refinement. | |
| target_digits: Target code length (6, 10, or 12). | |
| Returns: | |
| List of ClassificationResponse objects, one per input item. | |
| """ | |
| if not items: | |
| return [] | |
| sem: asyncio.Semaphore = self._get_semaphore() | |
| async def _classify_one(item: dict[str, str]) -> ClassificationResponse: | |
| desc: str = item.get("description", "") | |
| origin: str = item.get("country_origin", "") | |
| dest: str = item.get("country_destination", "") | |
| # Tier 1: In-memory cache (zero latency) | |
| cached: Optional[ClassificationResponse] = self._cache.get(desc) | |
| if cached is not None: | |
| return cached | |
| # Tier 2: DB cache (async, ~5ms) | |
| db_cached: Optional[ClassificationResponse] = await self._cache.get_from_db(desc) | |
| if db_cached is not None: | |
| self._cache.put(desc, db_cached) | |
| return db_cached | |
| # Tier 3: Live classification with semaphore-controlled concurrency | |
| async with sem: | |
| # Run the synchronous classify in a thread pool | |
| loop = asyncio.get_running_loop() | |
| result: ClassificationResponse = await loop.run_in_executor( | |
| None, | |
| lambda: self.classify( | |
| description=desc, | |
| country_origin=origin, | |
| country_destination=dest, | |
| use_llm=use_llm, | |
| target_digits=target_digits, | |
| ), | |
| ) | |
| return result | |
| tasks = [_classify_one(item) for item in items] | |
| results: list[ClassificationResponse] = await asyncio.gather( | |
| *tasks, return_exceptions=False | |
| ) | |
| # Cross-validate batch results | |
| results = self._cross_validate_batch(results) | |
| logger.info( | |
| "Batch classified %d items (%d LLM, %d cached)", | |
| len(results), | |
| sum(1 for r in results if r.model_used in ("pyhscodes+llm",)), | |
| sum(1 for r in results if r.model_used.startswith("cache")), | |
| ) | |
| return list(results) | |
| def _classify_with_llm( | |
| self, | |
| description: str, | |
| country_origin: str, | |
| country_destination: str, | |
| target_digits: int, | |
| ) -> Optional[HSClassification]: | |
| """Use LLM for HS code classification. | |
| Supports OpenAI, Anthropic, Google Gemini, and OpenCode-compatible | |
| providers (DeepSeek, etc.) via ModelConfig. | |
| Args: | |
| description: Product description. | |
| country_origin: Country of origin. | |
| country_destination: Destination country. | |
| target_digits: Target code length. | |
| Returns: | |
| HSClassification from LLM, or None if unavailable/failed. | |
| """ | |
| import os | |
| try: | |
| from hermes.config.settings import get_settings | |
| settings = get_settings() | |
| prompt: str = self._build_classification_prompt( | |
| description, country_origin, country_destination, target_digits | |
| ) | |
| # Determine provider: explicit config > env vars > fallback | |
| provider = settings.model.provider.lower() | |
| api_key = settings.model.api_key | |
| base_url = settings.model.base_url | |
| model_name = settings.model.name | |
| # Also check env vars as fallback | |
| openai_key: Optional[str] = os.environ.get("OPENAI_API_KEY") | |
| anthropic_key: Optional[str] = os.environ.get("ANTHROPIC_API_KEY") | |
| google_key: Optional[str] = os.environ.get("GOOGLE_API_KEY") or os.environ.get("GEMINI_API_KEY") | |
| # Priority: opencode/deepseek > google/gemini > openai > anthropic | |
| if provider in ("opencode", "deepseek") and api_key: | |
| return self._call_opencode(prompt, target_digits, api_key, base_url, model_name) | |
| elif provider in ("google", "gemini") and api_key: | |
| return self._call_gemini(prompt, target_digits, api_key) | |
| elif google_key: | |
| return self._call_gemini(prompt, target_digits, google_key) | |
| elif openai_key: | |
| return self._call_openai(prompt, target_digits) | |
| elif anthropic_key: | |
| return self._call_anthropic(prompt, target_digits) | |
| else: | |
| logger.warning("No LLM API keys available for classification") | |
| return None | |
| except Exception as exc: | |
| logger.error("LLM classification failed: %s", exc) | |
| return None | |
| def _build_classification_prompt( | |
| self, | |
| description: str, | |
| country_origin: str, | |
| country_destination: str, | |
| target_digits: int, | |
| ) -> str: | |
| """Build prompt for LLM classification with chapter context and examples. | |
| Args: | |
| description: Product description. | |
| country_origin: Country of origin. | |
| country_destination: Destination country. | |
| target_digits: Target code length. | |
| Returns: | |
| Formatted prompt string. | |
| """ | |
| # Build chapter context section (condensed for token efficiency) | |
| chapter_guide_lines: list[str] = [] | |
| for ch, desc in sorted(CHAPTER_DESCRIPTIONS.items()): | |
| chapter_guide_lines.append(f" {ch}: {desc}") | |
| chapter_guide = "\n".join(chapter_guide_lines) | |
| prompt: str = ( | |
| "You are a world-class trade compliance officer and HS code classification expert.\n\n" | |
| "## 4-Pillar Classification Framework\n" | |
| "Analyze the product using these 4 pillars:\n" | |
| "1. MATERIAL COMPOSITION — What is the product made of? (e.g., cotton, steel, plastic, wood)\n" | |
| "2. FUNCTION/USE — What does the product do? What is it used for?\n" | |
| "3. ESSENTIAL CHARACTER — What is the most important feature or component?\n" | |
| "4. MANUFACTURING PROCESS — How was it made? (e.g., knitted, woven, cast, machined)\n\n" | |
| "## HS Code Structure\n" | |
| "HS codes follow a hierarchical structure:\n" | |
| "- 2-digit Chapter (e.g., 09 = Coffee, tea, spices)\n" | |
| "- 4-digit Heading (e.g., 0902 = Tea, whether or not flavored)\n" | |
| "- 6-digit Subheading (e.g., 090210 = Green tea, in containers ≤3kg)\n" | |
| f"- {target_digits}-digit National code (HTSUS/TARIC extension)\n\n" | |
| "## HS Chapter Reference\n" | |
| f"{chapter_guide}\n\n" | |
| "## Classification Examples (with 4-Pillar Analysis)\n" | |
| "Example 1: \"Arabica coffee beans, roasted, 500g bags\"\n" | |
| "- Material: Coffee beans (agricultural product)\n" | |
| "- Function: Beverage ingredient\n" | |
| "- Essential: Roasted coffee\n" | |
| "- Process: Roasted\n" | |
| '→ hs_code: "090121" (Coffee, roasted, not decaffeinated)\n\n' | |
| "Example 2: \"Laptop computer, Intel i7 processor, 15.6 inch screen\"\n" | |
| "- Material: Electronic components, plastic/metal casing\n" | |
| "- Function: Data processing, computing\n" | |
| "- Essential: Portable digital computer\n" | |
| "- Process: Assembled electronic device\n" | |
| '→ hs_code: "847130" (Portable digital automatic data processing machines)\n\n' | |
| "Example 3: \"Cotton t-shirt, men's, 100% cotton\"\n" | |
| "- Material: 100% cotton\n" | |
| "- Function: Clothing, upper body wear\n" | |
| "- Essential: Knitted T-shirt\n" | |
| "- Process: Knitted/crocheted\n" | |
| '→ hs_code: "610910" (T-shirts, singlets, knitted/crocheted, of cotton)\n\n' | |
| "Example 4: \"Stainless steel screws, M6 x 20mm, box of 100\"\n" | |
| "- Material: Stainless steel\n" | |
| "- Function: Fastening\n" | |
| "- Essential: Screw (threaded fastener)\n" | |
| "- Process: Machined/forged\n" | |
| '→ hs_code: "731815" (Screws, of stainless steel)\n\n' | |
| "Example 5: \"Extra virgin olive oil, 500ml glass bottle\"\n" | |
| "- Material: Olive oil (vegetable oil)\n" | |
| "- Function: Cooking/food ingredient\n" | |
| "- Essential: Virgin olive oil\n" | |
| "- Process: First cold pressing\n" | |
| '→ hs_code: "150910" (Olive oil, virgin, in containers ≤18kg)\n\n' | |
| "Example 6: \"Silk women's dress, evening wear, embroidered\"\n" | |
| "- Material: Silk\n" | |
| "- Function: Women's clothing, formal wear\n" | |
| "- Essential: Silk dress (woven, not knitted)\n" | |
| "- Process: Woven + embroidered\n" | |
| '→ hs_code: "620449" (Women\'s dresses of silk, woven)\n\n' | |
| "Example 7: \"CNC milling machine, 3-axis, 10kW motor\"\n" | |
| "- Material: Steel/iron construction\n" | |
| "- Function: Metal cutting/milling\n" | |
| "- Essential: Machine tool for removing material\n" | |
| "- Process: CNC controlled, machined\n" | |
| '→ hs_code: "845710" (Machining centres for working metal)\n\n' | |
| "Example 8: \"Leather travel bag, brown, with shoulder strap\"\n" | |
| "- Material: Leather\n" | |
| "- Function: Carrying personal items\n" | |
| "- Essential: Travel goods/suitcase\n" | |
| "- Process: Sewn leather construction\n" | |
| '→ hs_code: "420212" (Trunks, suitcases with outer surface of plastics or textile)\n\n' | |
| "Example 9: \"Wooden dining table, oak, extendable\"\n" | |
| "- Material: Oak wood\n" | |
| "- Function: Dining surface, furniture\n" | |
| "- Essential: Table (wooden furniture)\n" | |
| "- Process: Joinery, finished wood\n" | |
| '→ hs_code: "940360" (Other wooden furniture)\n\n' | |
| "Example 10: \"Wool carpet, hand-tufted, 2m x 3m\"\n" | |
| "- Material: Wool\n" | |
| "- Function: Floor covering\n" | |
| "- Essential: Tufted carpet\n" | |
| "- Process: Hand-tufted\n" | |
| '→ hs_code: "570110" (Carpets of wool/fine animal hair, knotted)\n\n' | |
| ) | |
| if country_origin: | |
| prompt += f"Country of Origin: {country_origin}\n" | |
| if country_destination: | |
| prompt += f"Destination Country: {country_destination}\n" | |
| prompt += ( | |
| f"\n## Task\n" | |
| f"Classify the following product to a {target_digits}-digit HS code:\n\n" | |
| f"Product Description: {description}\n\n" | |
| "## Response Format\n" | |
| "Return ONLY valid JSON (no markdown, no explanation outside JSON):\n" | |
| "{\n" | |
| f' "hs_code": "{target_digits}-digit numeric code",\n' | |
| ' "description": "Brief description of what this HS code covers",\n' | |
| ' "confidence": 0.0 to 1.0 (your confidence in this classification),\n' | |
| ' "reasoning": "Why this code was chosen - key factors",\n' | |
| ' "chapter": "2-digit chapter code",\n' | |
| ' "heading": "4-digit heading code",\n' | |
| ' "alternatives": [\n' | |
| ' {"hs_code": "code", "description": "desc", "confidence": 0.0, "reason": "why"}\n' | |
| " ]\n" | |
| "}\n\n" | |
| "## Rules\n" | |
| "1. Use WCO HS nomenclature - classify by material composition and essential character\n" | |
| "2. Primary classification should be 6-digit international HS code\n" | |
| f"3. Provide {target_digits}-digit code (pad with 0s if only 6-digit is certain)\n" | |
| "4. ALWAYS provide top 3 alternative codes with confidence scores\n" | |
| "5. If unsure about the exact code, provide the most likely chapter and heading\n" | |
| "6. Consider: What is the product made of? What is it used for?\n" | |
| "7. Consider: What is the essential character or principal function?\n" | |
| "8. Be precise - avoid broad catch-all codes unless truly applicable\n" | |
| ) | |
| return prompt | |
| def _call_openai( | |
| self, prompt: str, target_digits: int | |
| ) -> Optional[HSClassification]: | |
| """Call OpenAI API for classification. | |
| Args: | |
| prompt: The classification prompt. | |
| target_digits: Target code length. | |
| Returns: | |
| HSClassification from OpenAI, or None on failure. | |
| """ | |
| try: | |
| import openai | |
| client = openai.OpenAI() | |
| response = client.chat.completions.create( | |
| model="gpt-4", | |
| messages=[{"role": "user", "content": prompt}], | |
| temperature=0.1, | |
| max_tokens=500, | |
| ) | |
| content: str = response.choices[0].message.content or "" | |
| return self._parse_llm_response(content, "openai", target_digits) | |
| except Exception as exc: | |
| raise LLMClassificationError("openai", str(exc)) from exc | |
| def _call_anthropic( | |
| self, prompt: str, target_digits: int | |
| ) -> Optional[HSClassification]: | |
| """Call Anthropic API for classification. | |
| Args: | |
| prompt: The classification prompt. | |
| target_digits: Target code length. | |
| Returns: | |
| HSClassification from Anthropic, or None on failure. | |
| """ | |
| try: | |
| import anthropic | |
| client = anthropic.Anthropic() | |
| response = client.messages.create( | |
| model="claude-3-sonnet-20240229", | |
| max_tokens=500, | |
| messages=[{"role": "user", "content": prompt}], | |
| ) | |
| content: str = response.content[0].text | |
| return self._parse_llm_response(content, "anthropic", target_digits) | |
| except Exception as exc: | |
| raise LLMClassificationError("anthropic", str(exc)) from exc | |
| def _call_opencode( | |
| self, | |
| prompt: str, | |
| target_digits: int, | |
| api_key: str, | |
| base_url: str, | |
| model_name: str, | |
| ) -> Optional[HSClassification]: | |
| """Call OpenCode-compatible API (DeepSeek, Qwen, etc.) for classification. | |
| Uses the OpenAI SDK with a custom base_url for OpenAI-compatible | |
| endpoints. Reasoning models (deepseek-v4-flash) use reasoning_content | |
| for thinking tokens; non-reasoning models (qwen3.7-plus) return | |
| directly in content. | |
| Args: | |
| prompt: The classification prompt. | |
| target_digits: Target code length. | |
| api_key: API key for the provider. | |
| base_url: Base URL for the API. | |
| model_name: Model name to use. | |
| Returns: | |
| HSClassification from the provider, or None on failure. | |
| """ | |
| try: | |
| import openai | |
| client = openai.OpenAI(api_key=api_key, base_url=base_url) | |
| # Reasoning models need more tokens for thinking | |
| is_reasoning: bool = "deepseek" in model_name.lower() | |
| max_tokens: int = 8192 if is_reasoning else 1024 | |
| response = client.chat.completions.create( | |
| model=model_name, | |
| messages=[{"role": "user", "content": prompt}], | |
| temperature=0.1, | |
| max_tokens=max_tokens, | |
| ) | |
| choice = response.choices[0] | |
| content: str = choice.message.content or "" | |
| # Reasoning models may put the answer in reasoning_content | |
| # when content is empty (truncated by max_tokens) | |
| if not content.strip(): | |
| reasoning = getattr(choice.message, "reasoning_content", None) | |
| if reasoning: | |
| content = reasoning | |
| logger.info( | |
| "LLM content empty, extracting from reasoning_content (%d chars)", | |
| len(reasoning), | |
| ) | |
| return self._parse_llm_response(content, "opencode", target_digits) | |
| except Exception as exc: | |
| raise LLMClassificationError("opencode", str(exc)) from exc | |
| def _call_gemini( | |
| self, prompt: str, target_digits: int, api_key: str | |
| ) -> Optional[HSClassification]: | |
| """Call Google Gemini API for classification. | |
| Uses the google-genai SDK (successor to google-generativeai). | |
| Args: | |
| prompt: The classification prompt. | |
| target_digits: Target code length. | |
| api_key: Google API key. | |
| Returns: | |
| HSClassification from Gemini, or None on failure. | |
| """ | |
| try: | |
| from google import genai | |
| client = genai.Client(api_key=api_key) | |
| response = client.models.generate_content( | |
| model="gemini-2.0-flash", | |
| contents=prompt, | |
| config=genai.types.GenerateContentConfig( | |
| temperature=0.1, | |
| max_output_tokens=500, | |
| ), | |
| ) | |
| content: str = response.text | |
| return self._parse_llm_response(content, "gemini", target_digits) | |
| except Exception as exc: | |
| raise LLMClassificationError("gemini", str(exc)) from exc | |
| def _parse_llm_response( | |
| self, content: str, source: str, target_digits: int | |
| ) -> Optional[HSClassification]: | |
| """Parse LLM response into HSClassification. | |
| LLM self-reported confidence is capped at LLM_MAX_CONFIDENCE (0.85) | |
| because LLMs are calibrationally poor and may output inflated confidence | |
| for hallucinated codes. | |
| Args: | |
| content: Raw LLM response text. | |
| source: LLM provider name (openai, anthropic). | |
| target_digits: Expected code length. | |
| Returns: | |
| HSClassification if parsing succeeds, None otherwise. | |
| """ | |
| try: | |
| import json | |
| # Strip markdown code blocks (```json ... ```) if present | |
| cleaned: str = content.strip() | |
| if cleaned.startswith("```"): | |
| # Remove opening fence (```json or ```) | |
| first_newline: int = cleaned.find("\n") | |
| if first_newline != -1: | |
| cleaned = cleaned[first_newline + 1 :] | |
| # Remove closing fence | |
| if cleaned.rstrip().endswith("```"): | |
| cleaned = cleaned.rstrip()[: -len("```")].rstrip() | |
| # Extract outermost JSON block by matching brace depth | |
| 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]) | |
| hs_code: str = data.get("hs_code", "") | |
| # Normalize: strip dots, dashes, spaces | |
| hs_code = re.sub(r"[.\-\s]", "", hs_code) | |
| if len(hs_code) != target_digits: | |
| logger.warning( | |
| "LLM returned %d-digit code, expected %d", | |
| len(hs_code), | |
| target_digits, | |
| ) | |
| if len(hs_code) > target_digits: | |
| hs_code = hs_code[:target_digits] | |
| else: | |
| hs_code = hs_code.ljust(target_digits, "0") | |
| 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, | |
| ) | |
| # Check if code exists in pyhscodes database | |
| in_database: bool = False | |
| if PYHSCODES_AVAILABLE: | |
| try: | |
| db_result = _pyhscodes.lookup(hs_code) | |
| if db_result and db_result.hscode: | |
| in_database = True | |
| except Exception: | |
| pass | |
| if not in_database: | |
| try: | |
| db_result = _pyhscodes.lookup(hs_code[:6]) | |
| if db_result and db_result.hscode: | |
| in_database = True | |
| except Exception: | |
| pass | |
| # Compute final confidence: LLM confidence × database_match × format_validity | |
| format_valid: bool = bool(HS6_PATTERN.match(hs_code[:6])) | |
| db_match_factor: float = 1.0 if in_database else 0.85 | |
| format_factor: float = 1.0 if format_valid else 0.8 | |
| final_confidence: float = min( | |
| capped_confidence * db_match_factor * format_factor, | |
| LLM_MAX_CONFIDENCE, | |
| ) | |
| alternatives: list[HSClassification] = [] | |
| for alt in data.get("alternatives", []): | |
| alt_code: str = alt.get("hs_code", "") | |
| alt_code = re.sub(r"[.\-\s]", "", alt_code) | |
| if len(alt_code) > target_digits: | |
| alt_code = alt_code[:target_digits] | |
| alt_conf: float = min(alt.get("confidence", 0.5), LLM_MAX_CONFIDENCE) | |
| alternatives.append( | |
| HSClassification( | |
| hs_code=alt_code, | |
| description=alt.get("description", ""), | |
| confidence=alt_conf, | |
| source=f"llm-{source}", | |
| ) | |
| ) | |
| return HSClassification( | |
| hs_code=hs_code, | |
| description=data.get("description", ""), | |
| confidence=final_confidence, | |
| source=f"llm-{source}", | |
| reasoning=data.get("reasoning", ""), | |
| alternatives=alternatives, | |
| ) | |
| except Exception as exc: | |
| logger.error("Failed to parse LLM response: %s", exc) | |
| return None | |
| def _needs_human_review( | |
| self, classification: HSClassification, query: str | |
| ) -> bool: | |
| """Determine if classification needs human review. | |
| Args: | |
| classification: The classification to evaluate. | |
| query: The original product query. | |
| Returns: | |
| True if human review is recommended. | |
| """ | |
| if classification.confidence < MEDIUM_CONFIDENCE_THRESHOLD: | |
| return True | |
| if not classification.is_valid: | |
| return True | |
| if classification.source.startswith("llm-"): | |
| return True | |
| query_lower: str = query.lower() | |
| if any(term in query_lower for term in _AMBIGUOUS_TERMS): | |
| return True | |
| return False | |
| def _keyword_pre_filter(description: str) -> Optional[HSClassification]: | |
| """Look up product in the keyword→HS code mapping for instant classification. | |
| Uses longest-match-first to handle multi-word terms (e.g., "olive oil" | |
| matches before "oil"). Uses word boundary matching to avoid false positives. | |
| Args: | |
| description: Product description to match. | |
| Returns: | |
| HSClassification with high confidence if exact keyword match found, | |
| None otherwise. | |
| """ | |
| desc_lower: str = description.lower().strip() | |
| for keyword in _KEYWORD_MATCH_ORDER: | |
| # Use word boundary regex to avoid substring matches | |
| # e.g., "ring" should not match "Bamboo flooring planks" | |
| pattern = re.compile(r'\b' + re.escape(keyword) + r'\b') | |
| if pattern.search(desc_lower): | |
| code: str = KEYWORD_HS_MAPPING[keyword] | |
| # Look up the full code description from pyhscodes if available | |
| desc_from_db: str = "" | |
| if PYHSCODES_AVAILABLE: | |
| try: | |
| result = _pyhscodes.lookup(code) | |
| if result: | |
| desc_from_db = result.description or "" | |
| except Exception: | |
| pass | |
| if not desc_from_db: | |
| # Fallback: use chapter description | |
| chapter: str = code[:2] | |
| desc_from_db = CHAPTER_DESCRIPTIONS.get(chapter, "HS code") | |
| return HSClassification( | |
| hs_code=code, | |
| description=desc_from_db, | |
| confidence=0.95, | |
| source="keyword-match", | |
| reasoning=f"Exact keyword match: '{keyword}' → {code}", | |
| ) | |
| return None | |
| def _post_process_validation( | |
| classification: HSClassification, | |
| ) -> HSClassification: | |
| """Validate and adjust a classification result. | |
| Checks format, database existence, and applies confidence adjustments. | |
| Args: | |
| classification: The classification to validate. | |
| Returns: | |
| Validated (possibly adjusted) classification. | |
| """ | |
| if classification.source == "none": | |
| return classification | |
| # Format validation | |
| code: str = classification.hs_code | |
| format_valid: bool = bool(HS6_PATTERN.match(code[:6])) if len(code) >= 6 else False | |
| # Database existence check | |
| in_database: bool = False | |
| if PYHSCODES_AVAILABLE: | |
| try: | |
| result = _pyhscodes.lookup(code) | |
| if result and result.hscode: | |
| in_database = True | |
| except Exception: | |
| pass | |
| # Also try 6-digit prefix | |
| if not in_database and len(code) >= 6: | |
| try: | |
| result = _pyhscodes.lookup(code[:6]) | |
| if result and result.hscode: | |
| in_database = True | |
| except Exception: | |
| pass | |
| # Confidence adjustment based on validation | |
| adjusted_conf: float = classification.confidence | |
| if not format_valid: | |
| adjusted_conf *= 0.8 # Penalize invalid format | |
| if not in_database and classification.source.startswith("llm-"): | |
| adjusted_conf *= 0.85 # Penalize LLM codes not in database | |
| classification.confidence = min(adjusted_conf, LLM_MAX_CONFIDENCE) | |
| # Flag if code not in database | |
| if not in_database and classification.source == "pyhscodes": | |
| classification.needs_human_review = True | |
| classification.reasoning += " [Code not found in pyhscodes database]" | |
| return classification | |
| def _find_closest_database_code( | |
| description: str, max_candidates: int = 5 | |
| ) -> list[HSClassification]: | |
| """Find closest matching codes in pyhscodes database using fuzzy matching. | |
| Args: | |
| description: Product description. | |
| max_candidates: Maximum number of candidates to return. | |
| Returns: | |
| List of HSClassification candidates sorted by relevance. | |
| """ | |
| if not PYHSCODES_AVAILABLE: | |
| return [] | |
| candidates: list[HSClassification] = [] | |
| # Split description into keywords and search each | |
| words: list[str] = re.findall(r"[a-zA-Z]{3,}", description.lower()) | |
| search_terms: list[str] = [w for w in words if w not in _LLM_STOP_WORDS][:5] | |
| for term in search_terms: | |
| try: | |
| results = _pyhscodes.search_fuzzy(term) | |
| if results: | |
| for r in results[:2]: | |
| # Calculate relevance using SequenceMatcher | |
| ratio: float = SequenceMatcher( | |
| None, description.lower(), r.description.lower() | |
| ).ratio() | |
| candidates.append(HSClassification( | |
| hs_code=r.hscode, | |
| description=r.description, | |
| confidence=min(0.6 + ratio * 0.3, 0.9), | |
| source="pyhscodes-fuzzy", | |
| section=r.section if hasattr(r, "section") else "", | |
| )) | |
| except Exception: | |
| continue | |
| # Deduplicate by code and sort by confidence | |
| seen: set[str] = set() | |
| unique: list[HSClassification] = [] | |
| for c in sorted(candidates, key=lambda x: x.confidence, reverse=True): | |
| if c.hs_code not in seen: | |
| seen.add(c.hs_code) | |
| unique.append(c) | |
| return unique[:max_candidates] | |
| def _cross_validate_batch( | |
| self, results: list[ClassificationResponse] | |
| ) -> list[ClassificationResponse]: | |
| """Cross-validate batch results: flag similar descriptions with different codes. | |
| Args: | |
| results: List of classification responses. | |
| Returns: | |
| Same list with needs_human_review flags updated. | |
| """ | |
| if len(results) < 2: | |
| return results | |
| for i, r1 in enumerate(results): | |
| for j, r2 in enumerate(results): | |
| if i >= j: | |
| continue | |
| # Compare descriptions using SequenceMatcher | |
| d1: str = r1.request.description.lower() | |
| d2: str = r2.request.description.lower() | |
| similarity: float = SequenceMatcher(None, d1, d2).ratio() | |
| # If descriptions are very similar but codes differ, flag both | |
| if similarity > 0.75: | |
| code1: str = r1.primary.hs_code[:4] # Compare at heading level | |
| code2: str = r2.primary.hs_code[:4] | |
| if code1 != code2: | |
| r1.primary.needs_human_review = True | |
| r2.primary.needs_human_review = True | |
| r1.primary.reasoning += ( | |
| f" [BATCH: Similar to '{r2.request.description[:40]}' " | |
| f"but different classification]" | |
| ) | |
| r2.primary.reasoning += ( | |
| f" [BATCH: Similar to '{r1.request.description[:40]}' " | |
| f"but different classification]" | |
| ) | |
| return results | |
| # ── Convenience Functions ───────────────────────────────────────────── | |
| _engine: Optional[HSClassificationEngine] = None | |
| _engine_lock: threading.Lock = threading.Lock() | |
| def get_engine() -> HSClassificationEngine: | |
| """Get or create the global HS classification engine (thread-safe). | |
| Returns: | |
| The singleton HSClassificationEngine instance. | |
| """ | |
| global _engine | |
| if _engine is None: | |
| with _engine_lock: | |
| if _engine is None: | |
| _engine = HSClassificationEngine() | |
| return _engine | |
| def classify_product( | |
| description: str, | |
| country_origin: str = "", | |
| country_destination: str = "", | |
| use_llm: bool = False, | |
| target_digits: TargetDigits = 6, | |
| ) -> ClassificationResponse: | |
| """Classify a product to HS code. | |
| Args: | |
| description: Product description. | |
| country_origin: Country of origin. | |
| country_destination: Destination country. | |
| use_llm: Whether to use LLM refinement. | |
| target_digits: Target code length. | |
| Returns: | |
| ClassificationResponse with primary code and alternatives. | |
| """ | |
| engine: HSClassificationEngine = get_engine() | |
| return engine.classify( | |
| description, country_origin, country_destination, use_llm, target_digits | |
| ) | |
| def lookup_code(code: str) -> Optional[HSClassification]: | |
| """Lookup a specific HS code. | |
| Args: | |
| code: The HS code to lookup. | |
| Returns: | |
| HSClassification if found, None otherwise. | |
| """ | |
| if not PYHSCODES_AVAILABLE: | |
| return None | |
| classifier: PyHSCodesClassifier = PyHSCodesClassifier() | |
| return classifier.lookup(code) | |