from __future__ import annotations """Tự động hóa toàn diện: Làm giàu chi tiết dòng hàng GDT và Phân loại - Hạch toán kép 4 Trụ Cột cho toàn bộ Hóa đơn Đầu vào DSCons.""" import json import logging import ssl import sys import time import urllib.error import urllib.request from decimal import Decimal from typing import Any from app.core.postgres.erp_client import ErpDatabaseClient from app.modules.invoices.application.invoice_processing.common import to_decimal from app.modules.invoices.application.invoice_processing.four_pillars_accounting_engine import ( FourPillarsAccountingEngine, ) from app.modules.invoices.application.invoice_processing_service import ( InvoiceProcessingService, ) logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[logging.StreamHandler(sys.stdout)], ) logger = logging.getLogger("AutoProcessInvoices") class BatchInvoiceProcessor: def __init__(self): self.svc = InvoiceProcessingService() self.db = ErpDatabaseClient() self.gdt_token: str | None = None self.gdt_cfg: dict[str, Any] | None = None self.gdt_password: str = "" self.gdt_extra: dict[str, Any] = {} self.default_company_id: str | None = None self.default_project_id: str | None = None self._init_defaults() def _init_defaults(self): configs = self.svc.get_sync_configs() self.gdt_cfg = next((c for c in configs if c["service_type"] == "gdt_portal"), None) with self.db.get_connection() as conn: with conn.cursor() as cur: if self.gdt_cfg: cur.execute( "SELECT password_hash, extra_config FROM erp_invoice_sync_configs WHERE id = %s", (self.gdt_cfg["id"],), ) row = cur.fetchone() if row: self.gdt_password = row["password_hash"] self.gdt_extra = row.get("extra_config") or {} cur.execute("SELECT id FROM companies WHERE tax_code = '0202111150' LIMIT 1;") comp = cur.fetchone() if comp: self.default_company_id = str(comp["id"]) cur.execute("SELECT id FROM projects ORDER BY created_at ASC LIMIT 1;") proj = cur.fetchone() if proj: self.default_project_id = str(proj["id"]) def ensure_token(self) -> str | None: if self.gdt_token: return self.gdt_token if not self.gdt_cfg: return None logger.info("Đang xác thực với Cổng Tổng Cục Thuế để lấy Token phiên làm việc...") auth = self.svc._authenticate_gdt_portal( self.gdt_cfg["tax_code"], self.gdt_cfg["username"], self.gdt_password, extra_config=self.gdt_extra, ) if auth.get("status") == "success" and auth.get("token"): self.gdt_token = auth["token"] logger.info("Xác thực GDT thành công! Token đã sẵn sàng.") return self.gdt_token else: logger.error(f"Lỗi xác thực GDT: {auth.get('message')}") return None def enrich_invoices_from_gdt(self, limit: int | None = None) -> int: """Fetch line items from GDT for all input invoices currently missing items with rate-limit handling.""" with self.db.get_connection() as conn: with conn.cursor() as cur: sql = """ SELECT i.id, i.invoice_number, i.invoice_series, i.seller_tax_code, i.seller_name, i.template_code, i.issue_date, i.total_amount_vnd FROM erp_invoices i LEFT JOIN erp_invoice_items it ON i.id = it.invoice_id WHERE (i.buyer_tax_code = %s OR i.direction = %s) AND it.id IS NULL ORDER BY i.issue_date DESC """ if limit: sql += f" LIMIT {limit}" cur.execute(sql, ("0202111150", "input")) invoices = cur.fetchall() total_to_enrich = len(invoices) logger.info(f"Tổng số hóa đơn đầu vào cần làm giàu chi tiết dòng hàng từ GDT: {total_to_enrich}") if total_to_enrich == 0: return 0 token = self.ensure_token() if not token: logger.error("Không có token GDT, bỏ qua bước làm giàu dữ liệu.") return 0 ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE headers = { "Authorization": f"Bearer {token}", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) DSConsERP/2.4.0", "Accept": "application/json, text/plain, */*", } success_count = 0 for idx, inv in enumerate(invoices, 1): inv_id = str(inv["id"]) inv_num = str(inv["invoice_number"]) series = str(inv["invoice_series"]) seller_tax = str(inv["seller_tax_code"]) tmpl = str(inv.get("template_code", "1")) khh_clean = series.removeprefix(tmpl) shdon = inv_num.lstrip("0") or "1" url = f"https://hoadondientu.gdt.gov.vn/api/query/invoices/detail?nbmst={seller_tax}&khhdon={khh_clean}&shdon={shdon}&khmshdon={tmpl}" for attempt in range(1, 4): try: req = urllib.request.Request(url, headers=headers) with urllib.request.urlopen(req, timeout=12, context=ctx) as res: raw_d = json.loads(res.read().decode("utf-8")) raw_items = raw_d.get("hdhhdvu", []) if raw_items: self._save_raw_items_to_db(inv_id, raw_items, raw_d) success_count += 1 # Polite pacing to avoid GDT 429 time.sleep(0.5) break except urllib.error.HTTPError as he: if he.code == 429: wait_time = 4.0 * attempt logger.warning(f"GDT 429 (Rate limit) tại HĐ #{inv_num}. Đang tạm dừng {wait_time:.1f}s trước khi thử lại...") time.sleep(wait_time) continue elif he.code in (401, 403) and attempt == 1: logger.warning("Token GDT hết hạn, đang tự động làm mới...") self.gdt_token = None new_token = self.ensure_token() if new_token: headers["Authorization"] = f"Bearer {new_token}" time.sleep(1.0) continue break except Exception: time.sleep(0.5) break if idx % 10 == 0 or idx == total_to_enrich: logger.info(f"Tiến độ làm giàu GDT: [{idx}/{total_to_enrich}] (Đã lưu chi tiết: {success_count} HĐ)") logger.info(f"Hoàn thành làm giàu chi tiết dòng hàng GDT: {success_count}/{total_to_enrich} hóa đơn thành công.") return success_count def _save_raw_items_to_db(self, invoice_id: str, raw_items: list[dict], raw_d: dict): with self.db.get_connection() as conn: with conn.cursor() as cur: # Update header from detail cur.execute( """ UPDATE erp_invoices SET xml_raw_content = %s, updated_at = CURRENT_TIMESTAMP WHERE id = %s; """, (json.dumps(raw_d, ensure_ascii=False), invoice_id), ) # Remove dummy/generic items if any cur.execute("DELETE FROM erp_invoice_items WHERE invoice_id = %s;", (invoice_id,)) insert_item_sql = """ INSERT INTO erp_invoice_items ( invoice_id, item_order, item_name, item_code, unit, quantity, unit_price_vnd, amount_before_vat_vnd, vat_rate_percent, vat_amount_vnd, total_item_amount_vnd, cost_category ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s ); """ for itm in raw_items: stt = int(itm.get("stt") or 1) name = str(itm.get("ten") or "Hàng hóa / Dịch vụ").strip() code = str(itm.get("mhhdvu") or itm.get("id") or "").strip() unit = str(itm.get("dvtinh") or "Gói").strip() qty = to_decimal(itm.get("sluong"), default=Decimal("0.0000")) unit_price = to_decimal(itm.get("dgia"), default=Decimal("0.0000")) amount_before = to_decimal(itm.get("thtien"), default=Decimal("0.0000")) vat_str = str(itm.get("ltsuat") or "8%").replace("%", "").strip() try: vat_rate = to_decimal(vat_str, default=Decimal("8.0000")) except Exception: vat_rate = Decimal("8.0000") vat_amt = ( to_decimal(itm.get("tthue")) if itm.get("tthue") is not None else (amount_before * vat_rate / Decimal(100)) ) tot_amt = amount_before + vat_amt cost_cat = self.svc.categorize_item_cost(name, code) cur.execute( insert_item_sql, ( invoice_id, stt, name, code, unit, qty, unit_price, amount_before, vat_rate, vat_amt, tot_amt, cost_cat, ), ) conn.commit() def classify_and_account_all_input_invoices(self) -> dict[str, Any]: """Classify and generate double-entry accounting entries for all input invoices.""" logger.info("Bắt đầu phân loại 4 Trụ Cột và định khoản kế toán kép cho toàn bộ hóa đơn đầu vào...") with self.db.get_connection() as conn: with conn.cursor() as cur: cur.execute(""" SELECT i.* FROM erp_invoices i WHERE i.buyer_tax_code = %s OR i.direction = %s ORDER BY i.issue_date DESC; """, ("0202111150", "input")) invoices = cur.fetchall() cur.execute(""" SELECT it.* FROM erp_invoice_items it JOIN erp_invoices i ON it.invoice_id = i.id WHERE i.buyer_tax_code = %s OR i.direction = %s ORDER BY it.invoice_id, it.item_order; """, ("0202111150", "input")) all_items = cur.fetchall() items_by_inv: dict[str, list[dict]] = {} for it in all_items: iid = str(it["invoice_id"]) if iid not in items_by_inv: items_by_inv[iid] = [] items_by_inv[iid].append(it) stats = { "total_processed": len(invoices), "pillar_1": {"count": 0, "amount": Decimal("0.0000")}, "pillar_2": {"count": 0, "amount": Decimal("0.0000")}, "pillar_3": {"count": 0, "amount": Decimal("0.0000")}, "pillar_4": {"count": 0, "amount": Decimal("0.0000")}, "financial_admin": {"count": 0, "amount": Decimal("0.0000")}, "total_pretax": Decimal("0.0000"), "total_vat": Decimal("0.0000"), "total_amount": Decimal("0.0000"), } with self.db.get_connection() as conn: with conn.cursor() as cur: for inv in invoices: inv_id = str(inv["id"]) items = items_by_inv.get(inv_id, []) tot = Decimal(str(inv.get("total_amount_vnd") or 0)) pretax = Decimal(str(inv.get("subtotal_amount_vnd") or 0)) vat = Decimal(str(inv.get("vat_amount_vnd") or 0)) classification = FourPillarsAccountingEngine.classify_invoice_and_account(inv, items) p_code = classification["pillar_code"] if p_code == "pillar_1": stats["pillar_1"]["count"] += 1 stats["pillar_1"]["amount"] += tot elif p_code == "pillar_2": stats["pillar_2"]["count"] += 1 stats["pillar_2"]["amount"] += tot elif p_code == "pillar_3": stats["pillar_3"]["count"] += 1 stats["pillar_3"]["amount"] += tot elif p_code == "pillar_4": stats["pillar_4"]["count"] += 1 stats["pillar_4"]["amount"] += tot else: stats["financial_admin"]["count"] += 1 stats["financial_admin"]["amount"] += tot stats["total_pretax"] += pretax stats["total_vat"] += vat stats["total_amount"] += tot pillar_title = classification["pillar"] debit = classification["debit_account"] credit = classification["credit_account"] summary = classification["cost_category_summary"] pay_req = classification["payment_requirement"] structured_note = ( f"[{pillar_title}] [Định khoản: Nợ TK {debit}, Nợ TK 1331 / Có TK {credit}] " f"[{pay_req}] {summary}" ) cur.execute( """ UPDATE erp_invoices SET notes = %s, updated_at = CURRENT_TIMESTAMP WHERE id = %s; """, (structured_note, inv_id), ) inv_num = inv.get("invoice_number") cur.execute( """ SELECT id FROM erp_financial_transactions WHERE invoice_number = %s AND direction = 'outflow' LIMIT 1; """, (inv_num,), ) exists_tx = cur.fetchone() if not exists_tx and tot > 0: proj_id = inv.get("matched_project_id") or self.default_project_id comp_id = self.default_company_id if proj_id and comp_id: tx_code = f"EXP-{inv.get('invoice_series')}-{inv_num}" cur.execute( """ INSERT INTO erp_financial_transactions ( company_id, project_id, amount, transaction_date, direction, transaction_code, transaction_type, payment_method, invoice_number, beneficiary_or_payer, notes, accounting_status ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s ); """, ( comp_id, proj_id, tot, inv.get("issue_date"), "outflow", tx_code, "INVOICE_EXPENSE", "bank_transfer" if tot >= 20_000_000 else "cash_or_bank", inv_num, inv.get("seller_name"), f"Bút toán Nợ TK {debit}, Nợ TK 1331 / Có TK {credit} - {pillar_title}", "recorded", ), ) conn.commit() logger.info("Đã hoàn tất phân loại và định khoản kế toán 100% hóa đơn đầu vào!") return stats if __name__ == "__main__": processor = BatchInvoiceProcessor() # Step 1: Enrich line items from GDT for ALL remaining historical invoices logger.info("Giai đoạn 1: Làm giàu chi tiết dòng hàng thực tế từ Cổng Thuế điện tử GDT cho toàn bộ các năm còn lại...") processor.enrich_invoices_from_gdt(limit=None) # Step 2: Classify and account for all input invoices logger.info("Giai đoạn 2: Phân loại 4 Trụ Cột và định khoản kế toán ghi sổ kép...") results = processor.classify_and_account_all_input_invoices() print("\n" + "=" * 80) print("BÁO CÁO TỔNG HỢP KẾT QUẢ HẠCH TOÁN 4 TRỤ CỘT ĐẦU VÀO (DSCons)") print("=" * 80) print(f"Tổng số hóa đơn đầu vào đã xử lý: {results['total_processed']} hóa đơn") print(f"Tổng tiền hàng trước thuế: {results['total_pretax']:,.0f} VNĐ") print(f"Tổng thuế GTGT đầu vào (TK 1331): {results['total_vat']:,.0f} VNĐ") print(f"Tổng giá trị thanh toán (TK 331): {results['total_amount']:,.0f} VNĐ") print("-" * 80) print(f"1. Trụ Cột 1 (Thi công Xây lắp & Thủy lợi - TK 621/622/627):") print(f" -> Số lượng: {results['pillar_1']['count']:>4} HĐ | Tổng giá trị: {results['pillar_1']['amount']:>18,.0f} VNĐ") print(f"2. Trụ Cột 2 (Vận tải Xe Ben & Logistics Cơ Giới - TK 154):") print(f" -> Số lượng: {results['pillar_2']['count']:>4} HĐ | Tổng giá trị: {results['pillar_2']['amount']:>18,.0f} VNĐ") print(f"3. Trụ Cột 3 (Cho thuê Ca máy & Bảo dưỡng Cơ giới - TK 623):") print(f" -> Số lượng: {results['pillar_3']['count']:>4} HĐ | Tổng giá trị: {results['pillar_3']['amount']:>18,.0f} VNĐ") print(f"4. Trụ Cột 4 (Mua bán Vật tư Thương mại & San lấp - TK 156):") print(f" -> Số lượng: {results['pillar_4']['count']:>4} HĐ | Tổng giá trị: {results['pillar_4']['amount']:>18,.0f} VNĐ") print(f"5. Quản lý Doanh nghiệp & Chi phí Ngân hàng (TK 642/635):") print(f" -> Số lượng: {results['financial_admin']['count']:>4} HĐ | Tổng giá trị: {results['financial_admin']['amount']:>18,.0f} VNĐ") print("=" * 80)