""" 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 ─────────────────────────────────────────────────────── @dataclass(frozen=False, slots=True) 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] @property def is_valid(self) -> bool: """Whether the code has a valid structural length.""" return self.level in VALID_CODE_LEVELS @property 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 @dataclass(frozen=False, slots=True) 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 = "" @dataclass(frozen=False, slots=True) 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 = "" @dataclass(frozen=False, slots=True) 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] = {} @staticmethod def _normalize(description: str) -> str: """Normalize description for deterministic hashing.""" normalized: str = description.lower().strip() normalized = re.sub(r"\s+", " ", normalized) return normalized @staticmethod 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. """ @staticmethod 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 @staticmethod 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]) @staticmethod 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]) @staticmethod 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 @staticmethod 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 @staticmethod 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 @staticmethod 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)