company-ai-platform/app/modules/reports/services/summaries.py

from datetime import date
from decimal import Decimal
from typing import Any

from sqlalchemy import func, select

from app.modules.business.constants import (
    ATTENDANCE_ABNORMAL_STATUSES,
    PROJECT_CLOSED_STATUSES,
    PENDING_APPROVAL_STATUSES,
    StatusValue,
)
from app.modules.business.models import (
    AttendanceRecord,
    Expense,
    FundAccount,
    Procurement,
    Project,
    WorkTask,
)
from app.modules.reports.constants import (
    ReportPushType,
    ReportResponseKey,
    ReportTitle,
)
from app.modules.risk.constants import RiskSummaryKey

from app.modules.reports.services.common import _money


class ReportSummaryMixin:
    def daily_brief(self) -> dict:
        project_count = self._count(Project)
        task_count = self._count(WorkTask)
        procurement_pending = self._count(
            Procurement,
            Procurement.approval_status.in_(PENDING_APPROVAL_STATUSES),
        )
        expense_pending = self._count(
            Expense,
            Expense.approval_status.in_(PENDING_APPROVAL_STATUSES),
        )
        fund_total = (
            self.db.execute(select(func.sum(FundAccount.current_balance))).scalar()
            or Decimal("0")
        )
        risk_summary = self.risks.summary()
        attendance = self.attendance_summary()
        lines = [
            f"- 项目总数：{project_count}",
            f"- 任务总数：{task_count}",
            f"- 待处理采购：{procurement_pending}",
            f"- 待处理费用：{expense_pending}",
            f"- 当前账户总余额：{_money(fund_total)}",
            (
                f"- 今日打卡记录：{attendance[ReportResponseKey.TOTAL]}，"
                f"异常：{attendance[ReportResponseKey.ABNORMAL_TOTAL]}"
            ),
            f"- 逾期任务：{len(risk_summary[RiskSummaryKey.OVERDUE_TASKS])}",
            f"- 延期项目：{len(risk_summary[RiskSummaryKey.DELAYED_PROJECTS])}",
            f"- 超预算项目：{len(risk_summary[RiskSummaryKey.OVER_BUDGET_PROJECTS])}",
            f"- 资金风险账户：{len(risk_summary[RiskSummaryKey.FUND_RISKS])}",
            f"- 供应商风险：{len(risk_summary[RiskSummaryKey.SUPPLIER_RISKS])}",
            f"- 打开风险事件：{len(risk_summary[RiskSummaryKey.OPEN_EVENTS])}",
            f"- 综合风险等级：{risk_summary[RiskSummaryKey.RISK_LEVEL]}",
        ]
        return {
            ReportResponseKey.TITLE: ReportTitle.DAILY_BRIEF,
            ReportResponseKey.REPORT_TYPE: ReportPushType.DAILY_BRIEF,
            ReportResponseKey.LINES: lines,
            ReportResponseKey.CONTENT: "\n".join(lines),
        }

    def project_weekly(self) -> dict:
        active = self._count(
            Project,
            Project.status.notin_(PROJECT_CLOSED_STATUSES),
        )
        delayed = self.risks.delayed_projects()
        over_budget = self.risks.over_budget_projects()
        open_risks = self.risks.list_events(status_filter=StatusValue.OPEN)
        lines = [
            f"- 活跃项目：{active}",
            f"- 延期项目：{len(delayed)}",
            f"- 超预算项目：{len(over_budget)}",
            f"- 打开风险事件：{len(open_risks)}",
            "- 需要管理层关注：",
        ]
        for item in delayed[:10]:
            lines.append(
                f"  - 延期：{item.get('code')} {item.get('name')}，"
                f"负责人 {item.get('owner')}"
            )
        for item in over_budget[:10]:
            lines.append(f"  - 超预算：{item.get('code')} {item.get('name')}")
        return {
            ReportResponseKey.TITLE: ReportTitle.PROJECT_WEEKLY,
            ReportResponseKey.REPORT_TYPE: ReportPushType.PROJECT_WEEKLY,
            ReportResponseKey.LINES: lines,
            ReportResponseKey.CONTENT: "\n".join(lines),
        }

    def risk_progress(self) -> dict[str, Any]:
        """Build a standalone project risk progress report."""

        summary = self.risks.summary()
        lines = [
            f"- 综合风险等级：{summary[RiskSummaryKey.RISK_LEVEL]}",
            f"- 风险分：{summary[RiskSummaryKey.RISK_SCORE]}",
            f"- 逾期任务：{len(summary[RiskSummaryKey.OVERDUE_TASKS])}",
            f"- 延期项目：{len(summary[RiskSummaryKey.DELAYED_PROJECTS])}",
            f"- 超预算项目：{len(summary[RiskSummaryKey.OVER_BUDGET_PROJECTS])}",
            f"- 资金风险账户：{len(summary[RiskSummaryKey.FUND_RISKS])}",
            f"- 供应商风险：{len(summary[RiskSummaryKey.SUPPLIER_RISKS])}",
            f"- 打开风险事件：{len(summary[RiskSummaryKey.OPEN_EVENTS])}",
        ]
        return {
            ReportResponseKey.TITLE: ReportTitle.RISK_PROGRESS,
            ReportResponseKey.REPORT_TYPE: ReportPushType.RISK_PROGRESS,
            ReportResponseKey.LINES: lines,
            ReportResponseKey.CONTENT: "\n".join(lines),
        }

    def attendance_summary(self, work_date: date | None = None) -> dict[str, Any]:
        """Summarize attendance records for one business day."""

        target_date = work_date or date.today()
        rows = self.db.execute(
            select(AttendanceRecord.status, func.count())
            .where(AttendanceRecord.work_date == target_date)
            .group_by(AttendanceRecord.status)
        ).all()
        status_counts = {str(status): int(count) for status, count in rows}
        abnormal_total = sum(
            count
            for status, count in status_counts.items()
            if status in ATTENDANCE_ABNORMAL_STATUSES
        )
        total = sum(status_counts.values())
        lines = [
            f"- 日期：{target_date.isoformat()}",
            f"- 打卡记录：{total}",
            f"- 异常记录：{abnormal_total}",
        ]
        for status, count in sorted(status_counts.items()):
            lines.append(f"- {status}：{count}")
        return {
            ReportResponseKey.TITLE: ReportTitle.ATTENDANCE_SUMMARY,
            ReportResponseKey.REPORT_TYPE: ReportPushType.ATTENDANCE_SUMMARY,
            ReportResponseKey.WORK_DATE: target_date.isoformat(),
            ReportResponseKey.TOTAL: total,
            ReportResponseKey.ABNORMAL_TOTAL: abnormal_total,
            ReportResponseKey.STATUS_COUNTS: status_counts,
            ReportResponseKey.LINES: lines,
            ReportResponseKey.CONTENT: "\n".join(lines),
        }