company-ai-platform/app/modules/reports/services/lifecycle/stats.py

from datetime import date
from typing import Any


from app.modules.business.constants import (
    ATTENDANCE_ABNORMAL_STATUSES,
    DONE_STATUSES,
    GENERATED_RISK_EVENT_TYPES,
    PROJECT_CLOSED_STATUSES,
    PENDING_APPROVAL_STATUSES,
    SUPPLIER_RISK_LEVELS,
    RiskLevel,
    StatusValue,
)
from app.modules.business.models import (
    AttendanceRecord,
    Expense,
    FundAccount,
    Procurement,
    Project,
    RiskEvent,
    Supplier,
    WorkTask,
)
from app.modules.reports.constants import (
    LIFECYCLE_RISK_SCORE_WEIGHTS,
    MetricKey,
)
from app.modules.reports.services.common import _rate
from app.modules.risk.constants import risk_level_for_score


class ReportLifecycleStatsMixin:
    def _lifecycle_project_stats(self, conditions: list[Any]) -> dict[str, Any]:
        total = self._count(Project, *conditions)
        active = self._count(Project, Project.status.notin_(PROJECT_CLOSED_STATUSES), *conditions)
        closed = total - active
        budget_total = self._sum(Project.budget_amount, *conditions)
        actual_total = self._sum(Project.actual_amount, *conditions)
        delayed = self._count(
            Project,
            Project.due_date.is_not(None),
            Project.due_date < date.today(),
            Project.status.notin_(PROJECT_CLOSED_STATUSES),
            *conditions,
        )
        over_budget = self._count(
            Project,
            Project.budget_amount > 0,
            Project.actual_amount > Project.budget_amount,
            *conditions,
        )
        return {
            MetricKey.TOTAL: total,
            MetricKey.ACTIVE: active,
            MetricKey.CLOSED: closed,
            MetricKey.AVERAGE_PROGRESS_PERCENT: self._avg(
                Project.progress_percent,
                *conditions,
            ),
            MetricKey.BY_STATUS: self._group_counts(Project, Project.status, *conditions),
            MetricKey.BY_RISK_LEVEL: self._group_counts(Project, Project.risk_level, *conditions),
            MetricKey.BUDGET_TOTAL: budget_total,
            MetricKey.ACTUAL_TOTAL: actual_total,
            MetricKey.BUDGET_USAGE_RATE: _rate(actual_total, budget_total),
            MetricKey.DELAYED: delayed,
            MetricKey.OVER_BUDGET: over_budget,
        }

    def _lifecycle_task_stats(self, conditions: list[Any]) -> dict[str, Any]:
        total = self._count(WorkTask, *conditions)
        completed = self._count(WorkTask, WorkTask.status.in_(DONE_STATUSES), *conditions)
        overdue = self._count(
            WorkTask,
            WorkTask.due_date.is_not(None),
            WorkTask.due_date < date.today(),
            WorkTask.status.notin_(DONE_STATUSES),
            *conditions,
        )
        blocked = self._count(
            WorkTask,
            WorkTask.blocker.is_not(None),
            WorkTask.status.notin_(DONE_STATUSES),
            *conditions,
        )
        return {
            MetricKey.TOTAL: total,
            MetricKey.COMPLETED: completed,
            MetricKey.OPEN: total - completed,
            MetricKey.OVERDUE: overdue,
            MetricKey.BLOCKED: blocked,
            MetricKey.COMPLETION_RATE: _rate(completed, total),
            MetricKey.BY_STATUS: self._group_counts(WorkTask, WorkTask.status, *conditions),
            MetricKey.BY_PRIORITY: self._group_counts(WorkTask, WorkTask.priority, *conditions),
        }

    def _lifecycle_procurement_stats(self, conditions: list[Any]) -> dict[str, Any]:
        total = self._count(Procurement, *conditions)
        pending_approval = self._count(
            Procurement,
            Procurement.approval_status.in_(PENDING_APPROVAL_STATUSES),
            *conditions,
        )
        pending_delivery = self._count(
            Procurement,
            Procurement.delivery_status.notin_(DONE_STATUSES),
            *conditions,
        )
        unpaid = self._count(
            Procurement,
            Procurement.payment_status != StatusValue.PAID,
            *conditions,
        )
        expected_total = self._sum(Procurement.expected_amount, *conditions)
        actual_total = self._sum(Procurement.actual_amount, *conditions)
        return {
            MetricKey.TOTAL: total,
            MetricKey.PENDING_APPROVAL: pending_approval,
            MetricKey.PENDING_DELIVERY: pending_delivery,
            MetricKey.UNPAID: unpaid,
            MetricKey.EXPECTED_TOTAL: expected_total,
            MetricKey.ACTUAL_TOTAL: actual_total,
            MetricKey.ACTUAL_VS_EXPECTED_RATE: _rate(actual_total, expected_total),
        }

    def _lifecycle_expense_stats(self, conditions: list[Any]) -> dict[str, Any]:
        total = self._count(Expense, *conditions)
        pending_approval = self._count(
            Expense,
            Expense.approval_status.in_(PENDING_APPROVAL_STATUSES),
            *conditions,
        )
        unpaid = self._count(Expense, Expense.payment_status != StatusValue.PAID, *conditions)
        return {
            MetricKey.TOTAL: total,
            MetricKey.PENDING_APPROVAL: pending_approval,
            MetricKey.UNPAID: unpaid,
            MetricKey.AMOUNT_TOTAL: self._sum(Expense.amount, *conditions),
            MetricKey.BY_TYPE: self._group_counts(Expense, Expense.expense_type, *conditions),
        }

    def _lifecycle_fund_stats(self) -> dict[str, Any]:
        balance = self._sum(FundAccount.current_balance)
        receivable = self._sum(FundAccount.expected_receivable)
        payable = self._sum(FundAccount.expected_payable)
        safety_line = self._sum(FundAccount.safety_line)
        risk_accounts = self._count(
            FundAccount,
            FundAccount.current_balance < FundAccount.safety_line,
        )
        return {
            MetricKey.ACCOUNTS_TOTAL: self._count(FundAccount),
            MetricKey.CURRENT_BALANCE_TOTAL: balance,
            MetricKey.EXPECTED_RECEIVABLE_TOTAL: receivable,
            MetricKey.EXPECTED_PAYABLE_TOTAL: payable,
            MetricKey.SAFETY_LINE_TOTAL: safety_line,
            MetricKey.NET_POSITION: balance + receivable - payable,
            MetricKey.RISK_ACCOUNTS: risk_accounts,
        }

    def _lifecycle_supplier_stats(self) -> dict[str, Any]:
        risky = self._count(
            Supplier,
            (Supplier.blacklist_status != StatusValue.NORMAL)
            | Supplier.risk_level.in_(SUPPLIER_RISK_LEVELS),
        )
        blacklisted = self._count(Supplier, Supplier.blacklist_status != StatusValue.NORMAL)
        return {
            MetricKey.TOTAL: self._count(Supplier),
            MetricKey.RISKY: risky,
            MetricKey.BLACKLISTED: blacklisted,
            MetricKey.BY_RISK_LEVEL: self._group_counts(Supplier, Supplier.risk_level),
        }

    def _lifecycle_attendance_stats(self, conditions: list[Any]) -> dict[str, Any]:
        total = self._count(AttendanceRecord, *conditions)
        abnormal = self._count(
            AttendanceRecord,
            AttendanceRecord.status.in_(ATTENDANCE_ABNORMAL_STATUSES),
            *conditions,
        )
        return {
            MetricKey.TOTAL: total,
            MetricKey.ABNORMAL: abnormal,
            MetricKey.ABNORMAL_RATE: _rate(abnormal, total),
            MetricKey.BY_STATUS: self._group_counts(
                AttendanceRecord,
                AttendanceRecord.status,
                *conditions,
            ),
        }

    def _lifecycle_risk_stats(
        self,
        project_conditions: list[Any],
        task_conditions: list[Any],
        risk_conditions: list[Any],
    ) -> dict[str, Any]:
        overdue_tasks = self._count(
            WorkTask,
            WorkTask.due_date.is_not(None),
            WorkTask.due_date < date.today(),
            WorkTask.status.notin_(DONE_STATUSES),
            *task_conditions,
        )
        delayed_projects = self._count(
            Project,
            Project.due_date.is_not(None),
            Project.due_date < date.today(),
            Project.status.notin_(PROJECT_CLOSED_STATUSES),
            *project_conditions,
        )
        over_budget_projects = self._count(
            Project,
            Project.budget_amount > 0,
            Project.actual_amount > Project.budget_amount,
            *project_conditions,
        )
        open_events = self._count(
            RiskEvent,
            RiskEvent.status == StatusValue.OPEN,
            *risk_conditions,
        )
        high_events = self._count(
            RiskEvent,
            RiskEvent.status == StatusValue.OPEN,
            RiskEvent.risk_level == RiskLevel.HIGH,
            *risk_conditions,
        )
        external_open_events = self._count(
            RiskEvent,
            RiskEvent.status == StatusValue.OPEN,
            RiskEvent.risk_type.notin_(GENERATED_RISK_EVENT_TYPES),
            *risk_conditions,
        )
        external_high_events = self._count(
            RiskEvent,
            RiskEvent.status == StatusValue.OPEN,
            RiskEvent.risk_level == RiskLevel.HIGH,
            RiskEvent.risk_type.notin_(GENERATED_RISK_EVENT_TYPES),
            *risk_conditions,
        )
        risk_score = (
            overdue_tasks * LIFECYCLE_RISK_SCORE_WEIGHTS[MetricKey.OVERDUE_TASKS]
            + delayed_projects * LIFECYCLE_RISK_SCORE_WEIGHTS[MetricKey.DELAYED_PROJECTS]
            + over_budget_projects * LIFECYCLE_RISK_SCORE_WEIGHTS[MetricKey.OVER_BUDGET_PROJECTS]
            + external_open_events * LIFECYCLE_RISK_SCORE_WEIGHTS[MetricKey.EXTERNAL_OPEN_EVENTS]
            + external_high_events * LIFECYCLE_RISK_SCORE_WEIGHTS[MetricKey.EXTERNAL_HIGH_EVENTS]
        )
        return {
            MetricKey.RISK_LEVEL: risk_level_for_score(risk_score),
            MetricKey.RISK_SCORE: risk_score,
            MetricKey.OVERDUE_TASKS: overdue_tasks,
            MetricKey.DELAYED_PROJECTS: delayed_projects,
            MetricKey.OVER_BUDGET_PROJECTS: over_budget_projects,
            MetricKey.OPEN_EVENTS: open_events,
            MetricKey.HIGH_EVENTS: high_events,
            MetricKey.EXTERNAL_OPEN_EVENTS: external_open_events,
            MetricKey.EXTERNAL_HIGH_EVENTS: external_high_events,
            MetricKey.EVENTS_BY_TYPE: self._group_counts(
                RiskEvent,
                RiskEvent.risk_type,
                *risk_conditions,
            ),
            MetricKey.EVENTS_BY_LEVEL: self._group_counts(
                RiskEvent,
                RiskEvent.risk_level,
                *risk_conditions,
            ),
        }