from __future__ import annotations import logging from datetime import date from decimal import Decimal from typing import Any from app.core.postgres.base import BasePostgresClient from app.modules.invoices.application.invoice_processing_service import DSCONS_DEFAULT_TAX_CODE from .constants import KNOWN_PARTNER_TAX_CODES logger = logging.getLogger("dscons.partners.sync") class PartnerSyncMixin(BasePostgresClient): """Invoice extraction and automatic partner/customer/vendor synchronization.""" def extract_and_sync_all_partners_from_invoices(self) -> dict[str, Any]: """ Quét toàn bộ hóa đơn trong erp_invoices, gom nhóm theo MST, xác định tên/địa chỉ mới nhất và ghi nhận lịch sử biến động. """ with self.get_connection() as conn: with conn.cursor() as cur: # 1. Lấy tất cả hóa đơn có thông tin seller hoặc buyer cur.execute(""" SELECT id, direction, invoice_number, invoice_series, issue_date, seller_name, seller_tax_code, seller_address, buyer_name, buyer_tax_code, buyer_address, total_amount_vnd, matched_project_id FROM erp_invoices ORDER BY issue_date ASC, created_at ASC; """) invoices = cur.fetchall() # Gom nhóm theo tax_code của đối tác (bỏ qua MST của Định Sơn) partners_data: dict[str, dict[str, Any]] = {} for inv in invoices: direction = inv["direction"] issue_dt = inv["issue_date"] inv_id = str(inv["id"]) inv_num = inv["invoice_number"] or "" inv_series = inv["invoice_series"] or "" amt = Decimal(str(inv["total_amount_vnd"] or "0.0000")) proj_id = ( str(inv["matched_project_id"]) if inv["matched_project_id"] else None ) # Xác định party ngoài Định Sơn if direction == "output": # Định Sơn bán cho Buyer -> Buyer là Khách hàng p_tax = (inv["buyer_tax_code"] or "").strip() p_name = (inv["buyer_name"] or "").strip() p_addr = (inv["buyer_address"] or "").strip() is_out = True else: # Định Sơn mua từ Seller -> Seller là Nhà cung cấp p_tax = (inv["seller_tax_code"] or "").strip() p_name = (inv["seller_name"] or "").strip() p_addr = (inv["seller_address"] or "").strip() is_out = False if not p_tax: if not p_name: continue p_tax = f"NO_MST_{abs(hash(p_name)) % 1000000:06d}" if p_tax == DSCONS_DEFAULT_TAX_CODE or p_tax == "0202111150": continue if p_tax not in partners_data: # Xác định phân loại cá nhân hay tổ chức is_indiv = ( p_tax.startswith("NO_MST_") or len(p_tax) == 12 # CCCD or "cá nhân" in p_name.lower() or "khách hàng không lấy" in p_name.lower() or "ông" in p_name.lower() or "bà" in p_name.lower() ) category = "individual" if is_indiv else "organization" partners_data[p_tax] = { "tax_code": p_tax, "category": category, "is_customer": False, "is_partner": p_tax in KNOWN_PARTNER_TAX_CODES, "is_vendor": False, "total_output_revenue": Decimal("0.0000"), "total_input_expense": Decimal("0.0000"), "invoices_count": 0, "first_transaction_date": issue_dt, "latest_transaction_date": issue_dt, "history_records": [], # list of (issue_date, inv_id, inv_num, inv_series, name, addr) "matched_projects": set(), } p_entry = partners_data[p_tax] p_entry["invoices_count"] += 1 if proj_id: p_entry["matched_projects"].add(proj_id) p_entry["is_partner"] = True if is_out: p_entry["is_customer"] = True p_entry["total_output_revenue"] += amt else: p_entry["is_vendor"] = True p_entry["total_input_expense"] += amt if issue_dt: if ( p_entry["first_transaction_date"] is None or issue_dt < p_entry["first_transaction_date"] ): p_entry["first_transaction_date"] = issue_dt if ( p_entry["latest_transaction_date"] is None or issue_dt > p_entry["latest_transaction_date"] ): p_entry["latest_transaction_date"] = issue_dt # Ghi nhận bản ghi thông tin hóa đơn if p_name: p_entry["history_records"].append( { "recorded_date": issue_dt or date(2023, 1, 1), "invoice_id": inv_id, "invoice_number": inv_num, "invoice_series": inv_series, "name": p_name, "address": p_addr, } ) # 2. Xử lý lưu trữ vào CSDL cur.execute("DELETE FROM erp_partner_history;") cur.execute("DELETE FROM erp_partners;") synced_count = 0 history_total_count = 0 for tax_code, p in partners_data.items(): # Sắp xếp các bản ghi lịch sử theo ngày records = p["history_records"] records.sort( key=lambda r: (r["recorded_date"], r["invoice_number"]) ) # Tên và địa chỉ mới nhất lấy từ bản ghi cuối cùng latest_name = records[-1]["name"] if records else tax_code latest_addr = records[-1]["address"] if records else "" # Thu thập các thay đổi lịch sử thực sự (tên khác hoặc địa chỉ khác so với bản ghi trước đó) historical_changes = [] seen_combos = set() for rec in records: combo = ( rec["name"].strip().lower(), rec["address"].strip().lower(), ) if combo not in seen_combos: seen_combos.add(combo) historical_changes.append(rec) # Số lần đổi thông tin (nếu có > 1 biến thể tên/địa chỉ) history_change_count = ( len(historical_changes) - 1 if len(historical_changes) > 1 else 0 ) # Insert vào erp_partners cur.execute( """ INSERT INTO erp_partners ( tax_code, company_name, partner_category, current_address, is_customer, is_partner, is_vendor, total_output_revenue_vnd, total_input_expense_vnd, total_invoices_count, first_transaction_date, latest_transaction_date, history_change_count, status ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'active') RETURNING id; """, ( tax_code, latest_name, p["category"], latest_addr, p["is_customer"], p["is_partner"], p["is_vendor"], p["total_output_revenue"], p["total_input_expense"], p["invoices_count"], p["first_transaction_date"], p["latest_transaction_date"], history_change_count, ), ) partner_id = str(cur.fetchone()["id"]) synced_count += 1 # Insert lịch sử thay đổi vào erp_partner_history (nếu có lịch sử) for h in historical_changes: cur.execute( """ INSERT INTO erp_partner_history ( partner_id, tax_code, historical_name, historical_address, source_invoice_id, source_invoice_number, source_invoice_series, recorded_date ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s); """, ( partner_id, tax_code, h["name"], h["address"], h["invoice_id"], h["invoice_number"], h["invoice_series"], h["recorded_date"], ), ) history_total_count += 1 conn.commit() return { "status": "success", "total_partners_synced": synced_count, "total_history_records_created": history_total_count, }