from __future__ import annotations import logging import re from decimal import Decimal from typing import Any from .classifier_rules import ( EQUIPMENT_PURCHASE_UNITS, EQUIPMENT_RENTAL_UNITS, MACHINE_KEYWORDS, MATERIAL_GROUP_KEYWORDS, ) logger = logging.getLogger("dscons.invoice_truth.classifier") class TruthClassifierMixin: """Invoice line item classification for inventory vs equipment vs expenses.""" def classify_item( self, item_name: str, unit: str, quantity: Decimal | float, unit_price: Decimal | float, total_amount: Decimal | float, seller_name: str = "", ) -> dict[str, Any]: """Classify an invoice line item into: - OWNED_EQUIPMENT: Physical machine asset purchase (added to `erp_equipment`) - RENTAL_EQUIPMENT: Machine shift rental / subcontractor service (logged to `erp_equipment_logs`) - WAREHOUSE_MATERIAL: Physical construction material (added to `erp_warehouse_inventory`) - SUBCONTRACT_SERVICE: Construction work service / labor """ name_lower = (item_name or "").lower().strip() unit_lower = (unit or "").lower().strip() seller_lower = (seller_name or "").lower().strip() # Check if item has machine keywords detected_machine_type = None for m_type, kws in MACHINE_KEYWORDS.items(): if any(kw in name_lower for kw in kws): detected_machine_type = m_type break # Check if rental indication is present is_rental = ( unit_lower in EQUIPMENT_RENTAL_UNITS or "thuê" in name_lower or "ca máy" in name_lower or "vận hành" in name_lower or "tiền ca" in name_lower or "công máy" in name_lower or "ca" == unit_lower ) if detected_machine_type: if not is_rental and ( unit_lower in EQUIPMENT_PURCHASE_UNITS or quantity <= 5 ): # 1. Physical Machine Asset Purchase brand_model = self._extract_brand_model(item_name) return { "classification": "OWNED_EQUIPMENT", "equipment_type": detected_machine_type, "brand_model": brand_model, "unit": unit or "Chiếc", "quantity": quantity, "unit_price_vnd": unit_price, "total_amount_vnd": total_amount, "ownership_type": "owned", "description": f"Mua sắm {item_name} ({brand_model})", } else: # 2. Machine Rental / Shift return { "classification": "RENTAL_EQUIPMENT", "equipment_type": detected_machine_type, "unit": unit or "Ca", "quantity": quantity, "unit_price_vnd": unit_price, "total_amount_vnd": total_amount, "ownership_type": "rented", "description": f"Thuê ca máy: {item_name}", } # Check for materials detected_mat_group = "OTHER_SUPPLIES" for group, kws in MATERIAL_GROUP_KEYWORDS.items(): if any(kw in name_lower for kw in kws) or ( group == "FUEL_OIL" and "petrolimex" in seller_lower ): detected_mat_group = group break # Check if it's transport service if "vận chuyển" in name_lower or "cước" in name_lower: return { "classification": "TRANSPORT_SERVICE", "unit": unit or "Chuyến", "quantity": quantity, "unit_price_vnd": unit_price, "total_amount_vnd": total_amount, "description": f"Dịch vụ vận chuyển: {item_name}", } # Check if it's construction execution / service if ( "thi công" in name_lower or "xây dựng" in name_lower or "lắp đặt" in name_lower or "nhân công" in name_lower ): if unit_lower in ("gói", "lần", "hợp đồng", "m2", "m3", "md"): return { "classification": "SUBCONTRACT_SERVICE", "material_group": "CONSTRUCTION_SERVICE", "unit": unit or "Gói", "quantity": quantity, "unit_price_vnd": unit_price, "total_amount_vnd": total_amount, "description": f"Gói thầu / Dịch vụ thi công: {item_name}", } # 3. Physical Construction Material -> Warehouse Inventory return { "classification": "WAREHOUSE_MATERIAL", "material_group": detected_mat_group, "unit": unit or "Lô", "quantity": quantity, "unit_price_vnd": unit_price, "total_amount_vnd": total_amount, "description": f"Nhập kho vật tư: {item_name}", } def _extract_brand_model(self, text: str) -> str: """Extract brand/model from equipment text (e.g. 'Volvo EC210D', 'Komatsu PC200-8').""" brands = [ "Volvo", "Komatsu", "Hitachi", "Caterpillar", "CAT", "Sakai", "Hyundai", "Kobelco", "Doosan", "Sumitomo", "Dynapac", "Bomag", "Cummins", "Sinotruk", "Howo", "Hino", "Isuzu", "Tadano", "Kato", "Sany", "Zoomlion", "XCMG", "Putzmeister", "Schwing", ] text_clean = text.strip() for b in brands: m = re.search(rf"\b{b}\b[A-Za-z0-9\s\-]+", text_clean, re.IGNORECASE) if m: return m.group(0).strip() # Fallback to text itself if short # Fallback to text itself if short if len(text_clean) <= 50: return text_clean return text_clean[:50] def classify_invoice_business_type( self, direction: str, items_classification: list[str], has_rental_input_history: bool = False ) -> dict[str, str]: """ Classifies the entire invoice business category and subcategory based on its items and history. - direction: 'input' or 'output' - items_classification: list of item classifications (e.g., ['WAREHOUSE_MATERIAL', 'RENTAL_EQUIPMENT']) - has_rental_input_history: Used for output invoices to check if we previously rented this equipment in. """ if not items_classification: return {"business_category": "unclassified", "business_subcategory": None} # Count item types types_set = set(items_classification) has_material = "WAREHOUSE_MATERIAL" in types_set or "OWNED_EQUIPMENT" in types_set has_rental = "RENTAL_EQUIPMENT" in types_set has_service = "SUBCONTRACT_SERVICE" in types_set has_transport = "TRANSPORT_SERVICE" in types_set # 1. Complex/Mixed (Hỗn hợp) if has_material and (has_service or has_rental or has_transport): return {"business_category": "mixed", "business_subcategory": "complex_mixed"} # 2. Input Invoices (Mua vào) if direction == "input": if has_rental: return {"business_category": "commercial", "business_subcategory": "machine_rental_in"} if has_transport: return {"business_category": "service", "business_subcategory": "transport_hire"} if has_service: return {"business_category": "service", "business_subcategory": "construction_work"} return {"business_category": "commercial", "business_subcategory": "purchase_in"} # 3. Output Invoices (Bán ra) if direction == "output": if has_rental: # Cross-reference rule: if we rented it in, renting it out is sublease. # If we didn't rent it in (meaning we own it), it's equipment leasing. if has_rental_input_history: return {"business_category": "commercial", "business_subcategory": "machine_sublease_out"} else: return {"business_category": "service", "business_subcategory": "equipment_leasing"} if has_transport: return {"business_category": "service", "business_subcategory": "transport_hire"} if has_service: return {"business_category": "service", "business_subcategory": "construction_work"} return {"business_category": "commercial", "business_subcategory": "sales_out"} return {"business_category": "unclassified", "business_subcategory": None}