from __future__ import annotations import json import re from datetime import date, datetime from decimal import Decimal from enum import Enum from io import BytesIO from typing import Any import httpx from openpyxl import Workbook from openpyxl.styles import Font, PatternFill, Alignment, Border, Side from openpyxl.utils import get_column_letter from app.application.services.chart_of_account_service import ChartOfAccountService from app.application.services.inventory_service import InventoryService from app.application.services.purchase_service import PurchaseService from app.application.services.report_service import ReportService from app.application.services.sales_service import SalesService from app.application.services.users_roles_service import UsersRolesService from app.application.services.voucher_service import VoucherService from app.application.exceptions import ValidationError from app.core.config import settings from app.domain.entities.user_registration import UserRegistration from app.domain.enums import StockReportStatus, VoucherStatus DATASETS = ( "trial_balance", "income_statement", "balance_sheet", "ledger", "stock_by_warehouse", "items", "stock", "vendors", "customers", "vouchers", "accounts", "warehouses", "departments", "department_issues", "purchase_orders", "sales_orders", "users", ) EXAMPLES = [ "Export trial balance to Excel", "Download income statement / P&L", "Export balance sheet", "Export general ledger", "Total items warehouse wise", "Export items to Excel", "Show current stock", "List all vendors with phone and city", "Show posted vouchers", ] _KEYWORD_MAP = ( ("stock_by_warehouse", ( "warehouse wise", "warehouse-wise", "warehousewise", "by warehouse", "per warehouse", "items warehouse", "item warehouse", "warehouse items", "warehouse item", "stock by warehouse", "stock warehouse", "total items warehouse", "items by warehouse", )), ("trial_balance", ("trial balance", "trial-balance", "financial report", "financial reports")), ("income_statement", ("income statement", "profit and loss", "profit & loss", "p&l", "pnl", "p and l")), ("balance_sheet", ("balance sheet", "statement of financial position")), ("ledger", ("general ledger", "consolidated ledger", "consolidated gl", " ledger ")), ("department_issues", ("department issue", "issue to department", "pending approval", "issues list")), ("purchase_orders", ("purchase order", "purchase orders", " po ", "vendor po")), ("sales_orders", ("sales order", "sales orders", "customer order")), ("vouchers", ("voucher", "journal", "posted voucher", "ledger entry")), ("accounts", ("chart of account", "chart of accounts", "coa", "account balance")), ("vendors", ("vendor", "supplier")), ("customers", ("customer", "client")), ("warehouses", ("warehouse", "godown")), ("departments", ("department list", "departments")), ("users", ("user list", "users", "employees", "staff")), ("stock", ("stock", "inventory balance", "low stock", "out of stock", "reorder")), ("items", ("item list", "items", "sku", "product")), ) class AiAgentService: def __init__( self, inventory: InventoryService, purchase: PurchaseService, sales: SalesService, vouchers: VoucherService, accounts: ChartOfAccountService, users: UsersRolesService, reports: ReportService | None = None, ) -> None: self._inventory = inventory self._purchase = purchase self._sales = sales self._vouchers = vouchers self._accounts = accounts self._users = users self._reports = reports async def query( self, user: UserRegistration, company_id: str, query: str ) -> dict: query = (query or "").strip() if len(query) < 2: raise ValidationError('Enter a question such as "export low stock items".') plan = await self._plan(query) dataset = plan["dataset"] title = plan["title"] filters = dict(plan.get("filters") or {}) columns, rows, total = await self._fetch(user, company_id, dataset, filters) note = None if dataset == "stock" and total == 0: all_columns, all_rows, all_total = await self._fetch(user, company_id, "stock", {}) wanted = str(filters.get("stock_status") or "") restock = [ row for row in all_rows if str(row.get("stock_status")) in ("low_stock", "out_of_stock") ] if wanted == "low_stock" and restock: columns, rows, total = all_columns, restock, len(restock) title = "Items needing restock" note = ( "No items are at reorder level. Showing out-of-stock and low-stock rows." ) elif all_total > 0: columns, rows, total = all_columns, all_rows, all_total title = "Stock" note = ( f"No rows matched “{self._filter_label(filters) if filters else 'low stock'}”. " "Showing all current stock instead." ) else: columns, rows, total = await self._fetch(user, company_id, "items", {}) if total > 0: dataset = "items" title = "Items" note = ( "No warehouse stock balances yet. Showing the item master instead. " "Stock rows appear after goods receipts or opening stock." ) elif total == 0 and filters: columns, rows, total = await self._fetch(user, company_id, dataset, {}) if total > 0: title = self._dataset_title(dataset) note = ( f"No rows matched “{self._filter_label(filters)}”. " f"Showing all {title.lower()} instead." ) summary = ( f"Found {total} {title.lower()} record(s) for “{query.strip()}”. " "Download Excel to open the full sheet in spreadsheet software." ) if note: summary = ( f"{note} Found {total} {title.lower()} record(s). " "Download Excel to open the sheet in spreadsheet software." ) elif total > len(rows): summary = ( f"Showing {len(rows)} of {total} {title.lower()} records. " "Excel includes the same preview set." ) return { "title": title, "summary": summary, "dataset": dataset, "llm_used": plan.get("llm_used", False), "columns": columns, "rows": rows, "row_count": total, "filename": self._filename(title), "examples": list(EXAMPLES), } async def export_excel( self, user: UserRegistration, company_id: str, query: str ) -> tuple[bytes, str]: result = await self.query(user, company_id, query) return self._to_xlsx(result), result["filename"] async def _plan(self, query: str) -> dict: llm_plan = await self._plan_with_llm(query) if llm_plan: return llm_plan return self._plan_with_keywords(query) def _plan_with_keywords(self, query: str) -> dict: text = f" {query.lower()} " dataset = "items" for name, needles in _KEYWORD_MAP: if any(needle in text for needle in needles): dataset = name break filters: dict[str, Any] = {} self._apply_date_filters(text, filters) if dataset == "stock": if "out of stock" in text or "zero stock" in text: filters["stock_status"] = StockReportStatus.OUT_OF_STOCK.value elif "low stock" in text or "reorder" in text: filters["stock_status"] = StockReportStatus.LOW_STOCK.value if dataset == "vouchers" and "posted" in text: filters["status"] = VoucherStatus.POSTED.value if dataset == "department_issues" and "pending" in text: filters["status"] = "review" title = self._dataset_title(dataset, filters) return { "dataset": dataset, "title": title, "filters": filters, "llm_used": False, } def _dataset_title(self, dataset: str, filters: dict | None = None) -> str: titles = { "items": "Items", "stock": "Stock", "vendors": "Vendors", "customers": "Customers", "vouchers": "Vouchers", "accounts": "Chart of Accounts", "warehouses": "Warehouses", "departments": "Departments", "department_issues": "Department Issues", "purchase_orders": "Purchase Orders", "sales_orders": "Sales Orders", "users": "Users", "trial_balance": "Trial Balance", "income_statement": "Income Statement", "balance_sheet": "Balance Sheet", "ledger": "General Ledger", "stock_by_warehouse": "Items by Warehouse", } filters = filters or {} if dataset == "stock" and filters.get("stock_status") == "low_stock": return "Low Stock Items" if dataset == "stock" and filters.get("stock_status") == "out_of_stock": return "Out of Stock Items" return titles.get(dataset, dataset.replace("_", " ").title()) def _filter_label(self, filters: dict) -> str: status = str(filters.get("stock_status") or filters.get("status") or "").replace("_", " ") return status or "the requested filter" def _apply_date_filters(self, text: str, filters: dict) -> None: dates = re.findall(r"\d{4}-\d{2}-\d{2}", text) if len(dates) >= 2: filters["from_date"] = dates[0] filters["to_date"] = dates[-1] filters["as_of"] = dates[-1] elif len(dates) == 1: filters["as_of"] = dates[0] filters["to_date"] = dates[0] today = date.today() if "this month" in text or "current month" in text: filters.setdefault("from_date", today.replace(day=1).isoformat()) filters.setdefault("to_date", today.isoformat()) elif "this year" in text or "current year" in text or " fy " in text: filters.setdefault("from_date", date(today.year, 1, 1).isoformat()) filters.setdefault("to_date", today.isoformat()) def _parse_iso_date(self, value: Any) -> date | None: if isinstance(value, date) and not isinstance(value, datetime): return value raw = str(value or "").strip()[:10] if not raw: return None try: return date.fromisoformat(raw) except ValueError: return None def _report_period(self, filters: dict) -> tuple[date, date, date]: today = date.today() as_of = self._parse_iso_date(filters.get("as_of") or filters.get("to_date")) or today from_date = self._parse_iso_date(filters.get("from_date")) or date(today.year, 1, 1) to_date = self._parse_iso_date(filters.get("to_date")) or today if from_date > to_date: from_date, to_date = to_date, from_date return from_date, to_date, as_of async def _fetch_stock_by_warehouse( self, user: UserRegistration, company_id: str ) -> tuple[list[dict], list[dict], int]: report = await self._inventory.get_stock_report( user, company_id, page=1, page_size=500 ) columns = [ {"key": "warehouse_code", "label": "Warehouse Code"}, {"key": "warehouse_name", "label": "Warehouse"}, {"key": "total_items", "label": "Total Items"}, {"key": "total_quantity", "label": "Total Quantity"}, {"key": "total_value", "label": "Stock Value"}, {"key": "percent_of_total", "label": "% of Value"}, ] rows: list[dict] = [] slices = list(report.by_warehouse_value or []) if slices: for item in slices: rows.append( { "warehouse_code": self._cell(item.code), "warehouse_name": self._cell(item.name), "total_items": int(item.issues_count or 0), "total_quantity": self._cell(item.total_quantity), "total_value": self._cell(item.total_value), "percent_of_total": self._cell(item.percent_of_total), } ) else: grouped: dict[str, dict[str, Any]] = {} for row in report.rows or []: key = row.warehouse_id or row.warehouse_name or "unknown" bucket = grouped.setdefault( key, { "warehouse_code": row.warehouse_code or "", "warehouse_name": row.warehouse_name or "Warehouse", "total_items": 0, "total_quantity": 0.0, "total_value": 0.0, }, ) bucket["total_items"] += 1 bucket["total_quantity"] += float(row.quantity_on_hand or 0) bucket["total_value"] += float(row.stock_value or 0) value_sum = sum(item["total_value"] for item in grouped.values()) or 1 for item in grouped.values(): rows.append( { "warehouse_code": item["warehouse_code"], "warehouse_name": item["warehouse_name"], "total_items": item["total_items"], "total_quantity": round(item["total_quantity"], 4), "total_value": round(item["total_value"], 2), "percent_of_total": round(item["total_value"] / value_sum * 100, 2), } ) if rows: rows.append( { "warehouse_code": "", "warehouse_name": "Total", "total_items": sum(int(row["total_items"] or 0) for row in rows), "total_quantity": round(sum(float(row["total_quantity"] or 0) for row in rows), 4), "total_value": round(sum(float(row["total_value"] or 0) for row in rows), 2), "percent_of_total": 100, } ) return columns, rows, max(len(rows) - (1 if rows else 0), 0) async def _fetch_financial( self, user: UserRegistration, company_id: str, dataset: str, filters: dict, ) -> tuple[list[dict], list[dict], int]: if self._reports is None: return [], [], 0 from_date, to_date, as_of = self._report_period(filters) if dataset == "trial_balance": report = await self._reports.generate_trial_balance( user, company_id, as_of, save=False ) columns = [ {"key": "account_code", "label": "Code"}, {"key": "account_name", "label": "Account"}, {"key": "account_type", "label": "Type"}, {"key": "debit", "label": "Debit"}, {"key": "credit", "label": "Credit"}, {"key": "balance", "label": "Balance"}, ] rows = [ self._row(item, ["account_code", "account_name", "account_type", "debit", "credit", "balance"]) for item in report.line_items ] totals = report.totals or {} if report.line_items: rows.append( { "account_code": "", "account_name": "Total", "account_type": "", "debit": self._cell(totals.get("total_debit")), "credit": self._cell(totals.get("total_credit")), "balance": self._cell(totals.get("difference")), } ) return columns, rows, len(report.line_items) if dataset == "income_statement": report = await self._reports.generate_income_statement( user, company_id, from_date, to_date, save=False ) columns = [ {"key": "account_code", "label": "Code"}, {"key": "account_name", "label": "Account"}, {"key": "account_type", "label": "Type"}, {"key": "balance", "label": "Amount"}, ] rows = [ self._row(item, ["account_code", "account_name", "account_type", "balance"]) for item in report.line_items ] totals = report.totals or {} if report.line_items: rows.extend( [ { "account_code": "", "account_name": "Total Revenue", "account_type": "revenue", "balance": self._cell(totals.get("total_revenue")), }, { "account_code": "", "account_name": "Total Expense", "account_type": "expense", "balance": self._cell(totals.get("total_expense")), }, { "account_code": "", "account_name": "Net Income", "account_type": "", "balance": self._cell(totals.get("net_income")), }, ] ) return columns, rows, len(report.line_items) if dataset == "balance_sheet": report = await self._reports.generate_balance_sheet( user, company_id, as_of, save=False ) columns = [ {"key": "account_code", "label": "Code"}, {"key": "account_name", "label": "Account"}, {"key": "account_type", "label": "Type"}, {"key": "balance", "label": "Balance"}, ] rows = [ self._row(item, ["account_code", "account_name", "account_type", "balance"]) for item in report.line_items ] totals = report.totals or {} if report.line_items: rows.extend( [ { "account_code": "", "account_name": "Total Assets", "account_type": "asset", "balance": self._cell(totals.get("total_assets")), }, { "account_code": "", "account_name": "Total Liabilities", "account_type": "liability", "balance": self._cell(totals.get("total_liabilities")), }, { "account_code": "", "account_name": "Total Equity", "account_type": "equity", "balance": self._cell(totals.get("total_equity")), }, ] ) return columns, rows, len(report.line_items) accounts, _ = await self._accounts.list_accounts( user, company_id, is_active=True, skip=0, limit=200 ) columns = [ {"key": "voucher_date", "label": "Date"}, {"key": "voucher_number", "label": "Voucher"}, {"key": "account_code", "label": "Code"}, {"key": "account_name", "label": "Account"}, {"key": "description", "label": "Description"}, {"key": "debit", "label": "Debit"}, {"key": "credit", "label": "Credit"}, {"key": "balance", "label": "Balance"}, ] rows: list[dict] = [] for account in accounts: if getattr(account, "is_group", False) or not account.id: continue report = await self._reports.generate_ledger( user, company_id, account.id, from_date, to_date, save=False, ) for item in report.line_items: meta = item.metadata or {} rows.append( { "voucher_date": self._cell(meta.get("voucher_date")), "voucher_number": self._cell(meta.get("voucher_number")), "account_code": self._cell(item.account_code), "account_name": self._cell(item.account_name), "description": self._cell(meta.get("description")), "debit": self._cell(item.debit), "credit": self._cell(item.credit), "balance": self._cell(item.balance), } ) if len(rows) >= 500: return columns, rows, len(rows) return columns, rows, len(rows) async def _plan_with_llm(self, query: str) -> dict | None: if not (settings.LLM_API_KEY or "").strip(): return None prompt = { "model": settings.LLM_MODEL, "temperature": 0, "messages": [ { "role": "system", "content": ( "You map accounting app questions to a dataset export. " "Reply with JSON only: " '{"dataset":"...","title":"...","filters":{}}. ' f"dataset must be one of: {', '.join(DATASETS)}. " "Optional filters: stock_status (in_stock|low_stock|out_of_stock), " "voucher status (draft|posted|cancelled), " "department issue status (draft|review|approved|issue_items|completed), " "as_of, from_date, to_date as YYYY-MM-DD. " "Use trial_balance, income_statement, balance_sheet, or ledger " "for financial reports. " "Use stock_by_warehouse for totals, item counts, or stock grouped " "by warehouse (warehouse wise)." ), }, {"role": "user", "content": query}, ], } try: async with httpx.AsyncClient(timeout=20.0) as client: response = await client.post( f"{settings.LLM_BASE_URL.rstrip('/')}/chat/completions", headers={ "Authorization": f"Bearer {settings.LLM_API_KEY}", "Content-Type": "application/json", }, json=prompt, ) response.raise_for_status() content = response.json()["choices"][0]["message"]["content"] match = re.search(r"\{[\s\S]*\}", content or "") if not match: return None data = json.loads(match.group(0)) dataset = str(data.get("dataset") or "").strip() if dataset not in DATASETS: return None title = str(data.get("title") or dataset.replace("_", " ").title())[:80] filters = data.get("filters") if isinstance(data.get("filters"), dict) else {} return { "dataset": dataset, "title": title, "filters": filters, "llm_used": True, } except Exception: return None async def _fetch( self, user: UserRegistration, company_id: str, dataset: str, filters: dict, ) -> tuple[list[dict], list[dict], int]: limit = 500 if dataset == "items": items, total = await self._inventory.list_items(user, company_id, skip=0, limit=limit) columns = [ {"key": "sku", "label": "SKU"}, {"key": "name", "label": "Item Name"}, {"key": "category_name", "label": "Category"}, {"key": "warehouse_name", "label": "Warehouse"}, {"key": "purchase_price", "label": "Purchase Price"}, {"key": "sale_price", "label": "Sale Price"}, {"key": "reorder_level", "label": "Reorder Level"}, {"key": "is_active", "label": "Active"}, ] rows = [ self._row( item, [ "sku", "name", "category_name", "warehouse_name", "purchase_price", "sale_price", "reorder_level", "is_active", ], ) for item in items ] return columns, rows, total if dataset == "stock": status = None raw_status = filters.get("stock_status") if raw_status: try: status = StockReportStatus(raw_status) except ValueError: status = None report = await self._inventory.get_stock_report( user, company_id, stock_status=status, page=1, page_size=limit, ) columns = [ {"key": "item_sku", "label": "SKU"}, {"key": "item_name", "label": "Item"}, {"key": "warehouse_name", "label": "Warehouse"}, {"key": "quantity_on_hand", "label": "On Hand"}, {"key": "available_qty", "label": "Available"}, {"key": "reorder_level", "label": "Reorder Level"}, {"key": "unit_cost", "label": "Unit Cost"}, {"key": "stock_value", "label": "Stock Value"}, {"key": "stock_status", "label": "Status"}, ] rows = [ self._row( row, [ "item_sku", "item_name", "warehouse_name", "quantity_on_hand", "available_qty", "reorder_level", "unit_cost", "stock_value", "stock_status", ], ) for row in report.rows ] return columns, rows, report.rows_total if dataset == "stock_by_warehouse": return await self._fetch_stock_by_warehouse(user, company_id) if dataset == "vendors": items, total = await self._purchase.list_vendors(user, company_id, skip=0, limit=limit) columns = [ {"key": "code", "label": "Code"}, {"key": "name", "label": "Vendor"}, {"key": "phone", "label": "Phone"}, {"key": "email", "label": "Email"}, {"key": "city", "label": "City"}, {"key": "credit_limit", "label": "Credit Limit"}, {"key": "is_active", "label": "Active"}, ] rows = [ self._row(item, ["code", "name", "phone", "email", "city", "credit_limit", "is_active"]) for item in items ] return columns, rows, total if dataset == "customers": items, total = await self._sales.list_customers(user, company_id, skip=0, limit=limit) columns = [ {"key": "code", "label": "Code"}, {"key": "name", "label": "Customer"}, {"key": "phone", "label": "Phone"}, {"key": "email", "label": "Email"}, {"key": "city", "label": "City"}, {"key": "credit_limit", "label": "Credit Limit"}, {"key": "is_active", "label": "Active"}, ] rows = [ self._row(item, ["code", "name", "phone", "email", "city", "credit_limit", "is_active"]) for item in items ] return columns, rows, total if dataset == "vouchers": status = None if filters.get("status"): try: status = VoucherStatus(filters["status"]) except ValueError: status = None items, total = await self._vouchers.list_vouchers( user, company_id, status=status, skip=0, limit=limit ) columns = [ {"key": "voucher_number", "label": "Voucher No."}, {"key": "voucher_type", "label": "Type"}, {"key": "voucher_date", "label": "Date"}, {"key": "narration", "label": "Narration"}, {"key": "status", "label": "Status"}, {"key": "total_debit", "label": "Debit"}, {"key": "total_credit", "label": "Credit"}, ] rows = [ self._row( item, [ "voucher_number", "voucher_type", "voucher_date", "narration", "status", "total_debit", "total_credit", ], ) for item in items ] return columns, rows, total if dataset == "accounts": items, total = await self._accounts.list_accounts(user, company_id, skip=0, limit=limit) columns = [ {"key": "code", "label": "Code"}, {"key": "name", "label": "Account"}, {"key": "account_type", "label": "Type"}, {"key": "nature", "label": "Nature"}, {"key": "opening_balance", "label": "Opening"}, {"key": "current_balance", "label": "Current Balance"}, {"key": "is_active", "label": "Active"}, ] rows = [ self._row( item, [ "code", "name", "account_type", "nature", "opening_balance", "current_balance", "is_active", ], ) for item in items ] return columns, rows, total if dataset == "warehouses": items, total = await self._inventory.list_warehouses(user, company_id, skip=0, limit=limit) columns = [ {"key": "code", "label": "Code"}, {"key": "name", "label": "Warehouse"}, {"key": "warehouse_type", "label": "Type"}, {"key": "status", "label": "Status"}, {"key": "city", "label": "City"}, {"key": "manager_name", "label": "Manager"}, {"key": "phone", "label": "Phone"}, ] rows = [ self._row( item, ["code", "name", "warehouse_type", "status", "city", "manager_name", "phone"], ) for item in items ] return columns, rows, total if dataset == "departments": items, total = await self._inventory.list_departments(user, company_id, skip=0, limit=limit) columns = [ {"key": "code", "label": "Code"}, {"key": "name", "label": "Department"}, {"key": "head_name", "label": "Head"}, {"key": "location_name", "label": "Location"}, {"key": "status", "label": "Status"}, ] rows = [ self._row(item, ["code", "name", "head_name", "location_name", "status"]) for item in items ] return columns, rows, total if dataset == "department_issues": status = filters.get("status") items, total = await self._inventory.list_department_issues( user, company_id, status=status, skip=0, limit=limit ) columns = [ {"key": "issue_number", "label": "Issue No."}, {"key": "issue_date", "label": "Date"}, {"key": "department_name", "label": "Department"}, {"key": "from_warehouse_name", "label": "Warehouse"}, {"key": "status", "label": "Status"}, {"key": "total_items", "label": "Items"}, {"key": "total_quantity", "label": "Qty"}, {"key": "total_estimated_value", "label": "Value"}, ] rows = [ self._row( item, [ "issue_number", "issue_date", "department_name", "from_warehouse_name", "status", "total_items", "total_quantity", "total_estimated_value", ], ) for item in items ] return columns, rows, total if dataset == "purchase_orders": items, total = await self._purchase.list_purchase_orders( user, company_id, skip=0, limit=limit ) columns = [ {"key": "po_number", "label": "PO No."}, {"key": "order_date", "label": "Date"}, {"key": "vendor_name", "label": "Vendor"}, {"key": "warehouse_name", "label": "Warehouse"}, {"key": "status", "label": "Status"}, {"key": "total_amount", "label": "Total"}, ] rows = [ self._row( item, ["po_number", "order_date", "vendor_name", "warehouse_name", "status", "total_amount"], ) for item in items ] return columns, rows, total if dataset == "sales_orders": items, total = await self._sales.list_sales_orders(user, company_id, skip=0, limit=limit) columns = [ {"key": "so_number", "label": "SO No."}, {"key": "order_date", "label": "Date"}, {"key": "customer_name", "label": "Customer"}, {"key": "warehouse_name", "label": "Warehouse"}, {"key": "status", "label": "Status"}, {"key": "total_amount", "label": "Total"}, ] rows = [ self._row( item, ["so_number", "order_date", "customer_name", "warehouse_name", "status", "total_amount"], ) for item in items ] return columns, rows, total if dataset in {"trial_balance", "income_statement", "balance_sheet", "ledger"}: return await self._fetch_financial(user, company_id, dataset, filters) items, total = await self._users.list_users(user, company_id, page=1, page_size=limit) columns = [ {"key": "full_name", "label": "Name"}, {"key": "username", "label": "Username"}, {"key": "email", "label": "Email"}, {"key": "role_name", "label": "Role"}, {"key": "department_name", "label": "Department"}, {"key": "status", "label": "Status"}, ] rows = [ self._row(item, ["full_name", "username", "email", "role_name", "department_name", "status"]) for item in items ] return columns, rows, total def _row(self, item: Any, keys: list[str]) -> dict: data = item.model_dump() if hasattr(item, "model_dump") else dict(item) return {key: self._cell(data.get(key)) for key in keys} def _cell(self, value: Any) -> str | int | float | bool | None: if value is None: return "" if isinstance(value, bool): return "Yes" if value else "No" if isinstance(value, Enum): return value.value if isinstance(value, Decimal): return float(value) if isinstance(value, datetime): return value.strftime("%Y-%m-%d %H:%M") if isinstance(value, date): return value.isoformat() return value def _filename(self, title: str) -> str: slug = re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-") or "export" return f"{slug}-{date.today().isoformat()}.xlsx" def _to_xlsx(self, result: dict) -> bytes: workbook = Workbook() sheet = workbook.active sheet.title = (result.get("title") or "Export")[:31] header_font = Font(bold=True, color="FFFFFF") header_fill = PatternFill("solid", fgColor="6D28D9") header_align = Alignment(horizontal="left", vertical="center") border = Border( left=Side(style="thin", color="E2E8F0"), right=Side(style="thin", color="E2E8F0"), top=Side(style="thin", color="E2E8F0"), bottom=Side(style="thin", color="E2E8F0"), ) columns = result.get("columns") or [] for index, column in enumerate(columns, start=1): cell = sheet.cell(1, index, column.get("label") or column.get("key")) cell.font = header_font cell.fill = header_fill cell.alignment = header_align cell.border = border for row_index, row in enumerate(result.get("rows") or [], start=2): for col_index, column in enumerate(columns, start=1): cell = sheet.cell(row_index, col_index, row.get(column["key"], "")) cell.border = border for index, column in enumerate(columns, start=1): width = max(len(str(column.get("label") or "")), 12) for row in result.get("rows") or []: width = max(width, min(len(str(row.get(column["key"], ""))), 40)) sheet.column_dimensions[get_column_letter(index)].width = width + 2 sheet.auto_filter.ref = sheet.dimensions sheet.freeze_panes = "A2" buffer = BytesIO() workbook.save(buffer) return buffer.getvalue()