"""Classifier and Router for Bank Statements and Payment Orders (UNC). Automatically detects if an uploaded document is a bank statement or UNC, classifies the issuing bank, safely routes the file to the dedicated Hot Watcher folder, and triggers automatic transaction parsing and ingestion. """ from __future__ import annotations import hashlib import io import logging import os import re import shutil from datetime import datetime from decimal import Decimal from typing import Any from uuid import UUID, uuid4 import openpyxl import pandas as pd import pypdf from app.core.postgres.base_pkg.base_client import BasePostgresClient from app.modules.banking.application.parsers.acb_excel_parser import ACBExcelParser from app.modules.banking.application.parsers.tcb_pdf_parser import TCBPdfParser from app.modules.banking.application.parsers.vcb_excel_parser import VCBExcelParser from app.modules.banking.application.parsers.vpbank_excel_parser import VPBankExcelParser logger = logging.getLogger("dscons.banking.unc_classifier_and_router") from pathlib import Path _UNC_BASE_DIR = Path(__file__).resolve().parents[4] HOT_WATCHER_ROOT = os.getenv( "UNC_HOT_WATCHER_ROOT", str(_UNC_BASE_DIR / "Finance" / "UNC-2026"), ) KNOWN_PASSWORDS = ["39912136", "0202111150"] BANK_FOLDER_MAP = { "ACB": "ACB-2026", "TECHCOMBANK": "Techcombank-2026", "VPBANK": "VPBank 2026", "VIETCOMBANK": "Vietcombank-2026", "VIETINBANK": "Vietinbank-2026", "BIDV": "BIDV-2026", "OTHER": "Other-2026", } DEFAULT_ACCOUNTS = { "ACB": ("13456888888", "CÔNG TY TNHH XÂY DỰNG ĐỊNH SƠN"), "TECHCOMBANK": ("19039912136018", "CÔNG TY TNHH XÂY DỰNG ĐỊNH SƠN"), "VPBANK": ("8667898888", "CÔNG TY TNHH XÂY DỰNG ĐỊNH SƠN"), "VIETCOMBANK": ("UNKNOWN", "CÔNG TY TNHH XÂY DỰNG ĐỊNH SƠN"), "OTHER": ("UNKNOWN", "CÔNG TY TNHH XÂY DỰNG ĐỊNH SƠN"), } class UncClassifierAndRouter: """Detects UNC / Statements, routes files to Hot Watcher directories, and triggers ingestion.""" def __init__(self, postgres_client: Any | None = None) -> None: self._client = postgres_client or BasePostgresClient() @staticmethod def detect_unc_and_bank( filename: str, text: str = "", content: bytes | None = None, enable_vision_ocr: bool = True, ) -> tuple[bool, str | None, dict[str, Any]]: """Determines if a document is an UNC / Bank Statement and identifies the bank. Returns: (is_unc, bank_code, details_dict) """ fn_lower = filename.lower() combined_text = f"{fn_lower} {text}".upper() # 1. Filename heuristic checks unc_filename_keywords = [ "unc", "uy nhiem chi", "uynhiemchi", "uy_nhiem_chi", "saoke", "sao ke", "sao_ke", "statement", "accountstmt", "debitnote", "creditnote", "debit_note", "credit_note", "transaction_history", "b2b", "pass.docx", ] is_filename_unc = any(k in fn_lower for k in unc_filename_keywords) # 2. Text content heuristic checks unc_content_keywords = [ "ỦY NHIỆM CHI", "UY NHIEM CHI", "LỆNH CHUYỂN TIỀN", "LENH CHUYEN TIEN", "GIẤY BÁO NỢ", "GIAY BAO NO", "GIẤY BÁO CÓ", "GIAY BAO CO", "SAO KÊ TÀI KHOẢN", "SAO KE TAI KHOAN", "ACCOUNT STATEMENT", "TRANSACTION HISTORY", "BẢNG KÊ GIAO DỊCH", "BANG KE GIAO DICH", "SỔ PHỤ TÀI KHOẢN", "SO PHU TAI KHOAN", "THÔNG BÁO GIAO DỊCH", "TÀI KHOẢN TRÍCH NỢ", "ĐƠN VỊ THỤ HƯỞNG", "NGÂN HÀNG THỤ HƯỞNG", "SỐ DƯ SAU GIAO DỊCH", "RUNNING BALANCE", "VALUE DATE", ] is_content_unc = any(k in combined_text for k in unc_content_keywords) # Quick in-memory text extraction from PDF if text is empty if not text and content and not is_filename_unc: ext = fn_lower.split(".")[-1] if "." in fn_lower else "" if ext == "pdf": try: import pymupdf doc = pymupdf.open(stream=content, filetype="pdf") extracted_pages = [] for page in doc[:3]: p_txt = page.get_text("text").strip() if p_txt: extracted_pages.append(p_txt) doc.close() if extracted_pages: text = "\n".join(extracted_pages) combined_text = f"{fn_lower} {text}".upper() is_content_unc = any(k in combined_text for k in unc_content_keywords) except Exception: try: reader = pypdf.PdfReader(io.BytesIO(content)) extracted_pages = [page.extract_text() or "" for page in reader.pages[:3]] if any(extracted_pages): text = "\n".join(extracted_pages) combined_text = f"{fn_lower} {text}".upper() is_content_unc = any(k in combined_text for k in unc_content_keywords) except Exception: pass # Detect bank bank_code = None if any(k in combined_text for k in ["VPBANK", "VIỆT NAM THỊNH VƯỢNG", "VIET NAM THINH VUONG", "8667898888", "ACCOUNTSTMT", "PDLD", "LD25", "LD26"]): bank_code = "VPBANK" elif any(k in combined_text for k in ["TECHCOMBANK", "KỸ THƯƠNG", "KY THUONG", "TCB", "07898888", "XXXXXXXXXX8888", "TT261"]): bank_code = "TECHCOMBANK" elif any(k in combined_text for k in ["ACB", "Á CHÂU", "A CHAU", "ASCB", "13456888888"]): bank_code = "ACB" elif any(k in combined_text for k in ["VIETCOMBANK", "NGOẠI THƯƠNG", "NGOAI THUONG", "VCB"]): bank_code = "VIETCOMBANK" elif any(k in combined_text for k in ["VIETINBANK", "CÔNG THƯƠNG", "CONG THUONG", "CTG"]): bank_code = "VIETINBANK" elif any(k in combined_text for k in ["BIDV", "ĐẦU TƯ VÀ PHÁT TRIỂN"]): bank_code = "BIDV" # Edge case: If filename or content clearly matches a bank and has statement pattern if not bank_code and is_filename_unc: bank_code = "OTHER" is_unc = bool(is_filename_unc or is_content_unc or (bank_code and any(ext in fn_lower for ext in [".xls", ".xlsx", ".pdf"]))) # 3. Multimodal OCR deep inspection if enabled, content provided, and not large non-bank dossier ocr_matched = False ocr_details = None is_large_non_bank = content and len(content) > 2_000_000 and not is_filename_unc and not is_content_unc if enable_vision_ocr and not is_large_non_bank and (not is_unc or not bank_code or bank_code == "OTHER") and content: ext = fn_lower.split(".")[-1] if "." in fn_lower else "" if ext in ("pdf", "png", "jpg", "jpeg", "webp", "tiff", "tif"): try: from app.modules.core.application.ocr.multimodal_ocr_service import MultimodalOcrService ocr_svc = MultimodalOcrService() ocr_res = ocr_svc.extract_banking_unc_data(content, filename, ext) if ocr_res and (ocr_res.get("amount") or ocr_res.get("bank_code")): if ocr_res.get("amount") and ocr_res.get("amount") > 0: is_unc = True ocr_matched = True ocr_details = ocr_res if ocr_res.get("bank_code") and ocr_res["bank_code"] != "OTHER": bank_code = ocr_res["bank_code"] except Exception as ocr_e: logger.debug("OCR classification check failed: %s", ocr_e) details = { "is_unc": is_unc, "bank_code": bank_code, "filename_matched": is_filename_unc, "content_matched": is_content_unc, "ocr_matched": ocr_matched, "ocr_data": ocr_details, "confidence": 0.95 if (is_filename_unc and bank_code) or ocr_matched else (0.85 if is_unc else 0.1), } return is_unc, bank_code, details @staticmethod def route_to_hot_folder( file_content: bytes, filename: str, bank_code: str | None = None, ) -> tuple[str, str]: """Saves file into the appropriate Hot Watcher folder. Returns: (target_folder_path, target_file_path) """ bank = (bank_code or "OTHER").upper() folder_name = BANK_FOLDER_MAP.get(bank, "Other-2026") target_dir = os.path.join(HOT_WATCHER_ROOT, folder_name) os.makedirs(target_dir, exist_ok=True) target_file_path = os.path.join(target_dir, filename) # If file exists, check hash to avoid duplicate write if os.path.exists(target_file_path): with open(target_file_path, "rb") as f: existing_hash = hashlib.sha256(f.read()).hexdigest() incoming_hash = hashlib.sha256(file_content).hexdigest() if existing_hash == incoming_hash: logger.info("File already exists with identical content in hot folder: %s", target_file_path) return target_dir, target_file_path # If different content, append timestamp to preserve both base, ext = os.path.splitext(filename) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") new_filename = f"{base}_{timestamp}{ext}" target_file_path = os.path.join(target_dir, new_filename) with open(target_file_path, "wb") as f: f.write(file_content) logger.info("Routed UNC file to hot folder: %s", target_file_path) return target_dir, target_file_path def ingest_statement_file( self, file_path: str, bank_code: str | None = None, file_content: bytes | None = None, ) -> dict[str, Any]: """Parses a bank statement file and inserts transactions into erp_bank_transactions.""" if not file_content: if not os.path.exists(file_path): raise FileNotFoundError(f"File not found: {file_path}") with open(file_path, "rb") as f: file_content = f.read() filename = os.path.basename(file_path) fn_lower = filename.lower() # Auto-detect bank if not provided if not bank_code: _, detected_bank, _ = self.detect_unc_and_bank(filename, content=file_content) bank_code = detected_bank or "OTHER" bank_code = bank_code.upper() parsed_txs: list[dict[str, Any]] = [] # 1. Parse based on bank and format if bank_code == "TECHCOMBANK" or "techcombank" in fn_lower: if fn_lower.endswith(".pdf"): # Try with known passwords success = False for pwd in KNOWN_PASSWORDS: try: parser = TCBPdfParser(password=pwd) parsed_txs = parser.parse(io.BytesIO(file_content), filename) if parsed_txs: success = True break except Exception: continue if not success and not parsed_txs: # Try without password try: parser = TCBPdfParser(password=None) parsed_txs = parser.parse(io.BytesIO(file_content), filename) except Exception as e: logger.warning("TCB PDF parser failed without password: %s", e) elif fn_lower.endswith((".xlsx", ".xls")): # Read TCB Excel format parsed_txs = self._parse_generic_or_tcb_excel(file_content, filename) elif bank_code == "ACB" or "acb" in fn_lower or "13456888888" in fn_lower: if fn_lower.endswith((".xlsx", ".xls")): try: parser = ACBExcelParser() parsed_txs = parser.parse(io.BytesIO(file_content), filename) except Exception as e: logger.warning("ACB parser failed, trying generic: %s", e) parsed_txs = self._parse_generic_or_tcb_excel(file_content, filename) elif bank_code == "VPBANK" or "vpbank" in fn_lower or "accountstmt" in fn_lower: if fn_lower.endswith((".xls", ".xlsx")): try: parser = VPBankExcelParser() parsed_txs = parser.parse(io.BytesIO(file_content), filename) except Exception as e: logger.warning("VPBank parser failed, trying xlrd direct: %s", e) parsed_txs = self._parse_vpbank_direct(file_content, filename) else: # Fallback parser try: parser = VCBExcelParser() parsed_txs = parser.parse(io.BytesIO(file_content), filename) except Exception: parsed_txs = self._parse_generic_or_tcb_excel(file_content, filename) # 1.1. Multimodal OCR fallback for single-sheet UNCs, Debit Notes, and Image vouchers if not parsed_txs: try: from app.modules.core.application.ocr.multimodal_ocr_service import ( MultimodalOcrService, ) ocr_service = MultimodalOcrService() _, ext = os.path.splitext(filename) unc_data = ocr_service.extract_banking_unc_data( file_content, filename, ext.lstrip(".") ) if unc_data and unc_data.get("amount"): amt = Decimal(str(unc_data["amount"])) direction = unc_data.get("direction", "outflow") debit = amt if direction == "outflow" else Decimal("0") credit = amt if direction == "inflow" else Decimal("0") txn_date_str = unc_data.get("txn_date") or datetime.now().strftime( "%Y-%m-%d" ) txn_date = datetime.strptime(txn_date_str, "%Y-%m-%d").date() # Update bank_code if detected by OCR if unc_data.get("bank_code") and unc_data["bank_code"] != "OTHER": bank_code = unc_data["bank_code"] acc_no = unc_data.get("sender_account") or DEFAULT_ACCOUNTS.get( bank_code, ("UNKNOWN", "") )[0] acc_name = unc_data.get("sender_name") or DEFAULT_ACCOUNTS.get( bank_code, ("", "CÔNG TY TNHH XÂY DỰNG ĐỊNH SƠN") )[1] parsed_txs.append( { "bank_code": bank_code, "account_number": acc_no, "account_name": acc_name, "transaction_date": txn_date, "value_date": txn_date, "reference_number": unc_data.get("ref_no") or f"OCR-{datetime.now().strftime('%Y%m%d%H%M%S')}", "debit_amount_vnd": debit, "credit_amount_vnd": credit, "running_balance_vnd": Decimal("0"), "description": unc_data.get("description") or f"Thanh toán UNC {filename}", "counterparty_account": unc_data.get("receiver_account"), "counterparty_name": unc_data.get("receiver_name"), "counterparty_bank": unc_data.get("receiver_bank"), } ) logger.info( "Multimodal OCR successfully parsed UNC transaction: %s, amount=%s", unc_data.get("ref_no"), amt, ) except Exception as ocr_err: logger.warning("Multimodal OCR fallback parsing error: %s", ocr_err) # 2. Insert into PostgreSQL erp_bank_transactions saved_count = 0 saved_ids: list[Any] = [] default_acc_no, default_acc_name = DEFAULT_ACCOUNTS.get( bank_code, ("UNKNOWN", "CÔNG TY TNHH XÂY DỰNG ĐỊNH SƠN") ) get_conn = getattr(self._client, "get_connection", None) or getattr(self._client, "_open_connection", None) with get_conn() as conn: with conn.cursor() as cur: # Get default company ID cur.execute("SELECT id FROM companies WHERE tax_code = '0202111150' LIMIT 1") row = cur.fetchone() if not row: cur.execute("SELECT id FROM companies LIMIT 1") row = cur.fetchone() company_id = row["id"] if row else uuid4() for tx in parsed_txs: t_date = tx.get("transaction_date") if isinstance(t_date, datetime): t_date = t_date.date() if not t_date: continue debit = Decimal(str(tx.get("debit_amount_vnd") or 0)) credit = Decimal(str(tx.get("credit_amount_vnd") or 0)) if credit > 0: direction = "inflow" amount = credit elif debit > 0: direction = "outflow" amount = debit else: continue desc = (tx.get("description") or "").strip() ref = ( tx.get("transaction_reference") or tx.get("reference_number") or f"TXN-{uuid4().hex[:8].upper()}" ) balance = Decimal(str(tx.get("running_balance_vnd") or 0)) if tx.get("running_balance_vnd") is not None else None tx_acc_no = tx.get("account_number") or default_acc_no tx_acc_name = tx.get("account_name") or default_acc_name category = self._categorize_transaction(desc, direction) cur.execute( """ INSERT INTO erp_bank_transactions ( company_id, bank_name, account_number, account_name, transaction_date, reference_number, direction, amount, running_balance, description, transaction_category, source_file, created_at ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW() ) ON CONFLICT (bank_name, account_number, reference_number, transaction_date, amount, direction) DO NOTHING RETURNING id """, ( company_id, bank_code, tx_acc_no, tx_acc_name, t_date, ref, direction, amount, balance, desc, category, filename, ), ) res = cur.fetchone() if res: saved_count += 1 saved_ids.append(res["id"]) conn.commit() # 3. Kích hoạt Smart Clearing tự động gạch nợ cho các giao dịch mới nạp cleared_count = 0 if saved_ids: try: from app.modules.financial.application.smart_clearing_service import SmartClearingService clearing_svc = SmartClearingService() for tx_id in saved_ids: settled = clearing_svc.reconcile_single_transaction(tx_id) if settled: cleared_count += len(settled) if cleared_count > 0: logger.info("Auto-cleared %d debt settlements for statement '%s'", cleared_count, filename) except Exception as cl_err: logger.warning("Smart Clearing trigger encountered an error: %s", cl_err) logger.info( "Ingested statement file '%s' (Bank: %s): parsed %d txs, newly saved %d txs, auto-cleared %d settlements", filename, bank_code, len(parsed_txs), saved_count, cleared_count, ) return { "status": "success", "file_name": filename, "bank_code": bank_code, "total_parsed": len(parsed_txs), "newly_saved": saved_count, "inserted": saved_count, "skipped": len(parsed_txs) - saved_count, "cleared_settlements": cleared_count, "file_path": file_path, } def _parse_generic_or_tcb_excel(self, content: bytes, filename: str) -> list[dict[str, Any]]: """Parses generic or Techcombank XLSX statement.""" transactions = [] try: wb = openpyxl.load_workbook(io.BytesIO(content), data_only=True) for sheet in wb.sheetnames: ws = wb[sheet] for row in ws.iter_rows(values_only=True): if not row or len(row) < 5: continue r0_str = str(row[0]).strip() if not re.match(r"\d{2}/\d{2}/\d{4}", r0_str): continue try: tx_date = datetime.strptime(r0_str[:10], "%d/%m/%Y") ref = str(row[1]) if len(row) > 1 and row[1] else None desc = str(row[2]) if len(row) > 2 and row[2] else "" debit = float(str(row[3]).replace(",", "")) if len(row) > 3 and row[3] else 0.0 credit = float(str(row[4]).replace(",", "")) if len(row) > 4 and row[4] else 0.0 balance = float(str(row[5]).replace(",", "")) if len(row) > 5 and row[5] else None transactions.append({ "transaction_date": tx_date, "transaction_reference": ref, "description": desc, "debit_amount_vnd": debit, "credit_amount_vnd": credit, "running_balance_vnd": balance, }) except Exception: continue except Exception as e: logger.warning("Generic Excel parser error: %s", e) return transactions def _parse_vpbank_direct(self, content: bytes, filename: str) -> list[dict[str, Any]]: """Direct parser for VPBank statements using xlrd or openpyxl.""" transactions = [] try: df = pd.read_excel(io.BytesIO(content)) is_data = False for _, row in df.iterrows(): stt = str(row.iloc[0]).strip() if stt == "1": is_data = True if not is_data: continue if pd.isna(row.iloc[0]) or "Tổng số tiền" in str(row.iloc[1]): break date_str = str(row.iloc[2]).strip() try: tx_date = datetime.strptime(date_str[:10], "%d/%m/%Y") ref = str(row.iloc[1]) credit = float(str(row.iloc[3]).replace(",", "")) if not pd.isna(row.iloc[3]) else 0.0 debit = float(str(row.iloc[4]).replace(",", "")) if not pd.isna(row.iloc[4]) else 0.0 desc = str(row.iloc[5]) if not pd.isna(row.iloc[5]) else "" balance = float(str(row.iloc[6]).replace(",", "")) if len(row) > 6 and not pd.isna(row.iloc[6]) else None transactions.append({ "transaction_date": tx_date, "transaction_reference": ref, "description": desc, "debit_amount_vnd": debit, "credit_amount_vnd": credit, "running_balance_vnd": balance, }) except Exception: continue except Exception as e: logger.warning("VPBank direct parse failed: %s", e) return transactions @staticmethod def _categorize_transaction(desc: str, direction: str) -> str: d_upper = desc.upper() if any(k in d_upper for k in ["LUONG", "LƯƠNG", "BHXH"]): return "SALARY" if any(k in d_upper for k in ["THUE", "THUẾ", "NTDT+KB"]): return "TAX" if any(k in d_upper for k in ["NO GOC", "NỢ GỐC", "LAI VAY", "LÃI VAY", "THANH TOAN LAI", "LAI-LD", "PDLD"]): return "LOAN_REPAYMENT" if any(k in d_upper for k in ["RUT TIEN", "RÚT TIỀN", "QUY TM"]): return "CASH_WITHDRAWAL" if any(k in d_upper for k in ["CHUYEN SANG TK", "NOI BO", "NỘI BỘ"]): return "INTERNAL_TRANSFER" if direction == "inflow": return "REVENUE_COLLECTION" return "SUPPLIER_PAYMENT"