from datetime import date from decimal import Decimal from math import ceil from sqlalchemy import Numeric, String, and_, case, cast, collate, func, literal, or_, select, true, union_all from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import aliased, selectinload from app.domain.entities.report import ( CustomReport, CustomReportConfig, CustomReportResultRow, CustomReportTotals, DeptReportComboPoint, DeptReportDetailRow, DeptReportIssuedReceived, DeptReportRow, FinancialReport, ReportLineItem, TxnReportRow, TxnReportSlice, TxnReportTrendPoint, ) from app.domain.enums import ( CustomReportField, CustomReportModule, ReportType, TxnReportPaymentStatus, TxnReportReferenceType, TxnReportTrendGranularity, TxnReportType, ) from app.domain.repositories.report_repository import ReportRepository from app.infrastructure.db.models import ( CustomReportModel, CustomerModel, DepartmentIssueModel, DepartmentModel, FinancialReportModel, ItemTransactionModel, PurchaseOrderModel, PurchasePaymentModel, ReportLineItemModel, SalesInvoiceModel, SalesOrderModel, SalesPaymentModel, StockTransferModel, UserRegistrationModel, VendorBillModel, VendorModel, WarehouseModel, new_id, ) from app.infrastructure.repositories.mysql_utils import to_decimal class MySQLReportRepository(ReportRepository): def __init__(self, session: AsyncSession) -> None: self._session = session def _to_entity(self, row: FinancialReportModel) -> FinancialReport: totals_raw = row.totals or {} totals = {key: to_decimal(value) for key, value in totals_raw.items()} line_items = [ ReportLineItem( account_id=item.account_id, account_code=item.account_code, account_name=item.account_name, account_type=item.account_type, debit=to_decimal(item.debit), credit=to_decimal(item.credit), balance=to_decimal(item.balance), metadata=item.metadata_json or {}, ) for item in (row.line_items or []) ] return FinancialReport( id=row.id, company_id=row.company_id, report_type=ReportType(row.report_type), report_title=row.report_title, from_date=row.from_date, to_date=row.to_date, report_date=row.report_date, account_id=row.account_id, account_code=row.account_code, line_items=line_items, totals=totals, generated_at=row.generated_at, parameters=row.parameters or {}, ) async def save(self, report: FinancialReport) -> FinancialReport: totals = {key: float(value) for key, value in (report.totals or {}).items()} if report.id: result = await self._session.execute( select(FinancialReportModel) .options(selectinload(FinancialReportModel.line_items)) .where(FinancialReportModel.id == report.id) ) row = result.scalar_one_or_none() if row: row.company_id = report.company_id row.report_type = report.report_type.value row.report_title = report.report_title row.from_date = report.from_date row.to_date = report.to_date row.report_date = report.report_date row.account_id = report.account_id row.account_code = report.account_code row.totals = totals row.parameters = report.parameters or {} row.generated_at = report.generated_at row.line_items.clear() for item in report.line_items: row.line_items.append( ReportLineItemModel( id=new_id(), report_id=row.id, account_id=item.account_id, account_code=item.account_code, account_name=item.account_name, account_type=item.account_type, debit=item.debit, credit=item.credit, balance=item.balance, metadata_json=item.metadata or {}, ) ) await self._session.commit() return await self.get_by_id(row.id) # type: ignore[return-value] report_id = report.id or new_id() row = FinancialReportModel( id=report_id, company_id=report.company_id, report_type=report.report_type.value, report_title=report.report_title, from_date=report.from_date, to_date=report.to_date, report_date=report.report_date, account_id=report.account_id, account_code=report.account_code, totals=totals, parameters=report.parameters or {}, generated_at=report.generated_at, line_items=[ ReportLineItemModel( id=new_id(), report_id=report_id, account_id=item.account_id, account_code=item.account_code, account_name=item.account_name, account_type=item.account_type, debit=item.debit, credit=item.credit, balance=item.balance, metadata_json=item.metadata or {}, ) for item in report.line_items ], ) self._session.add(row) await self._session.commit() return await self.get_by_id(report_id) # type: ignore[return-value] async def get_by_id(self, report_id: str, company_id: str | None = None) -> FinancialReport | None: query = ( select(FinancialReportModel) .options(selectinload(FinancialReportModel.line_items)) .where(FinancialReportModel.id == report_id) ) if company_id: query = query.where(FinancialReportModel.company_id == company_id) result = await self._session.execute(query) row = result.scalar_one_or_none() return self._to_entity(row) if row else None async def list_reports( self, company_id: str, report_type: ReportType | None = None, from_date: date | None = None, to_date: date | None = None, skip: int = 0, limit: int = 50, ) -> list[FinancialReport]: filters = [FinancialReportModel.company_id == company_id] if report_type: filters.append(FinancialReportModel.report_type == report_type.value) if from_date: filters.append(FinancialReportModel.report_date >= from_date) if to_date: filters.append(FinancialReportModel.report_date <= to_date) result = await self._session.execute( select(FinancialReportModel) .options(selectinload(FinancialReportModel.line_items)) .where(*filters) .order_by(FinancialReportModel.generated_at.desc()) .offset(skip) .limit(limit) ) return [self._to_entity(row) for row in result.scalars().unique().all()] _TYPE_LABELS = { TxnReportType.SALES_INVOICE.value: "Sales Invoice", TxnReportType.PURCHASE_BILL.value: "Purchase Bill", TxnReportType.PAYMENT_RECEIVED.value: "Payment Received", TxnReportType.PAYMENT_MADE.value: "Payment Made", TxnReportType.SALES_RETURN.value: "Sales Return", TxnReportType.PURCHASE_RETURN.value: "Purchase Return", TxnReportType.STOCK_TRANSFER.value: "Stock Transfer", TxnReportType.CREDIT_NOTE.value: "Credit Note", } _STATUS_LABELS = { TxnReportPaymentStatus.PAID.value: "Paid", TxnReportPaymentStatus.PARTIAL.value: "Partial", TxnReportPaymentStatus.UNPAID.value: "Unpaid", TxnReportPaymentStatus.N_A.value: "N/A", } _COLLATION = "utf8mb4_unicode_ci" def _c(self, expr, length: int = 200): return collate(cast(expr, String(length)), self._COLLATION) def _str_col(self, value, length: int = 40): return self._c(literal(value), length) def _invoice_payment_status(self, total_col, paid_col): return self._c( case( ( and_(total_col <= 0, paid_col <= 0), TxnReportPaymentStatus.UNPAID.value, ), (paid_col >= total_col, TxnReportPaymentStatus.PAID.value), (paid_col > 0, TxnReportPaymentStatus.PARTIAL.value), else_=TxnReportPaymentStatus.UNPAID.value, ), 20, ) def _due_amount(self, total_col, paid_col): return func.greatest(total_col - paid_col, 0) def _txn_union_selects( self, company_id: str, from_date: date, to_date: date, transaction_type: str | None, ) -> list: types = {transaction_type} if transaction_type else set(self._TYPE_LABELS) selects = [] zero = cast(literal(0), Numeric(18, 2)) na = self._str_col(TxnReportPaymentStatus.N_A.value, 20) if TxnReportType.SALES_INVOICE.value in types: paid = SalesInvoiceModel.amount_paid total = SalesInvoiceModel.total_amount selects.append( select( self._c(SalesInvoiceModel.id, 36).label("id"), self._str_col(TxnReportType.SALES_INVOICE.value).label("transaction_type"), SalesInvoiceModel.invoice_date.label("txn_date"), SalesInvoiceModel.created_at.label("txn_at"), self._c(SalesInvoiceModel.invoice_number, 50).label("reference_no"), self._str_col(TxnReportReferenceType.CUSTOMER.value).label("reference_type"), self._c(SalesInvoiceModel.customer_id, 36).label("party_id"), self._c(CustomerModel.name, 200).label("party_name"), self._c(SalesOrderModel.warehouse_id, 36).label("warehouse_id"), self._c(WarehouseModel.name, 200).label("warehouse_name"), total.label("total_amount"), paid.label("paid_amount"), self._due_amount(total, paid).label("due_amount"), self._invoice_payment_status(total, paid).label("payment_status"), self._c(SalesOrderModel.created_by, 36).label("created_by_id"), self._c(UserRegistrationModel.full_name, 200).label("created_by_name"), ) .select_from(SalesInvoiceModel) .join(CustomerModel, CustomerModel.id == SalesInvoiceModel.customer_id) .outerjoin( SalesOrderModel, SalesOrderModel.id == SalesInvoiceModel.sales_order_id, ) .outerjoin( WarehouseModel, WarehouseModel.id == SalesOrderModel.warehouse_id, ) .outerjoin( UserRegistrationModel, UserRegistrationModel.id == SalesOrderModel.created_by, ) .where( SalesInvoiceModel.company_id == company_id, SalesInvoiceModel.status != "cancelled", SalesInvoiceModel.invoice_date >= from_date, SalesInvoiceModel.invoice_date <= to_date, ) ) if TxnReportType.PURCHASE_BILL.value in types: paid = VendorBillModel.amount_paid total = VendorBillModel.total_amount selects.append( select( self._c(VendorBillModel.id, 36).label("id"), self._str_col(TxnReportType.PURCHASE_BILL.value).label("transaction_type"), VendorBillModel.bill_date.label("txn_date"), VendorBillModel.created_at.label("txn_at"), self._c(VendorBillModel.bill_number, 50).label("reference_no"), self._str_col(TxnReportReferenceType.SUPPLIER.value).label("reference_type"), self._c(VendorBillModel.vendor_id, 36).label("party_id"), self._c(VendorModel.name, 200).label("party_name"), self._c(PurchaseOrderModel.warehouse_id, 36).label("warehouse_id"), self._c(WarehouseModel.name, 200).label("warehouse_name"), total.label("total_amount"), paid.label("paid_amount"), self._due_amount(total, paid).label("due_amount"), self._invoice_payment_status(total, paid).label("payment_status"), self._c(PurchaseOrderModel.created_by, 36).label("created_by_id"), self._c(UserRegistrationModel.full_name, 200).label("created_by_name"), ) .select_from(VendorBillModel) .join(VendorModel, VendorModel.id == VendorBillModel.vendor_id) .outerjoin( PurchaseOrderModel, PurchaseOrderModel.id == VendorBillModel.purchase_order_id, ) .outerjoin( WarehouseModel, WarehouseModel.id == PurchaseOrderModel.warehouse_id, ) .outerjoin( UserRegistrationModel, UserRegistrationModel.id == PurchaseOrderModel.created_by, ) .where( VendorBillModel.company_id == company_id, VendorBillModel.status != "cancelled", VendorBillModel.bill_date >= from_date, VendorBillModel.bill_date <= to_date, ) ) if TxnReportType.PAYMENT_RECEIVED.value in types: selects.append( select( self._c(SalesPaymentModel.id, 36).label("id"), self._str_col(TxnReportType.PAYMENT_RECEIVED.value).label("transaction_type"), SalesPaymentModel.payment_date.label("txn_date"), SalesPaymentModel.created_at.label("txn_at"), self._c(SalesPaymentModel.payment_number, 50).label("reference_no"), self._str_col(TxnReportReferenceType.RECEIPT.value).label("reference_type"), self._c(SalesPaymentModel.customer_id, 36).label("party_id"), self._c(CustomerModel.name, 200).label("party_name"), self._c(CustomerModel.default_warehouse_id, 36).label("warehouse_id"), self._c(WarehouseModel.name, 200).label("warehouse_name"), SalesPaymentModel.total_amount.label("total_amount"), SalesPaymentModel.total_amount.label("paid_amount"), zero.label("due_amount"), self._str_col(TxnReportPaymentStatus.PAID.value, 20).label("payment_status"), self._str_col(None, 36).label("created_by_id"), self._str_col(None, 200).label("created_by_name"), ) .select_from(SalesPaymentModel) .join(CustomerModel, CustomerModel.id == SalesPaymentModel.customer_id) .outerjoin( WarehouseModel, WarehouseModel.id == CustomerModel.default_warehouse_id, ) .where( SalesPaymentModel.company_id == company_id, SalesPaymentModel.status != "cancelled", SalesPaymentModel.payment_date >= from_date, SalesPaymentModel.payment_date <= to_date, ) ) if TxnReportType.PAYMENT_MADE.value in types: selects.append( select( self._c(PurchasePaymentModel.id, 36).label("id"), self._str_col(TxnReportType.PAYMENT_MADE.value).label("transaction_type"), PurchasePaymentModel.payment_date.label("txn_date"), PurchasePaymentModel.created_at.label("txn_at"), self._c(PurchasePaymentModel.payment_number, 50).label("reference_no"), self._str_col(TxnReportReferenceType.PAYMENT.value).label("reference_type"), self._c(PurchasePaymentModel.vendor_id, 36).label("party_id"), self._c(VendorModel.name, 200).label("party_name"), self._c(VendorModel.default_warehouse_id, 36).label("warehouse_id"), self._c(WarehouseModel.name, 200).label("warehouse_name"), PurchasePaymentModel.total_amount.label("total_amount"), PurchasePaymentModel.total_amount.label("paid_amount"), zero.label("due_amount"), self._str_col(TxnReportPaymentStatus.PAID.value, 20).label("payment_status"), self._str_col(None, 36).label("created_by_id"), self._str_col(None, 200).label("created_by_name"), ) .select_from(PurchasePaymentModel) .join(VendorModel, VendorModel.id == PurchasePaymentModel.vendor_id) .outerjoin( WarehouseModel, WarehouseModel.id == VendorModel.default_warehouse_id, ) .where( PurchasePaymentModel.company_id == company_id, PurchasePaymentModel.status != "cancelled", PurchasePaymentModel.payment_date >= from_date, PurchasePaymentModel.payment_date <= to_date, ) ) if TxnReportType.STOCK_TRANSFER.value in types: from_wh = aliased(WarehouseModel) to_wh = aliased(WarehouseModel) selects.append( select( self._c(StockTransferModel.id, 36).label("id"), self._str_col(TxnReportType.STOCK_TRANSFER.value).label("transaction_type"), StockTransferModel.transfer_date.label("txn_date"), StockTransferModel.created_at.label("txn_at"), self._c(StockTransferModel.transfer_number, 50).label("reference_no"), self._str_col(TxnReportReferenceType.TRANSFER.value).label("reference_type"), self._c(StockTransferModel.to_warehouse_id, 36).label("party_id"), self._c(to_wh.name, 200).label("party_name"), self._c(StockTransferModel.from_warehouse_id, 36).label("warehouse_id"), self._c(from_wh.name, 200).label("warehouse_name"), StockTransferModel.total_transfer_value.label("total_amount"), zero.label("paid_amount"), zero.label("due_amount"), na.label("payment_status"), self._c(StockTransferModel.created_by, 36).label("created_by_id"), self._c(UserRegistrationModel.full_name, 200).label("created_by_name"), ) .select_from(StockTransferModel) .outerjoin(from_wh, from_wh.id == StockTransferModel.from_warehouse_id) .outerjoin(to_wh, to_wh.id == StockTransferModel.to_warehouse_id) .outerjoin( UserRegistrationModel, UserRegistrationModel.id == StockTransferModel.created_by, ) .where( StockTransferModel.company_id == company_id, StockTransferModel.status != "cancelled", StockTransferModel.transfer_date >= from_date, StockTransferModel.transfer_date <= to_date, ) ) return_item_types = [ value for value in ( TxnReportType.SALES_RETURN.value, TxnReportType.PURCHASE_RETURN.value, TxnReportType.CREDIT_NOTE.value, ) if value in types ] if return_item_types: ref_type = case( ( ItemTransactionModel.txn_type == TxnReportType.PURCHASE_RETURN.value, TxnReportReferenceType.SUPPLIER.value, ), else_=TxnReportReferenceType.CUSTOMER.value, ) selects.append( select( self._c(ItemTransactionModel.id, 36).label("id"), self._c(ItemTransactionModel.txn_type, 40).label("transaction_type"), func.date(ItemTransactionModel.txn_date).label("txn_date"), ItemTransactionModel.txn_date.label("txn_at"), self._c(ItemTransactionModel.txn_number, 50).label("reference_no"), self._c(ref_type, 40).label("reference_type"), self._c(ItemTransactionModel.vendor_id, 36).label("party_id"), self._c(VendorModel.name, 200).label("party_name"), self._c(ItemTransactionModel.warehouse_id, 36).label("warehouse_id"), self._c(WarehouseModel.name, 200).label("warehouse_name"), ItemTransactionModel.grand_total.label("total_amount"), zero.label("paid_amount"), zero.label("due_amount"), na.label("payment_status"), self._c(ItemTransactionModel.created_by, 36).label("created_by_id"), self._c(UserRegistrationModel.full_name, 200).label("created_by_name"), ) .select_from(ItemTransactionModel) .outerjoin(VendorModel, VendorModel.id == ItemTransactionModel.vendor_id) .outerjoin( WarehouseModel, WarehouseModel.id == ItemTransactionModel.warehouse_id, ) .outerjoin( UserRegistrationModel, UserRegistrationModel.id == ItemTransactionModel.created_by, ) .where( ItemTransactionModel.company_id == company_id, ItemTransactionModel.status != "cancelled", ItemTransactionModel.txn_type.in_(return_item_types), func.date(ItemTransactionModel.txn_date) >= from_date, func.date(ItemTransactionModel.txn_date) <= to_date, ) ) return selects def _apply_txn_filters( self, subq, *, reference_type: str | None, warehouse_id: str | None, party: str | None, created_by_id: str | None, payment_status: str | None, min_amount: Decimal | None, max_amount: Decimal | None, ) -> list: filters = [] if reference_type: filters.append(subq.c.reference_type == reference_type) if warehouse_id: filters.append(subq.c.warehouse_id == warehouse_id) if created_by_id: filters.append(subq.c.created_by_id == created_by_id) if payment_status: filters.append(subq.c.payment_status == payment_status) if min_amount is not None: filters.append(subq.c.total_amount >= min_amount) if max_amount is not None: filters.append(subq.c.total_amount <= max_amount) if party: term = f"%{party.strip().lower()}%" filters.append(func.lower(func.coalesce(subq.c.party_name, "")).like(term)) return filters def _empty_txn_data(self, page: int, page_size: int) -> dict: return { "count": 0, "total_amount": Decimal("0.00"), "total_received": Decimal("0.00"), "total_paid": Decimal("0.00"), "pending_amount": Decimal("0.00"), "rows": [], "rows_total": 0, "page": page, "page_size": page_size, "total_pages": 1, "trend": [], "by_type": [], "by_payment_status": [], } async def get_transaction_report_data( self, company_id: str, *, from_date: date, to_date: date, transaction_type: str | None = None, reference_type: str | None = None, warehouse_id: str | None = None, party: str | None = None, created_by_id: str | None = None, payment_status: str | None = None, min_amount: Decimal | None = None, max_amount: Decimal | None = None, trend_granularity: str = "daily", page: int = 1, page_size: int = 10, include_rows: bool = True, include_charts: bool = True, ) -> dict: selects = self._txn_union_selects( company_id, from_date, to_date, transaction_type ) if not selects: return self._empty_txn_data(page, page_size) union_stmt = union_all(*selects) if len(selects) > 1 else selects[0] subq = union_stmt.subquery("txn_report") extra = self._apply_txn_filters( subq, reference_type=reference_type, warehouse_id=warehouse_id, party=party, created_by_id=created_by_id, payment_status=payment_status, min_amount=min_amount, max_amount=max_amount, ) where_clause = extra if extra else [true()] summary = await self._session.execute( select( func.count(), func.coalesce(func.sum(subq.c.total_amount), 0), func.coalesce( func.sum( case( ( subq.c.transaction_type == TxnReportType.PAYMENT_RECEIVED.value, subq.c.total_amount, ), else_=0, ) ), 0, ), func.coalesce( func.sum( case( ( subq.c.transaction_type == TxnReportType.PAYMENT_MADE.value, subq.c.total_amount, ), else_=0, ) ), 0, ), func.coalesce( func.sum( case( ( subq.c.transaction_type.in_( ( TxnReportType.SALES_INVOICE.value, TxnReportType.PURCHASE_BILL.value, ) ), subq.c.due_amount, ), else_=0, ) ), 0, ), ) .select_from(subq) .where(*where_clause) ) row = summary.one() data = { "count": int(row[0] or 0), "total_amount": to_decimal(row[1]), "total_received": to_decimal(row[2]), "total_paid": to_decimal(row[3]), "pending_amount": to_decimal(row[4]), "rows": [], "rows_total": int(row[0] or 0), "page": page, "page_size": page_size, "total_pages": 1, "trend": [], "by_type": [], "by_payment_status": [], } if include_rows: rows_total = data["rows_total"] total_pages = max(1, ceil(rows_total / page_size)) if page_size else 1 page = max(1, min(page, total_pages)) skip = (page - 1) * page_size data["page"] = page data["total_pages"] = total_pages rows_result = await self._session.execute( select(subq) .where(*where_clause) .order_by(subq.c.txn_date.desc(), subq.c.txn_at.desc()) .offset(skip) .limit(page_size) ) data["rows"] = [ TxnReportRow( id=r.id, txn_date=r.txn_date, txn_at=r.txn_at, transaction_type=TxnReportType(r.transaction_type), reference_no=r.reference_no, reference_type=TxnReportReferenceType(r.reference_type), party_id=r.party_id, party_name=r.party_name, warehouse_id=r.warehouse_id, warehouse_name=r.warehouse_name, total_amount=to_decimal(r.total_amount), paid_amount=to_decimal(r.paid_amount), due_amount=to_decimal(r.due_amount), payment_status=TxnReportPaymentStatus(r.payment_status), created_by_id=r.created_by_id, created_by_name=r.created_by_name, ) for r in rows_result.all() ] if include_charts: type_result = await self._session.execute( select( subq.c.transaction_type, func.count(), func.coalesce(func.sum(subq.c.total_amount), 0), ) .select_from(subq) .where(*where_clause) .group_by(subq.c.transaction_type) ) type_rows = type_result.all() type_total = sum(int(r[1] or 0) for r in type_rows) or 0 by_type_map = {r[0]: r for r in type_rows} data["by_type"] = [] for key, label in self._TYPE_LABELS.items(): cnt = int(by_type_map[key][1]) if key in by_type_map else 0 amt = to_decimal(by_type_map[key][2]) if key in by_type_map else Decimal("0.00") data["by_type"].append( TxnReportSlice( key=key, label=label, count=cnt, total_amount=amt, percent_of_total=( (Decimal(cnt) / Decimal(type_total) * Decimal("100")).quantize( Decimal("0.1") ) if type_total else Decimal("0.0") ), ) ) status_result = await self._session.execute( select( subq.c.payment_status, func.count(), func.coalesce(func.sum(subq.c.total_amount), 0), ) .select_from(subq) .where(*where_clause) .group_by(subq.c.payment_status) ) status_rows = status_result.all() status_total = sum(int(r[1] or 0) for r in status_rows) or 0 status_map = {r[0]: r for r in status_rows} data["by_payment_status"] = [] for key, label in self._STATUS_LABELS.items(): cnt = int(status_map[key][1]) if key in status_map else 0 amt = to_decimal(status_map[key][2]) if key in status_map else Decimal("0.00") data["by_payment_status"].append( TxnReportSlice( key=key, label=label, count=cnt, total_amount=amt, percent_of_total=( (Decimal(cnt) / Decimal(status_total) * Decimal("100")).quantize( Decimal("0.1") ) if status_total else Decimal("0.0") ), ) ) if trend_granularity == TxnReportTrendGranularity.MONTHLY.value: bucket = func.date_format(subq.c.txn_date, "%Y-%m-01") elif trend_granularity == TxnReportTrendGranularity.WEEKLY.value: bucket = func.subdate(subq.c.txn_date, func.weekday(subq.c.txn_date)) else: bucket = subq.c.txn_date trend_result = await self._session.execute( select( bucket.label("bucket"), func.count(), func.coalesce(func.sum(subq.c.total_amount), 0), ) .select_from(subq) .where(*where_clause) .group_by(bucket) .order_by(bucket) ) trend = [] for tr in trend_result.all(): bucket_val = tr.bucket if isinstance(bucket_val, str): bucket_val = date.fromisoformat(bucket_val[:10]) elif hasattr(bucket_val, "date"): bucket_val = bucket_val.date() trend.append( TxnReportTrendPoint( date=bucket_val, count=int(tr[1] or 0), total_amount=to_decimal(tr[2]), ) ) data["trend"] = trend return data def _issued_dept_subq( self, company_id: str, from_date: date, to_date: date, warehouse_id: str | None, department_id: str | None, ): filters = [ DepartmentIssueModel.company_id == company_id, DepartmentIssueModel.status != "cancelled", DepartmentIssueModel.issue_date >= from_date, DepartmentIssueModel.issue_date <= to_date, ] if warehouse_id: filters.append(DepartmentIssueModel.from_warehouse_id == warehouse_id) if department_id: filters.append(DepartmentIssueModel.department_id == department_id) return ( select( DepartmentIssueModel.department_id.label("department_id"), func.count(DepartmentIssueModel.id).label("issued_count"), func.coalesce( func.sum(DepartmentIssueModel.total_estimated_value), 0 ).label("issued_amount"), ) .where(*filters) .group_by(DepartmentIssueModel.department_id) .subquery() ) def _received_dept_subq( self, company_id: str, from_date: date, to_date: date, warehouse_id: str | None, department_id: str | None, ): filters = [ PurchaseOrderModel.company_id == company_id, DepartmentModel.company_id == company_id, PurchaseOrderModel.status != "cancelled", PurchaseOrderModel.order_date >= from_date, PurchaseOrderModel.order_date <= to_date, PurchaseOrderModel.department.isnot(None), PurchaseOrderModel.department != "", ] if warehouse_id: filters.append(PurchaseOrderModel.warehouse_id == warehouse_id) if department_id: filters.append(DepartmentModel.id == department_id) return ( select( DepartmentModel.id.label("department_id"), func.count(PurchaseOrderModel.id).label("received_count"), func.coalesce(func.sum(PurchaseOrderModel.total_amount), 0).label( "received_amount" ), ) .select_from(PurchaseOrderModel) .join( DepartmentModel, and_( DepartmentModel.company_id == PurchaseOrderModel.company_id, or_( func.lower(DepartmentModel.name) == func.lower(PurchaseOrderModel.department), func.lower(DepartmentModel.code) == func.lower(PurchaseOrderModel.department), ), ), ) .where(*filters) .group_by(DepartmentModel.id) .subquery() ) async def _department_period_stats( self, company_id: str, from_date: date, to_date: date, warehouse_id: str | None, department_id: str | None, search: str | None, ) -> dict: issued = self._issued_dept_subq( company_id, from_date, to_date, warehouse_id, department_id ) received = self._received_dept_subq( company_id, from_date, to_date, warehouse_id, department_id ) filters = [DepartmentModel.company_id == company_id] if search: term = f"%{search.strip().lower()}%" filters.append( or_( func.lower(DepartmentModel.name).like(term), func.lower(DepartmentModel.code).like(term), ) ) activity = or_( func.coalesce(issued.c.issued_count, 0) > 0, func.coalesce(received.c.received_count, 0) > 0, ) result = await self._session.execute( select( func.count(DepartmentModel.id), func.coalesce( func.sum( func.coalesce(issued.c.issued_count, 0) + func.coalesce(received.c.received_count, 0) ), 0, ), func.coalesce(func.sum(issued.c.issued_amount), 0), func.coalesce(func.sum(received.c.received_amount), 0), ) .select_from(DepartmentModel) .outerjoin(issued, issued.c.department_id == DepartmentModel.id) .outerjoin(received, received.c.department_id == DepartmentModel.id) .where(*filters, activity) ) row = result.one() issued_amt = to_decimal(row[2]) received_amt = to_decimal(row[3]) return { "departments_count": int(row[0] or 0), "transactions_count": int(row[1] or 0), "issued_amount": issued_amt, "received_amount": received_amt, "total_value": issued_amt + received_amt, } async def get_department_report_data( self, company_id: str, *, from_date: date, to_date: date, warehouse_id: str | None = None, department_id: str | None = None, search: str | None = None, sort_by: str = "transactions", sort_dir: str = "desc", page: int = 1, page_size: int = 8, include_rows: bool = True, include_charts: bool = True, include_details: bool = False, previous_from: date | None = None, previous_to: date | None = None, ) -> dict: stats = await self._department_period_stats( company_id, from_date, to_date, warehouse_id, department_id, search ) data = { **stats, "rows": [], "details": [], "rows_total": stats["departments_count"], "page": page, "page_size": page_size, "total_pages": 1, "top_by_value": [], "by_department": [], "issued_vs_received": [], "contribution": [], } issued = self._issued_dept_subq( company_id, from_date, to_date, warehouse_id, department_id ) received = self._received_dept_subq( company_id, from_date, to_date, warehouse_id, department_id ) prev_issued = None prev_received = None if previous_from and previous_to: prev_issued = self._issued_dept_subq( company_id, previous_from, previous_to, warehouse_id, department_id ) prev_received = self._received_dept_subq( company_id, previous_from, previous_to, warehouse_id, department_id ) txn_count = func.coalesce(issued.c.issued_count, 0) + func.coalesce( received.c.received_count, 0 ) issued_amt = func.coalesce(issued.c.issued_amount, 0) received_amt = func.coalesce(received.c.received_amount, 0) total_value = issued_amt + received_amt net_value = issued_amt - received_amt filters = [DepartmentModel.company_id == company_id] if search: term = f"%{search.strip().lower()}%" filters.append( or_( func.lower(DepartmentModel.name).like(term), func.lower(DepartmentModel.code).like(term), ) ) activity = or_( func.coalesce(issued.c.issued_count, 0) > 0, func.coalesce(received.c.received_count, 0) > 0, ) query = ( select( DepartmentModel.id, DepartmentModel.code, DepartmentModel.name, txn_count.label("transactions_count"), issued_amt.label("issued_amount"), received_amt.label("received_amount"), net_value.label("net_value"), total_value.label("total_value"), ) .select_from(DepartmentModel) .outerjoin(issued, issued.c.department_id == DepartmentModel.id) .outerjoin(received, received.c.department_id == DepartmentModel.id) .where(*filters, activity) ) if prev_issued is not None and prev_received is not None: prev_total = func.coalesce(prev_issued.c.issued_amount, 0) + func.coalesce( prev_received.c.received_amount, 0 ) query = query.add_columns(prev_total.label("previous_total")).outerjoin( prev_issued, prev_issued.c.department_id == DepartmentModel.id ).outerjoin( prev_received, prev_received.c.department_id == DepartmentModel.id ) else: query = query.add_columns(literal(0).label("previous_total")) sort_map = { "name": DepartmentModel.name, "transactions": txn_count, "issued": issued_amt, "received": received_amt, "net": net_value, "total": total_value, } sort_col = sort_map.get(sort_by, txn_count) order = sort_col.asc() if sort_dir == "asc" else sort_col.desc() all_rows = [] if include_rows or include_charts: result = await self._session.execute(query.order_by(order, DepartmentModel.name.asc())) all_rows = result.all() grand_total = stats["total_value"] or Decimal("0.00") def _to_row(r) -> DeptReportRow: total = to_decimal(r.total_value) prev_total = to_decimal(r.previous_total) change = None if prev_total != 0: change = ((total - prev_total) / prev_total * Decimal("100")).quantize( Decimal("0.1") ) return DeptReportRow( department_id=r.id, department_code=r.code, department_name=r.name, transactions_count=int(r.transactions_count or 0), issued_amount=to_decimal(r.issued_amount), received_amount=to_decimal(r.received_amount), net_value=to_decimal(r.net_value), total_value=total, percent_of_total=( (total / grand_total * Decimal("100")).quantize(Decimal("0.1")) if grand_total else Decimal("0.0") ), change_percent=change, ) mapped = [_to_row(r) for r in all_rows] if include_rows: rows_total = len(mapped) total_pages = max(1, ceil(rows_total / page_size)) if page_size else 1 page = max(1, min(page, total_pages)) skip = (page - 1) * page_size data["page"] = page data["total_pages"] = total_pages data["rows_total"] = rows_total data["rows"] = mapped[skip : skip + page_size] if include_charts: chart_rows = mapped[:8] data["top_by_value"] = [ TxnReportSlice( key=row.department_id, label=row.department_name, count=row.transactions_count, total_amount=row.total_value, percent_of_total=row.percent_of_total, ) for row in sorted(mapped, key=lambda x: x.total_value, reverse=True)[:5] ] data["contribution"] = [ TxnReportSlice( key=row.department_id, label=row.department_name, count=row.transactions_count, total_amount=row.total_value, percent_of_total=row.percent_of_total, ) for row in sorted(mapped, key=lambda x: x.total_value, reverse=True) ] data["by_department"] = [ DeptReportComboPoint( department_id=row.department_id, department_name=row.department_name, transactions_count=row.transactions_count, total_amount=row.total_value, ) for row in chart_rows ] data["issued_vs_received"] = [ DeptReportIssuedReceived( department_id=row.department_id, department_name=row.department_name, issued_amount=row.issued_amount, received_amount=row.received_amount, ) for row in chart_rows ] if include_details: data["details"] = await self._department_detail_rows( company_id, from_date, to_date, warehouse_id, department_id, search, page, page_size, ) return data async def _department_detail_rows( self, company_id: str, from_date: date, to_date: date, warehouse_id: str | None, department_id: str | None, search: str | None, page: int, page_size: int, ) -> list[DeptReportDetailRow]: issue_filters = [ DepartmentIssueModel.company_id == company_id, DepartmentIssueModel.status != "cancelled", DepartmentIssueModel.issue_date >= from_date, DepartmentIssueModel.issue_date <= to_date, ] if warehouse_id: issue_filters.append(DepartmentIssueModel.from_warehouse_id == warehouse_id) if department_id: issue_filters.append(DepartmentIssueModel.department_id == department_id) if search: term = f"%{search.strip().lower()}%" issue_filters.append( or_( func.lower(DepartmentModel.name).like(term), func.lower(DepartmentModel.code).like(term), ) ) issued_select = ( select( self._c(DepartmentIssueModel.id, 36).label("id"), DepartmentIssueModel.issue_date.label("txn_date"), self._str_col("issued", 20).label("direction"), self._c(DepartmentIssueModel.issue_number, 50).label("reference_no"), self._c(DepartmentModel.id, 36).label("department_id"), self._c(DepartmentModel.code, 6).label("department_code"), self._c(DepartmentModel.name, 200).label("department_name"), self._c(DepartmentIssueModel.from_warehouse_id, 36).label("warehouse_id"), self._c(WarehouseModel.name, 200).label("warehouse_name"), DepartmentIssueModel.total_estimated_value.label("amount"), ) .join(DepartmentModel, DepartmentModel.id == DepartmentIssueModel.department_id) .outerjoin( WarehouseModel, WarehouseModel.id == DepartmentIssueModel.from_warehouse_id, ) .where(*issue_filters) ) po_filters = [ PurchaseOrderModel.company_id == company_id, DepartmentModel.company_id == company_id, PurchaseOrderModel.status != "cancelled", PurchaseOrderModel.order_date >= from_date, PurchaseOrderModel.order_date <= to_date, PurchaseOrderModel.department.isnot(None), PurchaseOrderModel.department != "", ] if warehouse_id: po_filters.append(PurchaseOrderModel.warehouse_id == warehouse_id) if department_id: po_filters.append(DepartmentModel.id == department_id) if search: term = f"%{search.strip().lower()}%" po_filters.append( or_( func.lower(DepartmentModel.name).like(term), func.lower(DepartmentModel.code).like(term), ) ) received_select = ( select( self._c(PurchaseOrderModel.id, 36).label("id"), PurchaseOrderModel.order_date.label("txn_date"), self._str_col("received", 20).label("direction"), self._c(PurchaseOrderModel.po_number, 50).label("reference_no"), self._c(DepartmentModel.id, 36).label("department_id"), self._c(DepartmentModel.code, 6).label("department_code"), self._c(DepartmentModel.name, 200).label("department_name"), self._c(PurchaseOrderModel.warehouse_id, 36).label("warehouse_id"), self._c(WarehouseModel.name, 200).label("warehouse_name"), PurchaseOrderModel.total_amount.label("amount"), ) .join( DepartmentModel, and_( DepartmentModel.company_id == PurchaseOrderModel.company_id, or_( func.lower(DepartmentModel.name) == func.lower(PurchaseOrderModel.department), func.lower(DepartmentModel.code) == func.lower(PurchaseOrderModel.department), ), ), ) .outerjoin( WarehouseModel, WarehouseModel.id == PurchaseOrderModel.warehouse_id ) .where(*po_filters) ) union_stmt = union_all(issued_select, received_select).subquery("dept_details") skip = max(0, (page - 1) * page_size) result = await self._session.execute( select(union_stmt) .order_by(union_stmt.c.txn_date.desc(), union_stmt.c.reference_no.desc()) .offset(skip) .limit(page_size) ) return [ DeptReportDetailRow( id=row.id, txn_date=row.txn_date, direction=str(row.direction), reference_no=row.reference_no, department_id=row.department_id, department_code=row.department_code, department_name=row.department_name, warehouse_id=row.warehouse_id, warehouse_name=row.warehouse_name, amount=to_decimal(row.amount), ) for row in result.all() ] def _to_custom_report(self, row: CustomReportModel) -> CustomReport: config_raw = row.configuration or {} config = CustomReportConfig.model_validate(config_raw) return CustomReport( id=row.id, company_id=row.company_id, name=row.name, module=CustomReportModule(row.module), description=row.description, configuration=config, is_template=bool(row.is_template), created_by=row.created_by, created_at=row.created_at, updated_at=row.updated_at, ) async def create_custom_report(self, report: CustomReport) -> CustomReport: row = CustomReportModel( id=report.id or new_id(), company_id=report.company_id, name=report.name, module=report.module.value, description=report.description, configuration=report.configuration.model_dump(mode="json"), is_template=report.is_template, created_by=report.created_by, created_at=report.created_at, updated_at=report.updated_at, ) self._session.add(row) await self._session.commit() await self._session.refresh(row) return self._to_custom_report(row) async def get_custom_report( self, report_id: str, company_id: str ) -> CustomReport | None: result = await self._session.execute( select(CustomReportModel).where( CustomReportModel.id == report_id, CustomReportModel.company_id == company_id, ) ) row = result.scalar_one_or_none() return self._to_custom_report(row) if row else None async def list_custom_reports( self, company_id: str, *, module: str | None = None, search: str | None = None, skip: int = 0, limit: int = 50, ) -> list[CustomReport]: filters = [CustomReportModel.company_id == company_id] if module: filters.append(CustomReportModel.module == module) if search: term = f"%{search.strip().lower()}%" filters.append(func.lower(CustomReportModel.name).like(term)) result = await self._session.execute( select(CustomReportModel) .where(*filters) .order_by(CustomReportModel.updated_at.desc()) .offset(skip) .limit(limit) ) return [self._to_custom_report(row) for row in result.scalars().all()] async def count_custom_reports( self, company_id: str, *, module: str | None = None, search: str | None = None, ) -> int: filters = [CustomReportModel.company_id == company_id] if module: filters.append(CustomReportModel.module == module) if search: term = f"%{search.strip().lower()}%" filters.append(func.lower(CustomReportModel.name).like(term)) result = await self._session.execute( select(func.count()).select_from(CustomReportModel).where(*filters) ) return int(result.scalar_one() or 0) async def update_custom_report( self, report_id: str, report: CustomReport ) -> CustomReport | None: row = await self._session.get(CustomReportModel, report_id) if not row or row.company_id != report.company_id: return None row.name = report.name row.module = report.module.value row.description = report.description row.configuration = report.configuration.model_dump(mode="json") row.is_template = report.is_template row.updated_at = report.updated_at await self._session.commit() await self._session.refresh(row) return self._to_custom_report(row) async def delete_custom_report(self, report_id: str, company_id: str) -> bool: row = await self._session.get(CustomReportModel, report_id) if not row or row.company_id != company_id: return False await self._session.delete(row) await self._session.commit() return True def _custom_group_col(self, subq, field: str): mapping = { CustomReportField.DATE.value: subq.c.txn_date, CustomReportField.TRANSACTION_TYPE.value: subq.c.transaction_type, CustomReportField.PARTY_NAME.value: subq.c.party_name, CustomReportField.WAREHOUSE_NAME.value: subq.c.warehouse_name, CustomReportField.PAYMENT_STATUS.value: subq.c.payment_status, CustomReportField.REFERENCE_NO.value: subq.c.reference_no, CustomReportField.REFERENCE_TYPE.value: subq.c.reference_type, } return mapping.get(field, subq.c.transaction_type) def _custom_label(self, field: str, value) -> str: if value is None: return "N/A" text = str(value) if field == CustomReportField.TRANSACTION_TYPE.value: return self._TYPE_LABELS.get(text, text) if field == CustomReportField.PAYMENT_STATUS.value: return self._STATUS_LABELS.get(text, text) return text async def run_custom_report_data( self, company_id: str, *, from_date: date, to_date: date, transaction_types: list[str] | None = None, warehouse_id: str | None = None, payment_status: str | None = None, party: str | None = None, group_by: str = "transaction_type", then_by: str | None = None, sort_by: str = "total_amount", sort_dir: str = "desc", page: int = 1, page_size: int = 50, ) -> dict: selects = self._txn_union_selects(company_id, from_date, to_date, None) if not selects: empty_totals = CustomReportTotals() return { "rows": [], "totals": empty_totals, "rows_total": 0, "page": page, "page_size": page_size, "total_pages": 1, "trend": [], "top_parties": [], "by_type": [], "by_payment_status": [], } union_stmt = union_all(*selects) if len(selects) > 1 else selects[0] subq = union_stmt.subquery("custom_txn") filters = self._apply_txn_filters( subq, reference_type=None, warehouse_id=warehouse_id, party=party, created_by_id=None, payment_status=payment_status, min_amount=None, max_amount=None, ) if transaction_types: filters.append(subq.c.transaction_type.in_(transaction_types)) where_clause = filters if filters else [true()] group_col = self._custom_group_col(subq, group_by) then_col = self._custom_group_col(subq, then_by) if then_by else None group_cols = [group_col] if then_col is not None: group_cols.append(then_col) txn_count = func.count().label("total_transactions") qty = func.count().label("total_quantity") amount = func.coalesce(func.sum(subq.c.total_amount), 0).label("total_amount") avg_amt = func.coalesce(func.avg(subq.c.total_amount), 0).label("avg_amount") select_cols = [ group_col.label("group_key"), txn_count, qty, amount, avg_amt, ] if then_col is not None: select_cols.insert(1, then_col.label("then_key")) sort_map = { "total_amount": amount, "quantity": qty, "transactions": txn_count, "avg_amount": avg_amt, "group_key": group_col, } sort_col = sort_map.get(sort_by, amount) order = sort_col.asc() if sort_dir == "asc" else sort_col.desc() grouped = ( select(*select_cols) .select_from(subq) .where(*where_clause) .group_by(*group_cols) .order_by(order) ) count_result = await self._session.execute( select(func.count()).select_from(grouped.subquery()) ) rows_total = int(count_result.scalar_one() or 0) total_pages = max(1, ceil(rows_total / page_size)) if page_size else 1 page = max(1, min(page, total_pages)) skip = (page - 1) * page_size rows_result = await self._session.execute( grouped.offset(skip).limit(page_size) ) rows = [] for r in rows_result.all(): then_key = getattr(r, "then_key", None) if then_col is not None else None rows.append( CustomReportResultRow( group_key=str(r.group_key) if r.group_key is not None else "N/A", group_label=self._custom_label(group_by, r.group_key), then_key=str(then_key) if then_key is not None else None, then_label=( self._custom_label(then_by, then_key) if then_by and then_key is not None else None ), total_transactions=int(r.total_transactions or 0), total_quantity=to_decimal(r.total_quantity), total_amount=to_decimal(r.total_amount), avg_amount=to_decimal(r.avg_amount), ) ) totals_result = await self._session.execute( select( func.count(), func.coalesce(func.sum(subq.c.total_amount), 0), func.coalesce(func.avg(subq.c.total_amount), 0), ) .select_from(subq) .where(*where_clause) ) totals_row = totals_result.one() total_txns = int(totals_row[0] or 0) total_amount = to_decimal(totals_row[1]) avg_amount = to_decimal(totals_row[2]) totals = CustomReportTotals( total_transactions=total_txns, total_quantity=Decimal(str(total_txns)), total_amount=total_amount, avg_amount=avg_amount, ) trend_result = await self._session.execute( select( subq.c.txn_date.label("bucket"), func.count(), func.coalesce(func.sum(subq.c.total_amount), 0), ) .select_from(subq) .where(*where_clause) .group_by(subq.c.txn_date) .order_by(subq.c.txn_date) ) trend = [ TxnReportTrendPoint( date=tr.bucket, count=int(tr[1] or 0), total_amount=to_decimal(tr[2]), ) for tr in trend_result.all() ] parties_result = await self._session.execute( select( func.coalesce(subq.c.party_name, "N/A").label("name"), func.count(), func.coalesce(func.sum(subq.c.total_amount), 0), ) .select_from(subq) .where(*where_clause) .group_by(subq.c.party_name) .order_by(func.sum(subq.c.total_amount).desc()) .limit(5) ) top_parties = [] for pr in parties_result.all(): amt = to_decimal(pr[2]) top_parties.append( TxnReportSlice( key=str(pr.name), label=str(pr.name), count=int(pr[1] or 0), total_amount=amt, percent_of_total=( (amt / total_amount * Decimal("100")).quantize(Decimal("0.1")) if total_amount else Decimal("0.0") ), ) ) type_result = await self._session.execute( select( subq.c.transaction_type, func.count(), func.coalesce(func.sum(subq.c.total_amount), 0), ) .select_from(subq) .where(*where_clause) .group_by(subq.c.transaction_type) ) by_type = [] for tr in type_result.all(): amt = to_decimal(tr[2]) key = str(tr[0]) by_type.append( TxnReportSlice( key=key, label=self._TYPE_LABELS.get(key, key), count=int(tr[1] or 0), total_amount=amt, percent_of_total=( (amt / total_amount * Decimal("100")).quantize(Decimal("0.1")) if total_amount else Decimal("0.0") ), ) ) status_result = await self._session.execute( select( subq.c.payment_status, func.count(), func.coalesce(func.sum(subq.c.total_amount), 0), ) .select_from(subq) .where(*where_clause) .group_by(subq.c.payment_status) ) status_total = total_txns or 1 by_payment_status = [] status_map = {str(r[0]): r for r in status_result.all()} for key, label in self._STATUS_LABELS.items(): row = status_map.get(key) cnt = int(row[1] or 0) if row else 0 amt = to_decimal(row[2]) if row else Decimal("0.00") by_payment_status.append( TxnReportSlice( key=key, label=label, count=cnt, total_amount=amt, percent_of_total=( (Decimal(cnt) / Decimal(status_total) * Decimal("100")).quantize( Decimal("0.1") ) if total_txns else Decimal("0.0") ), ) ) return { "rows": rows, "totals": totals, "rows_total": rows_total, "page": page, "page_size": page_size, "total_pages": total_pages, "trend": trend, "top_parties": top_parties, "by_type": by_type, "by_payment_status": by_payment_status, }