from __future__ import annotations """GDT invoice batch fetcher and connection execution.""" import json import logging import ssl import urllib.error import urllib.parse import urllib.request import uuid from datetime import date, datetime, timedelta from typing import Any from uuid import UUID from app.core.vn_time import format_vn_time logger = logging.getLogger(__name__) class InvoiceGdtFetcherMixin: """Mixin for communicating with GDT endpoints.""" def _execute_real_gdt_connection( self, config_id: UUID, tax_code: str, username: str, password: str, extra: dict[str, Any], from_date: str | None = None, to_date: str | None = None, sync_type: str = "all", ) -> dict[str, Any]: """Kết nối thực tế và tải hóa đơn từng phần từ Cổng HĐĐT Tổng Cục Thuế (vượt Captcha tự động bằng AI).""" # 1. Kiểm tra chế độ chờ an toàn 3 giờ (Cooldown) chống khóa tài khoản if hasattr(self, "check_auth_cooldown"): in_cooldown, cd_until = self.check_auth_cooldown("gdt_portal", tax_code) if in_cooldown: msg = ( f"Tài khoản Cổng Thuế (MST: {tax_code}) đang trong thời gian chờ an toàn 3 giờ " f"(đến {format_vn_time(cd_until)}) sau lần thử thất bại để chống bị khóa tài khoản. Đã tự động bỏ qua phiên kết nối." ) logger.warning("⏳ [GDT_AUTH] %s", msg) return { "service": "gdt_portal", "tax_code": tax_code, "status": "cooldown_active", "message": msg, "cooldown_until": cd_until, } auth_res = self._authenticate_gdt_portal( tax_code=tax_code, username=username, password=password, extra_config=extra ) if auth_res["status"] == "success": jwt_token = auth_res.get("token") or "" cookie_header = auth_res.get("cookie_header") or "" new_count, dup_count, details = self._fetch_and_persist_gdt_invoices( token=jwt_token, tax_code=tax_code, from_date=from_date, to_date=to_date, sync_type=sync_type, cookie_header=cookie_header, ) msg = f"Đã kết nối Tổng Cục Thuế và nạp {new_count} hóa đơn mới ({dup_count} đã tồn tại)." if details: msg += f" Chi tiết: {', '.join(details)}." if hasattr(self, "clear_auth_cooldown"): self.clear_auth_cooldown("gdt_portal", tax_code) self._update_sync_status(config_id, "SUCCESS", msg) return { "service": "gdt_portal", "tax_code": tax_code, "status": "success", "message": msg, "new_imported": new_count, "duplicates": dup_count, "from_date": from_date, "to_date": to_date, } else: err_msg = auth_res.get("message", "Lỗi xác thực Cổng Thuế.") cd_until = None is_real_auth_err = (auth_res.get("status") == "auth_error") and auth_res.get("is_auth_error", True) if is_real_auth_err and hasattr(self, "set_auth_cooldown"): cd_until = self.set_auth_cooldown("gdt_portal", tax_code, hours=3.0) err_msg += f" (Đã tự động kích hoạt chế độ chờ an toàn 3 giờ đến {format_vn_time(cd_until)} để chống khóa tài khoản)." self._update_sync_status(config_id, "AUTH_ERROR" if is_real_auth_err else "AI_ERROR", err_msg) return { "service": "gdt_portal", "tax_code": tax_code, "status": auth_res.get("status", "auth_error"), "message": err_msg, "cooldown_until": cd_until, } def _fetch_and_persist_gdt_invoices( self, token: str, tax_code: str, from_date: str | None = None, to_date: str | None = None, sync_type: str = "all", cookie_header: str = "", ) -> tuple[int, int, list[str]]: """Truy vấn danh sách hóa đơn từ Cổng Thuế theo khoảng thời gian và lưu trữ vào PostgreSQL. Quy định Cổng Thuế: Khoảng thời gian mỗi lần truy vấn không được vượt quá 1 tháng. Hàm tự động phân rã (chunking) khoảng ngày bất kỳ thành các kỳ tháng liên tiếp và hỗ trợ phân trang con trỏ (state cursor pagination). """ import calendar today = date.today() if not to_date: to_d = today else: try: to_d = datetime.strptime(to_date, "%Y-%m-%d").date() except ValueError: to_d = today if not from_date: from_d = to_d - timedelta(days=30) else: try: from_d = datetime.strptime(from_date, "%Y-%m-%d").date() except ValueError: from_d = to_d - timedelta(days=30) if from_d > to_d: from_d, to_d = to_d, from_d # 1. Phân rã khoảng thời gian thành các kỳ tháng liên tiếp (không quá 1 tháng / kỳ) chunks: list[tuple[date, date]] = [] curr = from_d while curr <= to_d: _, last_day = calendar.monthrange(curr.year, curr.month) curr_month_end = date(curr.year, curr.month, last_day) chunk_end = min(curr_month_end, to_d) chunks.append((curr, chunk_end)) if curr.month == 12: curr = date(curr.year + 1, 1, 1) else: curr = date(curr.year, curr.month + 1, 1) ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE total_new = 0 total_dup = 0 details = [] directions_to_sync = [] if sync_type in ("all", "purchase"): directions_to_sync.append(("input", "purchase")) if sync_type in ("all", "sold"): directions_to_sync.append(("output", "sold")) for s_d, e_d in chunks: s_str = s_d.strftime("%d/%m/%Y") e_str = e_d.strftime("%d/%m/%Y") search_param = f"tdlap=ge={s_str}T00:00:00;tdlap=le={e_str}T23:59:59" for direction, path_type in directions_to_sync: state_cursor = None page_count = 0 max_pages = 25 # Hỗ trợ tối đa 1250 hóa đơn / tháng action_label = ( "Tìm kiếm (hóa đơn mua vào)" if path_type == "purchase" else "Tìm kiếm (hóa đơn bán ra)" ) headers = { "Authorization": f"Bearer {token}" if token else "", "Cookie": cookie_header or "", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36", "Accept": "application/json, text/plain, */*", "Referer": "https://hoadondientu.gdt.gov.vn/tra-cuu/tra-cuu-hoa-don", "Origin": "https://hoadondientu.gdt.gov.vn", "request-id": str(uuid.uuid4()), "end-point": "/tra-cuu/tra-cuu-hoa-don", "action": urllib.parse.quote(action_label), } while page_count < max_pages: page_count += 1 url = f"https://hoadondientu.gdt.gov.vn/api/query/invoices/{path_type}?sort=tdlap:desc&size=50&search={search_param}" if state_cursor: url += f"&state={urllib.parse.quote(state_cursor)}" headers["request-id"] = str(uuid.uuid4()) try: req = urllib.request.Request(url, headers=headers) with urllib.request.urlopen( req, timeout=20, context=ctx ) as resp: data = json.loads(resp.read().decode("utf-8")) invoices_list = ( data.get("datas") or data.get("items") or (data if isinstance(data, list) else []) ) chunk_new = 0 chunk_dup = 0 for raw_inv in invoices_list: if isinstance(raw_inv, dict): res = self._persist_gdt_raw_invoice( raw_inv, direction, tax_code ) if res == "new": chunk_new += 1 total_new += 1 elif res == "dup": chunk_dup += 1 total_dup += 1 state_cursor = data.get("state") if not state_cursor or len(invoices_list) == 0: break except Exception as e: logger.warning( f"Error fetching GDT invoices ({path_type}) for {s_str}-{e_str}: {e}" ) details.append(f"Lỗi {s_str}-{e_str} ({str(e)[:30]})") break logger.info( f"Synced chunk {s_str} to {e_str}: new={total_new}, dup={total_dup}" ) dir_label = ( "Mua vào & Bán ra" if sync_type == "all" else ("Mua vào" if sync_type == "purchase" else "Bán ra") ) details.append(f"{dir_label} ({len(chunks)} kỳ tháng)") return total_new, total_dup, details