from __future__ import annotations """Invoice details retrieval and reconciliation queries.""" import json import logging from typing import Any logger = logging.getLogger(__name__) class InvoiceQueryDetailsMixin: """Mixin for get_invoice_detail, reconcile, and sync configs query.""" def get_invoice_detail(self, invoice_id: str) -> dict[str, Any] | None: """Lấy chi tiết 1 hóa đơn kèm toàn bộ danh sách items và lịch sử audit logs.""" with self.get_connection() as conn: with conn.cursor() as cur: cur.execute( """ SELECT i.*, p.project_name AS matched_project_name, p.project_code AS matched_project_code, w.wbs_code AS matched_wbs_code, w.task_name AS matched_wbs_task_name FROM erp_invoices i LEFT JOIN projects p ON i.matched_project_id = p.id LEFT JOIN erp_project_wbs w ON i.matched_wbs_id = w.id WHERE i.id = %s """, (invoice_id,), ) inv_row = cur.fetchone() if not inv_row: return None inv_data = dict(inv_row) inv_data["id"] = str(inv_data["id"]) if inv_data.get("matched_project_id"): inv_data["matched_project_id"] = str(inv_data["matched_project_id"]) if inv_data.get("matched_wbs_id"): inv_data["matched_wbs_id"] = str(inv_data["matched_wbs_id"]) if inv_data.get("issue_date"): inv_data["issue_date"] = str(inv_data["issue_date"]) if inv_data.get("signed_at"): inv_data["signed_at"] = str(inv_data["signed_at"]) if inv_data.get("created_at"): inv_data["created_at"] = str(inv_data["created_at"]) if inv_data.get("updated_at"): inv_data["updated_at"] = str(inv_data["updated_at"]) for num_col in ( "subtotal_amount_vnd", "vat_rate_percent", "vat_amount_vnd", "total_amount_vnd", "exchange_rate", "remaining_amount_vnd", ): if num_col in inv_data and inv_data[num_col] is not None: inv_data[num_col] = float(inv_data[num_col]) cur.execute( "SELECT * FROM erp_invoice_items WHERE invoice_id = %s ORDER BY item_order ASC", (invoice_id,), ) item_rows = cur.fetchall() items: list[dict[str, Any]] = [] for itm in item_rows: idict = dict(itm) idict["id"] = str(idict["id"]) idict["invoice_id"] = str(idict["invoice_id"]) for num_col in ( "quantity", "unit_price_vnd", "amount_before_vat_vnd", "vat_rate_percent", "vat_amount_vnd", "total_item_amount_vnd", ): if num_col in idict and idict[num_col] is not None: idict[num_col] = float(idict[num_col]) if len(items) == 0 and inv_data.get("xml_raw_content"): raw_str = inv_data["xml_raw_content"] if raw_str.startswith("{"): try: raw_obj = json.loads(raw_str) thttl = raw_obj.get("thttltsuat") or [] if thttl and isinstance(thttl, list) and len(thttl) > 0: for idx, vb in enumerate(thttl, start=1): t_rate = float(str(vb.get("tsuat") or "8").replace("%", "").strip() or 8) t_amt = float(vb.get("thtien") or 0) t_vat = float(vb.get("tthue") or (t_amt * t_rate / 100)) cat_info = self.infer_invoice_category( inv_data.get("seller_name", ""), inv_data.get("notes", "") ) items.append({ "id": f"summary-{idx}", "invoice_id": invoice_id, "item_order": idx, "item_name": f"Cung cấp hàng hóa, dịch vụ theo HĐ {inv_data.get('invoice_number')} ({inv_data.get('invoice_series')})", "item_code": f"GDT-SUM-{idx:02d}", "unit": "Gói", "quantity": 1.0, "unit_price_vnd": t_amt, "amount_before_vat_vnd": t_amt, "vat_rate_percent": t_rate, "vat_amount_vnd": t_vat, "total_item_amount_vnd": t_amt + t_vat, "cost_category": cat_info.get("primary_category", "material"), "is_summary_line": True, }) elif (inv_data.get("subtotal_amount_vnd") or 0) > 0: sub = float(inv_data.get("subtotal_amount_vnd") or 0) vat = float(inv_data.get("vat_amount_vnd") or 0) v_rate = float(inv_data.get("vat_rate_percent") or 8) cat_info = self.infer_invoice_category( inv_data.get("seller_name", ""), inv_data.get("notes", "") ) items.append({ "id": "summary-1", "invoice_id": invoice_id, "item_order": 1, "item_name": f"Cung cấp hàng hóa, dịch vụ theo HĐ {inv_data.get('invoice_number')} ({inv_data.get('invoice_series')})", "item_code": "GDT-SUM-01", "unit": "Gói", "quantity": 1.0, "unit_price_vnd": sub, "amount_before_vat_vnd": sub, "vat_rate_percent": v_rate, "vat_amount_vnd": vat, "total_item_amount_vnd": sub + vat, "cost_category": cat_info.get("primary_category", "material"), "is_summary_line": True, }) except Exception: pass inv_data["items"] = items inv_data["is_cash_register"] = self.is_cash_register_invoice( inv_data.get("invoice_series"), str(inv_data.get("template_code", "")), ) inv_data["invoice_type_info"] = self.get_invoice_type_metadata( inv_data.get("invoice_series"), str(inv_data.get("template_code", "")), inv_data.get("seller_name"), ) inv_data["category_info"] = self.infer_invoice_category( inv_data.get("seller_name", ""), inv_data.get("notes", "") ) # Fetch linked equipment from erp_equipment cur.execute( """ SELECT id, equipment_code, equipment_name, equipment_type, brand_model, purchase_cost_vnd, status, hourly_rate_standard, fuel_norm_per_hour FROM erp_equipment WHERE purchase_invoice_id = %s OR source_invoice_id = %s ORDER BY equipment_code ASC; """, (invoice_id, invoice_id), ) linked_eqs = [dict(eq) for eq in cur.fetchall()] for eq in linked_eqs: eq["id"] = str(eq["id"]) if eq.get("purchase_cost_vnd") is not None: eq["purchase_cost_vnd"] = float(eq["purchase_cost_vnd"]) if eq.get("hourly_rate_standard") is not None: eq["hourly_rate_standard"] = float(eq["hourly_rate_standard"]) if eq.get("fuel_norm_per_hour") is not None: eq["fuel_norm_per_hour"] = float(eq["fuel_norm_per_hour"]) inv_data["linked_equipment"] = linked_eqs cur.execute( "SELECT * FROM erp_invoice_audit_logs WHERE invoice_id = %s ORDER BY created_at DESC", (invoice_id,), ) logs = [dict(l) for l in cur.fetchall()] for l in logs: l["id"] = str(l["id"]) l["invoice_id"] = str(l["invoice_id"]) if l.get("created_at"): l["created_at"] = str(l["created_at"]) inv_data["audit_logs"] = logs # On-demand enrichment: If synced from GDT and only has generic fallback line item, enrich full items from GDT detail API if inv_data.get("source_channel") == "gdt_sync" and ( len(items) == 0 or ( len(items) == 1 and items[0].get("item_code", "").startswith("GDT-") ) ): enriched = self._enrich_gdt_invoice_detail_on_demand( invoice_id, inv_data ) if enriched: return enriched return inv_data def reconcile_invoice( self, invoice_id: str, project_id: str, wbs_id: str | None = None, user_id: str = "SuperAdmin", ) -> dict[str, Any]: """Khớp nối thủ công hóa đơn với dự án và hạng mục WBS.""" with self.get_connection() as conn: with conn.cursor() as cur: cur.execute( """ UPDATE erp_invoices SET matched_project_id = %s, matched_wbs_id = %s, reconciliation_status = 'manual_approved', updated_at = CURRENT_TIMESTAMP WHERE id = %s RETURNING invoice_number, direction, total_amount_vnd; """, (project_id, wbs_id, invoice_id), ) row = cur.fetchone() if not row: return {"status": "error", "message": "Hóa đơn không tồn tại."} cur.execute( """ INSERT INTO erp_invoice_audit_logs (invoice_id, action_type, performed_by, details_json) VALUES (%s, 'RECONCILE_MANUAL', %s, %s); """, ( invoice_id, user_id, json.dumps( { "matched_project_id": project_id, "matched_wbs_id": wbs_id, }, ensure_ascii=False, ), ), ) conn.commit() return { "status": "success", "message": f"Đã khớp thành công hóa đơn số {row['invoice_number']} với dự án.", "invoice_id": invoice_id, } def get_sync_configs(self) -> list[dict[str, Any]]: """Lấy danh sách cấu hình tài khoản kết nối Cổng Thuế, meInvoice và Email IMAP.""" with self.get_connection() as conn: with conn.cursor() as cur: cur.execute(""" SELECT id, service_type, tax_code, username, extra_config, is_active, last_sync_at, last_sync_status, last_sync_message, created_at, updated_at FROM erp_invoice_sync_configs ORDER BY service_type ASC; """) rows = cur.fetchall() results = [] for r in rows: d = dict(r) d["id"] = str(d["id"]) if d.get("last_sync_at"): d["last_sync_at"] = str(d["last_sync_at"]) if d.get("created_at"): d["created_at"] = str(d["created_at"]) if d.get("updated_at"): d["updated_at"] = str(d["updated_at"]) # Never expose password d["has_password"] = True results.append(d) return results