from __future__ import annotations """VAT Tax Schedule calculation and summary grouping.""" import logging from typing import Any logger = logging.getLogger(__name__) class InvoiceVatCalculatorMixin: """Mixin for computing VAT declaration schedules.""" def get_vat_tax_schedule( self, from_date: str | None = None, to_date: str | None = None, vat_period: str | None = None, ) -> dict[str, Any]: """Lập bảng kê thuế GTGT mua vào và bán ra theo mẫu Cục Thuế kèm phân loại nhóm thuế suất.""" with self.get_connection() as conn: with conn.cursor() as cur: where_clauses = ["status = 'valid'"] params: list[Any] = [] if from_date: where_clauses.append("issue_date >= %s") params.append(from_date) if to_date: where_clauses.append("issue_date <= %s") params.append(to_date) where_sql = " AND ".join(where_clauses) # Fetch all valid input invoices cur.execute( f""" SELECT id, invoice_number, invoice_series, template_code, issue_date, seller_tax_code, seller_name, subtotal_amount_vnd, vat_rate_percent, vat_amount_vnd, total_amount_vnd, signature_valid, tax_authority_code FROM erp_invoices WHERE direction = 'input' AND {where_sql} ORDER BY issue_date ASC, invoice_number ASC; """, params, ) input_rows = cur.fetchall() # Fetch all valid output invoices cur.execute( f""" SELECT id, invoice_number, invoice_series, template_code, issue_date, buyer_tax_code, buyer_name, subtotal_amount_vnd, vat_rate_percent, vat_amount_vnd, total_amount_vnd, signature_valid, tax_authority_code FROM erp_invoices WHERE direction = 'output' AND {where_sql} ORDER BY issue_date ASC, invoice_number ASC; """, params, ) output_rows = cur.fetchall() def _format_inv_list(rows, is_input=True): res = [] for idx, r in enumerate(rows, 1): d = dict(r) d["stt"] = idx d["id"] = str(d["id"]) d["issue_date"] = str(d["issue_date"]) d["subtotal_amount_vnd"] = float(d["subtotal_amount_vnd"]) d["vat_rate_percent"] = ( float(d["vat_rate_percent"]) if d["vat_rate_percent"] is not None else 8.0 ) d["vat_amount_vnd"] = float(d["vat_amount_vnd"]) d["total_amount_vnd"] = float(d["total_amount_vnd"]) res.append(d) return res formatted_inputs = _format_inv_list(input_rows, is_input=True) formatted_outputs = _format_inv_list(output_rows, is_input=False) # Group by Tax Rate brackets: 0%, 5%, 8%, 10%, kct def _group_by_tax_rate(items): groups = { "kct": { "subtotal_vnd": 0.0, "vat_vnd": 0.0, "count": 0, "label": "HHDV Không chịu thuế", }, "rate_0": { "subtotal_vnd": 0.0, "vat_vnd": 0.0, "count": 0, "label": "Thuế suất 0%", }, "rate_5": { "subtotal_vnd": 0.0, "vat_vnd": 0.0, "count": 0, "label": "Thuế suất 5%", }, "rate_8": { "subtotal_vnd": 0.0, "vat_vnd": 0.0, "count": 0, "label": "Thuế suất 8%", }, "rate_10": { "subtotal_vnd": 0.0, "vat_vnd": 0.0, "count": 0, "label": "Thuế suất 10%", }, "other": { "subtotal_vnd": 0.0, "vat_vnd": 0.0, "count": 0, "label": "Khác", }, } for itm in items: rate = itm.get("vat_rate_percent", 8.0) vat_amt = itm.get("vat_amount_vnd", 0.0) sub_amt = itm.get("subtotal_amount_vnd", 0.0) if vat_amt == 0.0 and rate == 0.0: key = "kct" elif abs(rate - 0.0) < 0.1: key = "rate_0" elif abs(rate - 5.0) < 0.1: key = "rate_5" elif abs(rate - 8.0) < 0.1: key = "rate_8" elif abs(rate - 10.0) < 0.1: key = "rate_10" else: key = "other" groups[key]["subtotal_vnd"] += sub_amt groups[key]["vat_vnd"] += vat_amt groups[key]["count"] += 1 return groups input_groups = _group_by_tax_rate(formatted_inputs) output_groups = _group_by_tax_rate(formatted_outputs) total_input_subtotal = sum( i["subtotal_amount_vnd"] for i in formatted_inputs ) total_input_vat = sum(i["vat_amount_vnd"] for i in formatted_inputs) total_output_subtotal = sum( o["subtotal_amount_vnd"] for o in formatted_outputs ) total_output_vat = sum(o["vat_amount_vnd"] for o in formatted_outputs) net_vat_balance = total_input_vat - total_output_vat vat_payable = ( (total_output_vat - total_input_vat) if total_output_vat > total_input_vat else 0.0 ) vat_deductible = ( (total_input_vat - total_output_vat) if total_input_vat >= total_output_vat else 0.0 ) return { "from_date": from_date, "to_date": to_date, "vat_period": vat_period or "Kỳ lựa chọn", "input_schedule": formatted_inputs, "output_schedule": formatted_outputs, "tax_breakdown": {"input": input_groups, "output": output_groups}, "kpis": { "total_input_invoices": len(formatted_inputs), "total_input_subtotal_vnd": total_input_subtotal, "total_input_vat_vnd": total_input_vat, "total_output_invoices": len(formatted_outputs), "total_output_subtotal_vnd": total_output_subtotal, "total_output_vat_vnd": total_output_vat, "net_vat_balance_vnd": net_vat_balance, "vat_payable_vnd": vat_payable, "vat_deductible_carried_forward_vnd": vat_deductible, "vat_status_label": "Còn được khấu trừ chuyển kỳ sau" if net_vat_balance >= 0 else "Thuế GTGT phải nộp vào ngân sách Nhà nước", }, } def export_vat_schedule_csv( self, from_date: str | None = None, to_date: str | None = None, direction: str = "all", ) -> str: """Xuất bảng kê thuế GTGT dạng CSV (UTF-8 BOM) để kế toán mở trực tiếp trên Excel.""" schedule = self.get_vat_tax_schedule(from_date=from_date, to_date=to_date) lines: list[str] = ["\ufeff"] # UTF-8 BOM from_label = from_date if from_date else "Đầu kỳ" to_label = to_date if to_date else "Hiện tại" lines.append('"BẢNG KÊ THUẾ GIÁ TRỊ GIA TĂNG (DSCons ERP)"') lines.append(f'"Kỳ kê khai: Từ {from_label} đến {to_label}"') lines.append("") if direction in ("all", "input"): lines.append( '"--- BẢNG KÊ HÓA ĐƠN, CHỨNG TỪ HÀNG HÓA DỊCH VỤ MUA VÀO (MẪU 01-1/GTGT) ---"' ) lines.append( '"STT","Ký hiệu mẫu","Ký hiệu HĐ","Số HĐ","Ngày lập","Tên người bán","MST người bán","Doanh số mua chưa thuế (VNĐ)","Thuế suất (%)","Tiền thuế GTGT (VNĐ)","Tổng thanh toán (VNĐ)","Mã CQT"' ) for itm in schedule["input_schedule"]: seller_clean = str(itm.get("seller_name", "")).replace('"', '""') lines.append( f'"{itm["stt"]}","{itm.get("template_code", "1")}","{itm.get("invoice_series", "")}","{itm.get("invoice_number", "")}","{itm.get("issue_date", "")}","{seller_clean}","{itm.get("seller_tax_code", "")}","{itm.get("subtotal_amount_vnd", 0):.0f}","{itm.get("vat_rate_percent", 8):.1f}%","{itm.get("vat_amount_vnd", 0):.0f}","{itm.get("total_amount_vnd", 0):.0f}","{itm.get("tax_authority_code", "")}"' ) tot_inp_sub = schedule["kpis"]["total_input_subtotal_vnd"] tot_inp_vat = schedule["kpis"]["total_input_vat_vnd"] lines.append( f'"TỔNG CỘNG ĐẦU VÀO","","","","","","","{tot_inp_sub:.0f}","","{tot_inp_vat:.0f}","{tot_inp_sub + tot_inp_vat:.0f}",""' ) lines.append("") if direction in ("all", "output"): lines.append( '"--- BẢNG KÊ HÓA ĐƠN, CHỨNG TỪ HÀNG HÓA DỊCH VỤ BÁN RA (MẪU 01-2/GTGT) ---"' ) lines.append( '"STT","Ký hiệu mẫu","Ký hiệu HĐ","Số HĐ","Ngày lập","Tên người mua","MST người mua","Doanh thu chưa thuế (VNĐ)","Thuế suất (%)","Tiền thuế GTGT (VNĐ)","Tổng thanh toán (VNĐ)","Mã CQT"' ) for itm in schedule["output_schedule"]: buyer_clean = str(itm.get("buyer_name", "")).replace('"', '""') lines.append( f'"{itm["stt"]}","{itm.get("template_code", "1")}","{itm.get("invoice_series", "")}","{itm.get("invoice_number", "")}","{itm.get("issue_date", "")}","{buyer_clean}","{itm.get("buyer_tax_code", "")}","{itm.get("subtotal_amount_vnd", 0):.0f}","{itm.get("vat_rate_percent", 8):.1f}%","{itm.get("vat_amount_vnd", 0):.0f}","{itm.get("total_amount_vnd", 0):.0f}","{itm.get("tax_authority_code", "")}"' ) tot_out_sub = schedule["kpis"]["total_output_subtotal_vnd"] tot_out_vat = schedule["kpis"]["total_output_vat_vnd"] lines.append( f'"TỔNG CỘNG ĐẦU RA","","","","","","","{tot_out_sub:.0f}","","{tot_out_vat:.0f}","{tot_out_sub + tot_out_vat:.0f}",""' ) lines.append("") lines.append('"--- TỔNG HỢP CÂN ĐỐI NGHĨA VỤ THUẾ GTGT ---"') lines.append( f'"Tổng thuế GTGT đầu vào được khấu trừ","{schedule["kpis"]["total_input_vat_vnd"]:.0f} VNĐ"' ) lines.append( f'"Tổng thuế GTGT đầu ra phát sinh","{schedule["kpis"]["total_output_vat_vnd"]:.0f} VNĐ"' ) if schedule["kpis"]["vat_payable_vnd"] > 0: lines.append( f'"Số thuế GTGT phải nộp vào NSNN","{schedule["kpis"]["vat_payable_vnd"]:.0f} VNĐ"' ) else: lines.append( f'"Số thuế GTGT còn được khấu trừ chuyển kỳ sau","{schedule["kpis"]["vat_deductible_carried_forward_vnd"]:.0f} VNĐ"' ) return "\n".join(lines)