from datetime import date, datetime, timedelta from decimal import Decimal from app.application.company_access import resolve_company_for_user from app.application.exceptions import ConflictError, ForbiddenError, NotFoundError, ValidationError from app.domain.entities.company import Company from app.domain.entities.report import ( CustomReport, CustomReportConfig, CustomReportResult, DepartmentReport, FinancialReport, ReportLineItem, TransactionReport, TxnReportKpi, TxnReportTrendPoint, ) from app.domain.entities.user_registration import UserRegistration from app.domain.enums import ( AccountNature, AccountType, CustomReportField, CustomReportMetric, CustomReportModule, CustomReportSort, DeptReportCompareWith, DeptReportSort, DeptReportView, ReportType, TxnReportPaymentStatus, TxnReportReferenceType, TxnReportTrendGranularity, TxnReportType, ) from app.domain.repositories.chart_of_account_repository import ChartOfAccountRepository from app.domain.repositories.company_repository import CompanyRepository from app.domain.repositories.report_repository import ReportRepository from app.domain.repositories.voucher_repository import VoucherRepository class ReportService: def __init__( self, account_repository: ChartOfAccountRepository, voucher_repository: VoucherRepository, report_repository: ReportRepository, company_repository: CompanyRepository, ) -> None: self._accounts = account_repository self._vouchers = voucher_repository self._reports = report_repository self._companies = company_repository async def generate_ledger( self, user: UserRegistration, company_id: str, account_id: str, from_date: date | None = None, to_date: date | None = None, save: bool = True, ) -> FinancialReport: await self._get_user_company(user, company_id) account = await self._accounts.get_by_id(account_id, company_id) if not account: raise NotFoundError("Account not found") entries = await self._vouchers.get_ledger_entries(company_id, account_id, from_date, to_date) line_items = [ ReportLineItem( account_id=account_id, account_code=entry.account_code, account_name=entry.account_name, debit=entry.debit_amount, credit=entry.credit_amount, balance=entry.running_balance, metadata={ "voucher_id": entry.voucher_id, "voucher_number": entry.voucher_number, "voucher_date": str(entry.voucher_date), "description": entry.description, }, ) for entry in entries ] total_debit = sum((item.debit for item in line_items), Decimal("0.00")) total_credit = sum((item.credit for item in line_items), Decimal("0.00")) closing_balance = entries[-1].running_balance if entries else account.current_balance report = FinancialReport( company_id=company_id, report_type=ReportType.LEDGER, report_title=f"Ledger - {account.code} {account.name}", from_date=from_date, to_date=to_date, report_date=to_date or date.today(), account_id=account_id, account_code=account.code, line_items=line_items, totals={ "total_debit": total_debit, "total_credit": total_credit, "closing_balance": closing_balance, }, parameters={"company_id": company_id, "account_id": account_id}, ) return await self._reports.save(report) if save else report async def generate_trial_balance( self, user: UserRegistration, company_id: str, as_of_date: date, save: bool = True, ) -> FinancialReport: await self._get_user_company(user, company_id) accounts = await self._accounts.list_all(company_id, is_active=True, skip=0, limit=10000) line_items: list[ReportLineItem] = [] total_debit = Decimal("0.00") total_credit = Decimal("0.00") for account in accounts: if account.is_group: continue balance = await self._get_account_balance_as_of( company_id, account.id or "", as_of_date, account ) if balance == Decimal("0.00"): continue if account.nature == AccountNature.DEBIT: debit = balance if balance > 0 else Decimal("0.00") credit = abs(balance) if balance < 0 else Decimal("0.00") else: credit = balance if balance > 0 else Decimal("0.00") debit = abs(balance) if balance < 0 else Decimal("0.00") line_items.append( ReportLineItem( account_id=account.id, account_code=account.code, account_name=account.name, account_type=account.account_type.value, debit=debit, credit=credit, balance=balance, ) ) total_debit += debit total_credit += credit report = FinancialReport( company_id=company_id, report_type=ReportType.TRIAL_BALANCE, report_title="Trial Balance", report_date=as_of_date, line_items=sorted(line_items, key=lambda item: item.account_code), totals={ "total_debit": total_debit, "total_credit": total_credit, "difference": total_debit - total_credit, }, parameters={"company_id": company_id, "as_of_date": str(as_of_date)}, ) return await self._reports.save(report) if save else report async def generate_income_statement( self, user: UserRegistration, company_id: str, from_date: date, to_date: date, save: bool = True, ) -> FinancialReport: await self._get_user_company(user, company_id) if from_date > to_date: raise ValidationError("from_date cannot be after to_date") accounts = await self._accounts.list_all(company_id, is_active=True, skip=0, limit=10000) revenue_items: list[ReportLineItem] = [] expense_items: list[ReportLineItem] = [] total_revenue = Decimal("0.00") total_expense = Decimal("0.00") for account in accounts: if account.is_group: continue if account.account_type == AccountType.REVENUE: amount = await self._get_period_movement( company_id, account.id or "", from_date, to_date, account ) if amount != Decimal("0.00"): revenue_items.append( ReportLineItem( account_id=account.id, account_code=account.code, account_name=account.name, account_type=account.account_type.value, balance=amount, ) ) total_revenue += amount elif account.account_type == AccountType.EXPENSE: amount = await self._get_period_movement( company_id, account.id or "", from_date, to_date, account ) if amount != Decimal("0.00"): expense_items.append( ReportLineItem( account_id=account.id, account_code=account.code, account_name=account.name, account_type=account.account_type.value, balance=amount, ) ) total_expense += amount net_income = total_revenue - total_expense line_items = sorted(revenue_items, key=lambda item: item.account_code) + sorted( expense_items, key=lambda item: item.account_code ) report = FinancialReport( company_id=company_id, report_type=ReportType.INCOME_STATEMENT, report_title="Income Statement", from_date=from_date, to_date=to_date, report_date=to_date, line_items=line_items, totals={ "total_revenue": total_revenue, "total_expense": total_expense, "net_income": net_income, }, parameters={"company_id": company_id, "from_date": str(from_date), "to_date": str(to_date)}, ) return await self._reports.save(report) if save else report async def generate_balance_sheet( self, user: UserRegistration, company_id: str, as_of_date: date, save: bool = True, ) -> FinancialReport: await self._get_user_company(user, company_id) accounts = await self._accounts.list_all(company_id, is_active=True, skip=0, limit=10000) assets: list[ReportLineItem] = [] liabilities: list[ReportLineItem] = [] equity: list[ReportLineItem] = [] total_assets = Decimal("0.00") total_liabilities = Decimal("0.00") total_equity = Decimal("0.00") for account in accounts: if account.is_group: continue balance = await self._get_account_balance_as_of( company_id, account.id or "", as_of_date, account ) if balance == Decimal("0.00"): continue item = ReportLineItem( account_id=account.id, account_code=account.code, account_name=account.name, account_type=account.account_type.value, balance=balance, ) if account.account_type == AccountType.ASSET: assets.append(item) total_assets += balance elif account.account_type == AccountType.LIABILITY: liabilities.append(item) total_liabilities += balance elif account.account_type == AccountType.EQUITY: equity.append(item) total_equity += balance # Unclosed P&L must appear in equity so Assets = Liabilities + Equity. # Books are not year-closed into Retained Earnings, so use inception → as_of. earnings_from = date(1900, 1, 1) net_income = await self._compute_net_income( company_id, accounts, earnings_from, as_of_date ) if net_income != Decimal("0.00"): equity.append( ReportLineItem( account_id=None, account_code="CYE", account_name="Current Earnings (Unclosed)", account_type=AccountType.EQUITY.value, balance=net_income, metadata={ "synthetic": True, "source": "income_statement", "from_date": str(earnings_from), "to_date": str(as_of_date), }, ) ) total_equity += net_income liabilities_and_equity = total_liabilities + total_equity difference = (total_assets - liabilities_and_equity).quantize(Decimal("0.01")) line_items = ( sorted(assets, key=lambda item: item.account_code) + sorted(liabilities, key=lambda item: item.account_code) + sorted(equity, key=lambda item: item.account_code) ) report = FinancialReport( company_id=company_id, report_type=ReportType.BALANCE_SHEET, report_title="Balance Sheet", report_date=as_of_date, line_items=line_items, totals={ "total_assets": total_assets, "total_liabilities": total_liabilities, "total_equity": total_equity, "liabilities_and_equity": liabilities_and_equity, "net_income": net_income, "difference": difference, }, parameters={ "company_id": company_id, "as_of_date": str(as_of_date), "is_balanced": abs(difference) < Decimal("0.01"), }, ) return await self._reports.save(report) if save else report async def get_report( self, user: UserRegistration, company_id: str, report_id: str, ) -> FinancialReport: await self._get_user_company(user, company_id) report = await self._reports.get_by_id(report_id, company_id) if not report: raise NotFoundError("Report not found") return report async def list_reports( self, user: UserRegistration, 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]: await self._get_user_company(user, company_id) return await self._reports.list_reports( company_id, report_type, from_date, to_date, skip, limit ) async def get_transaction_report( self, user: UserRegistration, company_id: str, from_date: date, to_date: date, *, transaction_type: TxnReportType | None = None, reference_type: TxnReportReferenceType | None = None, warehouse_id: str | None = None, department_id: str | None = None, party: str | None = None, created_by_id: str | None = None, payment_status: TxnReportPaymentStatus | None = None, min_amount: Decimal | None = None, max_amount: Decimal | None = None, trend_granularity: TxnReportTrendGranularity = TxnReportTrendGranularity.DAILY, page: int = 1, page_size: int = 10, ) -> TransactionReport: await self._get_user_company(user, company_id) if to_date < from_date: raise ValidationError("to_date must be on or after from_date") if min_amount is not None and max_amount is not None and max_amount < min_amount: raise ValidationError("max_amount must be on or after min_amount") _ = department_id filters = { "transaction_type": transaction_type.value if transaction_type else None, "reference_type": reference_type.value if reference_type else None, "warehouse_id": warehouse_id, "party": party, "created_by_id": created_by_id, "payment_status": payment_status.value if payment_status else None, "min_amount": min_amount, "max_amount": max_amount, } previous_from, previous_to = self._previous_period(from_date, to_date) current = await self._reports.get_transaction_report_data( company_id, from_date=from_date, to_date=to_date, trend_granularity=trend_granularity.value, page=page, page_size=page_size, include_rows=True, include_charts=True, **filters, ) previous = await self._reports.get_transaction_report_data( company_id, from_date=previous_from, to_date=previous_to, trend_granularity=trend_granularity.value, page=1, page_size=1, include_rows=False, include_charts=False, **filters, ) return TransactionReport( from_date=from_date, to_date=to_date, previous_from_date=previous_from, previous_to_date=previous_to, trend_granularity=trend_granularity, total_transactions=self._kpi_metric(current["count"], previous["count"]), total_amount=self._kpi_metric( current["total_amount"], previous["total_amount"] ), total_received=self._kpi_metric( current["total_received"], previous["total_received"] ), total_paid=self._kpi_metric(current["total_paid"], previous["total_paid"]), pending_amount=self._kpi_metric( current["pending_amount"], previous["pending_amount"] ), rows=current["rows"], rows_total=current["rows_total"], page=current["page"], page_size=current["page_size"], total_pages=current["total_pages"], trend=self._fill_txn_trend( from_date, to_date, current["trend"], trend_granularity ), by_type=current["by_type"], by_payment_status=current["by_payment_status"], ) async def get_department_report( self, user: UserRegistration, company_id: str, from_date: date, to_date: date, *, report_view: DeptReportView = DeptReportView.SUMMARY, compare_with: DeptReportCompareWith = DeptReportCompareWith.PREVIOUS_PERIOD, warehouse_id: str | None = None, department_id: str | None = None, search: str | None = None, sort_by: DeptReportSort = DeptReportSort.TRANSACTIONS, sort_dir: str = "desc", page: int = 1, page_size: int = 8, ) -> DepartmentReport: await self._get_user_company(user, company_id) if to_date < from_date: raise ValidationError("to_date must be on or after from_date") if sort_dir not in ("asc", "desc"): raise ValidationError("sort_dir must be asc or desc") _ = compare_with previous_from, previous_to = self._previous_period(from_date, to_date) current = await self._reports.get_department_report_data( company_id, from_date=from_date, to_date=to_date, warehouse_id=warehouse_id, department_id=department_id, search=search, sort_by=sort_by.value, sort_dir=sort_dir, page=page, page_size=page_size, include_rows=True, include_charts=True, include_details=report_view == DeptReportView.DETAILED, previous_from=previous_from, previous_to=previous_to, ) previous = await self._reports.get_department_report_data( company_id, from_date=previous_from, to_date=previous_to, warehouse_id=warehouse_id, department_id=department_id, search=search, page=1, page_size=1, include_rows=False, include_charts=False, include_details=False, ) return DepartmentReport( from_date=from_date, to_date=to_date, previous_from_date=previous_from, previous_to_date=previous_to, report_view=report_view, compare_with=compare_with, total_departments=self._kpi_metric( current["departments_count"], previous["departments_count"] ), total_transactions=self._kpi_metric( current["transactions_count"], previous["transactions_count"] ), total_value=self._kpi_metric(current["total_value"], previous["total_value"]), total_issued=self._kpi_metric( current["issued_amount"], previous["issued_amount"] ), total_received=self._kpi_metric( current["received_amount"], previous["received_amount"] ), rows=current["rows"], details=current["details"], rows_total=current["rows_total"], page=current["page"], page_size=current["page_size"], total_pages=current["total_pages"], top_by_value=current["top_by_value"], by_department=current["by_department"], issued_vs_received=current["issued_vs_received"], contribution=current["contribution"], ) async def get_custom_report_builder_meta( self, user: UserRegistration, company_id: str ) -> dict: await self._get_user_company(user, company_id) return self.get_custom_report_meta() def get_custom_report_meta(self) -> dict: fields = [ {"key": CustomReportField.DATE.value, "label": "Date", "groupable": True}, { "key": CustomReportField.TRANSACTION_TYPE.value, "label": "Transaction Type", "groupable": True, }, { "key": CustomReportField.PARTY_NAME.value, "label": "Party Name", "groupable": True, }, { "key": CustomReportField.WAREHOUSE_NAME.value, "label": "Warehouse", "groupable": True, }, { "key": CustomReportField.PAYMENT_STATUS.value, "label": "Payment Status", "groupable": True, }, { "key": CustomReportField.REFERENCE_NO.value, "label": "Reference No.", "groupable": False, }, { "key": CustomReportField.REFERENCE_TYPE.value, "label": "Reference Type", "groupable": True, }, ] metrics = [ { "key": CustomReportMetric.TOTAL_AMOUNT.value, "label": "Total Amount (Rs.)", }, {"key": CustomReportMetric.QUANTITY.value, "label": "Quantity"}, { "key": CustomReportMetric.TRANSACTIONS.value, "label": "Total Transactions", }, {"key": CustomReportMetric.AVG_AMOUNT.value, "label": "Avg. Amount (Rs.)"}, ] templates = [ { "key": "high_value_transactions", "name": "High Value Transactions", "description": "Transactions grouped by type and party, sorted by amount", "configuration": CustomReportConfig( module=CustomReportModule.TRANSACTIONS, fields=[ CustomReportField.TRANSACTION_TYPE, CustomReportField.PARTY_NAME, ], metrics=[ CustomReportMetric.TRANSACTIONS, CustomReportMetric.QUANTITY, CustomReportMetric.TOTAL_AMOUNT, CustomReportMetric.AVG_AMOUNT, ], group_by=CustomReportField.TRANSACTION_TYPE, then_by=CustomReportField.PARTY_NAME, sort_by=CustomReportSort.TOTAL_AMOUNT, sort_dir="desc", ).model_dump(mode="json"), }, { "key": "sales_summary", "name": "Sales Summary", "description": "Sales invoices and receipts by party", "configuration": CustomReportConfig( module=CustomReportModule.TRANSACTIONS, fields=[ CustomReportField.TRANSACTION_TYPE, CustomReportField.PARTY_NAME, ], metrics=[ CustomReportMetric.TRANSACTIONS, CustomReportMetric.TOTAL_AMOUNT, ], transaction_types=[ TxnReportType.SALES_INVOICE, TxnReportType.PAYMENT_RECEIVED, ], group_by=CustomReportField.PARTY_NAME, sort_by=CustomReportSort.TOTAL_AMOUNT, sort_dir="desc", ).model_dump(mode="json"), }, { "key": "inventory_movement", "name": "Inventory Movement", "description": "Stock transfers and related inventory documents", "configuration": CustomReportConfig( module=CustomReportModule.TRANSACTIONS, fields=[ CustomReportField.TRANSACTION_TYPE, CustomReportField.WAREHOUSE_NAME, ], metrics=[ CustomReportMetric.TRANSACTIONS, CustomReportMetric.TOTAL_AMOUNT, ], transaction_types=[ TxnReportType.STOCK_TRANSFER, TxnReportType.PURCHASE_RETURN, TxnReportType.SALES_RETURN, ], group_by=CustomReportField.TRANSACTION_TYPE, then_by=CustomReportField.WAREHOUSE_NAME, sort_by=CustomReportSort.TRANSACTIONS, sort_dir="desc", ).model_dump(mode="json"), }, ] return { "modules": [ {"key": CustomReportModule.TRANSACTIONS.value, "label": "Transactions"} ], "fields": fields, "metrics": metrics, "templates": templates, } async def run_custom_report( self, user: UserRegistration, company_id: str, configuration: CustomReportConfig, *, page: int = 1, page_size: int = 50, ) -> CustomReportResult: await self._get_user_company(user, company_id) if configuration.module != CustomReportModule.TRANSACTIONS: raise ValidationError("Only module=transactions is supported currently") if not configuration.from_date or not configuration.to_date: raise ValidationError("from_date and to_date are required to run the report") if configuration.to_date < configuration.from_date: raise ValidationError("to_date must be on or after from_date") if configuration.sort_dir not in ("asc", "desc"): raise ValidationError("sort_dir must be asc or desc") _ = configuration.department_id data = await self._reports.run_custom_report_data( company_id, from_date=configuration.from_date, to_date=configuration.to_date, transaction_types=[t.value for t in configuration.transaction_types] or None, warehouse_id=configuration.warehouse_id, payment_status=( configuration.payment_status.value if configuration.payment_status else None ), party=configuration.party, group_by=configuration.group_by.value, then_by=configuration.then_by.value if configuration.then_by else None, sort_by=configuration.sort_by.value, sort_dir=configuration.sort_dir, page=page, page_size=page_size, ) return CustomReportResult( module=configuration.module, configuration=configuration, rows=data["rows"], totals=data["totals"], rows_total=data["rows_total"], page=data["page"], page_size=data["page_size"], total_pages=data["total_pages"], trend=data["trend"], top_parties=data["top_parties"], by_type=data["by_type"], by_payment_status=data["by_payment_status"], ) async def create_custom_report( self, user: UserRegistration, company_id: str, name: str, configuration: CustomReportConfig, *, description: str | None = None, is_template: bool = False, ) -> CustomReport: await self._get_user_company(user, company_id) existing = await self._reports.list_custom_reports( company_id, search=name, limit=50 ) if any(r.name.lower() == name.strip().lower() for r in existing): raise ConflictError(f"Custom report '{name}' already exists") now = datetime.utcnow() report = CustomReport( company_id=company_id, name=name.strip(), module=configuration.module, description=description, configuration=configuration, is_template=is_template, created_by=user.id, created_at=now, updated_at=now, ) return await self._reports.create_custom_report(report) async def list_custom_reports( self, user: UserRegistration, company_id: str, *, module: CustomReportModule | None = None, search: str | None = None, page: int = 1, page_size: int = 50, ) -> tuple[list[CustomReport], int]: await self._get_user_company(user, company_id) skip = (page - 1) * page_size items = await self._reports.list_custom_reports( company_id, module=module.value if module else None, search=search, skip=skip, limit=page_size, ) total = await self._reports.count_custom_reports( company_id, module=module.value if module else None, search=search, ) return items, total async def get_custom_report( self, user: UserRegistration, company_id: str, report_id: str ) -> CustomReport: await self._get_user_company(user, company_id) report = await self._reports.get_custom_report(report_id, company_id) if not report: raise NotFoundError("Custom report not found") return report async def update_custom_report( self, user: UserRegistration, company_id: str, report_id: str, *, name: str | None = None, description: str | None = None, configuration: CustomReportConfig | None = None, is_template: bool | None = None, ) -> CustomReport: await self._get_user_company(user, company_id) report = await self._reports.get_custom_report(report_id, company_id) if not report: raise NotFoundError("Custom report not found") if name is not None: report.name = name.strip() if description is not None: report.description = description if configuration is not None: report.configuration = configuration report.module = configuration.module if is_template is not None: report.is_template = is_template report.updated_at = datetime.utcnow() updated = await self._reports.update_custom_report(report_id, report) if not updated: raise NotFoundError("Custom report not found") return updated async def delete_custom_report( self, user: UserRegistration, company_id: str, report_id: str ) -> None: await self._get_user_company(user, company_id) deleted = await self._reports.delete_custom_report(report_id, company_id) if not deleted: raise NotFoundError("Custom report not found") def _kpi_metric(self, current: Decimal | int, previous: Decimal | int) -> TxnReportKpi: cur = Decimal(str(current)) prev = Decimal(str(previous)) change = None if prev != 0: change = ((cur - prev) / prev * Decimal("100")).quantize(Decimal("0.1")) return TxnReportKpi(current=cur, previous=prev, change_percent=change) def _previous_period(self, from_date: date, to_date: date) -> tuple[date, date]: days = (to_date - from_date).days + 1 previous_to = from_date - timedelta(days=1) previous_from = previous_to - timedelta(days=days - 1) return previous_from, previous_to def _fill_txn_trend( self, from_date: date, to_date: date, rows: list, granularity: TxnReportTrendGranularity, ) -> list: if granularity != TxnReportTrendGranularity.DAILY: return rows by_date = {row.date: row for row in rows} points = [] cursor = from_date while cursor <= to_date: points.append( by_date.get( cursor, TxnReportTrendPoint( date=cursor, total_amount=Decimal("0.00"), count=0 ), ) ) cursor += timedelta(days=1) return points async def _get_user_company(self, user: UserRegistration, company_id: str) -> Company: return await resolve_company_for_user(self._companies, user, company_id) async def _compute_net_income( self, company_id: str, accounts, from_date: date, to_date: date, ) -> Decimal: total_revenue = Decimal("0.00") total_expense = Decimal("0.00") for account in accounts: if account.is_group: continue if account.account_type == AccountType.REVENUE: total_revenue += await self._get_period_movement( company_id, account.id or "", from_date, to_date, account ) elif account.account_type == AccountType.EXPENSE: total_expense += await self._get_period_movement( company_id, account.id or "", from_date, to_date, account ) return (total_revenue - total_expense).quantize(Decimal("0.01")) async def _get_account_balance_as_of( self, company_id: str, account_id: str, as_of_date: date, account, ) -> Decimal: entries = await self._vouchers.get_ledger_entries(company_id, account_id, None, as_of_date) if entries: return entries[-1].running_balance return account.opening_balance async def _get_period_movement( self, company_id: str, account_id: str, from_date: date, to_date: date, account, ) -> Decimal: entries = await self._vouchers.get_ledger_entries( company_id, account_id, from_date, to_date ) if not entries: return Decimal("0.00") movement = Decimal("0.00") for entry in entries: if account.nature == AccountNature.DEBIT: movement += entry.debit_amount - entry.credit_amount else: movement += entry.credit_amount - entry.debit_amount return movement