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

from collections import defaultdict
from datetime import date, timedelta
from decimal import Decimal
from typing import Any

from sqlalchemy import func, select

from app.core.config import get_settings
from app.core.constants import ActorValue
from app.modules.ai_agent.constants import AIProviderName, AIResponseKey
from app.modules.business.constants import (
    CashFlowDirection,
    CashFlowType,
    DataQualityStatus,
    SourceSystem,
    StatusValue,
)
from app.modules.business.models import (
    Project,
    ProjectCashFlow,
    ProjectContract,
    ProjectMilestone,
    RiskEvent,
    SourceSyncCursor,
    WorkTask,
)
from app.modules.reports.constants import ReportResponseKey, ReportTitle


ZERO = Decimal("0")
FINANCE_HORIZONS = (7, 30, 90)


class FinanceNeedsReportMixin:
    def project_finance_needs_report(
        self,
        project_code: str | None = None,
        department: str | None = None,
        owner: str | None = None,
        as_of: date | None = None,
        include_ai: bool = False,
        actor: str = ActorValue.API,
    ) -> dict[str, Any]:
        reference = as_of or date.today()
        project_stmt = select(Project).where(
            Project.source_system == SourceSystem.LEGACY_MYSQL,
            Project.is_active.is_(True),
        )
        if project_code:
            project_stmt = project_stmt.where(
                (Project.code == project_code) | (Project.display_code == project_code)
            )
        if department:
            project_stmt = project_stmt.where(Project.department_name == department)
        if owner:
            project_stmt = project_stmt.where(Project.owner == owner)
        projects = list(self.db.execute(project_stmt).scalars())
        codes = {project.code for project in projects}

        contracts = list(
            self.db.execute(
                select(ProjectContract).where(
                    ProjectContract.is_active.is_(True),
                    ProjectContract.data_quality_status == DataQualityStatus.VALID,
                    ProjectContract.project_code.in_(codes or {""}),
                )
            ).scalars()
        )
        flows = list(
            self.db.execute(
                select(ProjectCashFlow).where(
                    ProjectCashFlow.is_active.is_(True),
                    ProjectCashFlow.project_code.in_(codes or {""}),
                    ProjectCashFlow.data_quality_status.notin_(
                        {
                            DataQualityStatus.ORPHAN_CONTRACT,
                            DataQualityStatus.ORPHAN_PROJECT,
                            DataQualityStatus.DELETED_PROJECT,
                        }
                    ),
                )
            ).scalars()
        )
        contract_amounts: dict[str, Decimal] = defaultdict(lambda: ZERO)
        for contract in contracts:
            if contract.project_code:
                contract_amounts[contract.project_code] += contract.amount or ZERO
        flows_by_project: dict[str, list[ProjectCashFlow]] = defaultdict(list)
        for flow in flows:
            if flow.project_code:
                flows_by_project[flow.project_code].append(flow)

        delivery_risk_codes = self._finance_delivery_risk_codes(codes, reference)
        items = [
            self._finance_project_item(
                project,
                contract_amounts.get(project.code, ZERO),
                flows_by_project.get(project.code, []),
                reference,
                project.code in delivery_risk_codes,
            )
            for project in projects
        ]
        attention = sorted(
            [item for item in items if item["finance_covered"]],
            key=lambda item: (
                not item["uncovered_pending_outflow"],
                not item["delivery_risk"],
                -item["funding_need"]["30"]["lower"],
                item["project_code"],
            ),
        )
        summary = self._finance_summary(items)
        quality = self._finance_data_quality(codes if project_code or department else None)
        latest_sync = self._finance_latest_sync_at()
        lines = self._finance_lines(reference, summary, quality, attention)
        chart_data = {
            "as_of": reference.isoformat(),
            "receivables": {
                "overdue": summary["overdue_receivable"],
                "due_30d": summary["horizons"]["30"]["receivable_due"],
            },
            "cashflows": {
                "confirmed_inflow": summary["confirmed_inflow"],
                "confirmed_outflow": summary["confirmed_outflow"],
                "pending_outflow": summary["pending_outflow"],
            },
            "top_projects": [
                {
                    "name": item["project_name"][:18],
                    "amount": item["funding_need"]["30"]["upper"],
                }
                for item in attention[:5]
                if item["funding_need"]["30"]["upper"] > 0
            ],
        } if summary["data_available"] else None
        report: dict[str, Any] = {
            ReportResponseKey.TITLE: ReportTitle.PROJECT_FINANCE_NEEDS,
            "as_of": reference.isoformat(),
            "currency": "CNY",
            "source_last_sync_at": latest_sync,
            "summary": summary,
            "data_quality": quality,
            "items": items,
            "attention": attention[:10],
            ReportResponseKey.LINES: lines,
            ReportResponseKey.CONTENT: "\n".join(lines),
            "finance_chart_data": chart_data,
            "disclaimer": "项目资金安排需求不包含公司账户余额，不代表真实融资缺口。",
        }
        ai_analysis = self._finance_ai_analysis(report, actor) if include_ai else None
        report["ai_analysis"] = ai_analysis
        if ai_analysis and ai_analysis.get(AIResponseKey.OK):
            answer = str(ai_analysis.get(AIResponseKey.ANSWER) or "")[:3000]
            lines.append("- AI 资金分析：")
            lines.extend(f"  {line}" for line in answer.splitlines() if line.strip())
            report[ReportResponseKey.CONTENT] = "\n".join(lines)
        return report

    def _finance_project_item(
        self,
        project: Project,
        contract_amount: Decimal,
        flows: list[ProjectCashFlow],
        reference: date,
        delivery_risk: bool,
    ) -> dict[str, Any]:
        if not contract_amount and project.source_contract_amount:
            contract_amount = project.source_contract_amount
        actual_receipt = ZERO
        overdue = ZERO
        confirmed_inflow = ZERO
        confirmed_outflow = ZERO
        pending_outflow = ZERO
        receivable_by_horizon = {days: ZERO for days in FINANCE_HORIZONS}
        for flow in flows:
            actual = flow.actual_amount or ZERO
            planned = flow.planned_amount or ZERO
            if flow.flow_type == CashFlowType.CONTRACT_RECEIVABLE:
                actual_receipt += max(actual, ZERO)
                confirmed_inflow += max(actual, ZERO)
                if flow.payment_status == "N":
                    outstanding = max(planned - max(actual, ZERO), ZERO)
                    if flow.planned_date and flow.planned_date < reference:
                        overdue += outstanding
                    elif flow.planned_date:
                        for days in FINANCE_HORIZONS:
                            if flow.planned_date <= reference + timedelta(days=days):
                                receivable_by_horizon[days] += outstanding
            elif flow.flow_type == CashFlowType.PROJECT_FUND:
                if flow.confirmation_status == "Y" and flow.approval_status == "1":
                    if flow.direction == CashFlowDirection.INFLOW:
                        confirmed_inflow += actual
                    else:
                        confirmed_outflow += actual
                elif (
                    flow.direction == CashFlowDirection.OUTFLOW
                    and flow.approval_status == "1"
                    and flow.confirmation_status == "N"
                ):
                    pending_outflow += planned
        funding_need = {
            str(days): {
                "lower": _amount(max(pending_outflow - receivable_by_horizon[days], ZERO)),
                "upper": _amount(pending_outflow),
            }
            for days in FINANCE_HORIZONS
        }
        covered = bool(contract_amount or flows)
        if not covered:
            return {
                "project_code": project.code,
                "display_code": project.display_code,
                "project_name": project.name,
                "department": project.department_name,
                "owner": project.owner,
                "stage": project.source_stage_label or project.source_stage,
                "contract_revenue": None,
                "project_investment_context": (
                    _amount(project.source_project_investment_amount)
                    if project.source_project_investment_amount is not None
                    else None
                ),
                "actual_receipt": None,
                "overdue_receivable": None,
                "confirmed_inflow": None,
                "confirmed_outflow": None,
                "pending_outflow": None,
                "receivable_due": {str(days): None for days in FINANCE_HORIZONS},
                "funding_need": {
                    str(days): {"lower": None, "upper": None} for days in FINANCE_HORIZONS
                },
                "uncovered_pending_outflow": False,
                "delivery_risk": delivery_risk,
                "finance_covered": False,
            }
        return {
            "project_code": project.code,
            "display_code": project.display_code,
            "project_name": project.name,
            "department": project.department_name,
            "owner": project.owner,
            "stage": project.source_stage_label or project.source_stage,
            "contract_revenue": _amount(contract_amount),
            "project_investment_context": _amount(
                project.source_project_investment_amount or ZERO
            ),
            "actual_receipt": _amount(actual_receipt),
            "overdue_receivable": _amount(overdue),
            "confirmed_inflow": _amount(confirmed_inflow),
            "confirmed_outflow": _amount(confirmed_outflow),
            "pending_outflow": _amount(pending_outflow),
            "receivable_due": {
                str(days): _amount(receivable_by_horizon[days]) for days in FINANCE_HORIZONS
            },
            "funding_need": funding_need,
            "uncovered_pending_outflow": funding_need["30"]["lower"] > 0,
            "delivery_risk": delivery_risk,
            "finance_covered": True,
        }

    def _finance_summary(self, items: list[dict[str, Any]]) -> dict[str, Any]:
        result = {
            "projects_total": len(items),
            "projects_covered": sum(1 for item in items if item["finance_covered"]),
            "contract_revenue": _sum_items(items, "contract_revenue"),
            "actual_receipt": _sum_items(items, "actual_receipt"),
            "overdue_receivable": _sum_items(items, "overdue_receivable"),
            "confirmed_inflow": _sum_items(items, "confirmed_inflow"),
            "confirmed_outflow": _sum_items(items, "confirmed_outflow"),
            "pending_outflow": _sum_items(items, "pending_outflow"),
            "horizons": {},
        }
        result["coverage_rate"] = round(
            result["projects_covered"] * 100 / result["projects_total"], 2
        ) if result["projects_total"] else 0.0
        result["data_available"] = result["projects_covered"] > 0
        result["horizons"] = {
            str(days): {
                "receivable_due": round(
                    sum(
                        item["receivable_due"][str(days)] or 0
                        for item in items
                    ), 2
                ),
                "funding_need_lower": round(
                    sum(
                        item["funding_need"][str(days)]["lower"] or 0
                        for item in items
                    ), 2
                ),
                "funding_need_upper": round(
                    sum(
                        item["funding_need"][str(days)]["upper"] or 0
                        for item in items
                    ), 2
                ),
            }
            for days in FINANCE_HORIZONS
        }
        if not result["data_available"]:
            for key in (
                "contract_revenue",
                "actual_receipt",
                "overdue_receivable",
                "confirmed_inflow",
                "confirmed_outflow",
                "pending_outflow",
            ):
                result[key] = None
            result["horizons"] = {
                str(days): {
                    "receivable_due": None,
                    "funding_need_lower": None,
                    "funding_need_upper": None,
                }
                for days in FINANCE_HORIZONS
            }
        return result

    def _finance_data_quality(self, project_codes: set[str] | None) -> dict[str, int]:
        conditions: list[Any] = [ProjectCashFlow.is_active.is_(True)]
        if project_codes is not None:
            conditions.append(ProjectCashFlow.project_code.in_(project_codes or {""}))
        rows = self.db.execute(
            select(
                ProjectCashFlow.flow_type,
                ProjectCashFlow.data_quality_status,
                func.count(),
            )
            .where(*conditions)
            .group_by(ProjectCashFlow.flow_type, ProjectCashFlow.data_quality_status)
        ).all()
        counts = {
            (str(flow_type), str(quality_status)): int(count)
            for flow_type, quality_status, count in rows
        }
        by_status: dict[str, int] = defaultdict(int)
        for (_, quality_status), count in counts.items():
            by_status[quality_status] += count
        return {
            "orphan_contracts": counts.get(
                (CashFlowType.CONTRACT_RECEIVABLE, DataQualityStatus.ORPHAN_CONTRACT), 0
            ),
            "orphan_receivable_projects": counts.get(
                (CashFlowType.CONTRACT_RECEIVABLE, DataQualityStatus.ORPHAN_PROJECT), 0
            ),
            "orphan_funds": counts.get(
                (CashFlowType.PROJECT_FUND, DataQualityStatus.ORPHAN_PROJECT), 0
            ),
            "deleted_projects": by_status[DataQualityStatus.DELETED_PROJECT],
            "paid_amount_missing": by_status[DataQualityStatus.PAID_AMOUNT_MISSING],
            "status_amount_mismatch": by_status[DataQualityStatus.STATUS_AMOUNT_MISMATCH],
        }

    def _finance_delivery_risk_codes(self, codes: set[str], reference: date) -> set[str]:
        if not codes:
            return set()
        milestone_codes = set(
            self.db.execute(
                select(ProjectMilestone.project_code).where(
                    ProjectMilestone.project_code.in_(codes),
                    ProjectMilestone.is_active.is_(True),
                    ProjectMilestone.is_overdue.is_(True),
                )
            ).scalars()
        )
        task_codes = set(
            self.db.execute(
                select(WorkTask.project_code).where(
                    WorkTask.project_code.in_(codes),
                    WorkTask.is_active.is_(True),
                    WorkTask.status != StatusValue.COMPLETED,
                    WorkTask.due_date.is_not(None),
                    WorkTask.due_date < reference,
                )
            ).scalars()
        )
        event_codes = set(
            self.db.execute(
                select(RiskEvent.project_code).where(
                    RiskEvent.project_code.in_(codes),
                    RiskEvent.source_domain == "intasect_project_event",
                    RiskEvent.status == StatusValue.OPEN,
                )
            ).scalars()
        )
        return {str(code) for code in milestone_codes | task_codes | event_codes if code}

    def _finance_latest_sync_at(self) -> str | None:
        datasets = {"projects", "contracts", "contract_receivables", "project_funds"}
        cursors = list(
            self.db.execute(
                select(SourceSyncCursor).where(SourceSyncCursor.dataset.in_(datasets))
            ).scalars()
        )
        if (
            len(cursors) != len(datasets)
            or any(cursor.status != StatusValue.COMPLETED for cursor in cursors)
            or any(cursor.last_success_at is None for cursor in cursors)
        ):
            return None
        value = min(cursor.last_success_at for cursor in cursors if cursor.last_success_at)
        return value.isoformat()

    def _finance_lines(
        self,
        reference: date,
        summary: dict[str, Any],
        quality: dict[str, int],
        attention: list[dict[str, Any]],
    ) -> list[str]:
        horizon = summary["horizons"]["30"]
        if not summary["data_available"]:
            return [
                f"- 资金分析基准日：{reference.isoformat()}（人民币元）",
                f"- 财务覆盖：0/{summary['projects_total']} 个项目。",
                "- 项目财务数据未接入或无有效记录，金额不按零值解释。",
                "- 注意：源库没有公司账户余额，不能计算真实融资缺口。",
            ]
        lines = [
            f"- 资金分析基准日：{reference.isoformat()}（人民币元）",
            f"- 财务覆盖：{summary['projects_covered']}/{summary['projects_total']} 个项目，"
            f"覆盖率 {summary['coverage_rate']}%",
            f"- 逾期应收：{_money(summary['overdue_receivable'])}，"
            f"未来30天应收：{_money(horizon['receivable_due'])}",
            f"- 待确认支出：{_money(summary['pending_outflow'])}，"
            f"已确认流入/流出：{_money(summary['confirmed_inflow'])}/"
            f"{_money(summary['confirmed_outflow'])}",
            f"- 未来30天项目资金安排需求：{_money(horizon['funding_need_lower'])}"
            f" 至 {_money(horizon['funding_need_upper'])}",
            "- 注意：该区间不包含公司账户余额，不代表真实融资缺口。",
            f"- 数据异常：孤儿合同付款 {quality['orphan_contracts']}，"
            f"孤儿资金记录 {quality['orphan_funds']}，"
            f"已支付缺金额 {quality['paid_amount_missing']}，"
            f"状态金额不一致 {quality['status_amount_mismatch']}",
        ]
        for item in attention[:5]:
            need = item["funding_need"]["30"]
            if need["upper"] <= 0 and item["overdue_receivable"] <= 0:
                continue
            lines.append(
                f"  - {item['project_name']}（{item['display_code'] or item['project_code']}）："
                f"资金安排 {_money(need['lower'])}-{_money(need['upper'])}，"
                f"逾期应收 {_money(item['overdue_receivable'])}"
            )
        return lines

    def _finance_ai_analysis(self, report: dict[str, Any], actor: str) -> dict[str, Any]:
        from app.modules.ai_agent.service import AIService
        from app.modules.ai_agent.skills import AISkillId

        context = {
            "as_of": report["as_of"],
            "currency": report["currency"],
            "summary": report["summary"],
            "data_quality": report["data_quality"],
            "attention": [
                {
                    "project_code": item["project_code"],
                    "display_code": item["display_code"],
                    "project_name": item["project_name"],
                    "stage": item["stage"],
                    "overdue_receivable": item["overdue_receivable"],
                    "pending_outflow": item["pending_outflow"],
                    "receivable_due_30d": item["receivable_due"]["30"],
                    "funding_need_30d": item["funding_need"]["30"],
                    "delivery_risk": item["delivery_risk"],
                }
                for item in report["attention"][:10]
            ],
            "disclaimer": report["disclaimer"],
        }
        last_error: Exception | None = None
        attempts = max(1, min(get_settings().ai_analysis_max_attempts, 5))
        for _ in range(attempts):
            try:
                result = AIService(self.db).run_skill(
                    AISkillId.PROJECT_FINANCE_NEEDS_ANALYSIS,
                    context=context,
                    actor=actor,
                )
                if result.get(AIResponseKey.PROVIDER) == AIProviderName.NOOP:
                    return {AIResponseKey.OK: False, AIResponseKey.ERROR: "AI unavailable"}
                return {AIResponseKey.OK: True, **result}
            except Exception as exc:
                last_error = exc
                self.db.rollback()
        return {
            AIResponseKey.OK: False,
            AIResponseKey.ERROR: str(last_error) if last_error else "AI unavailable",
            AIResponseKey.TYPE: type(last_error).__name__ if last_error else "AIUnavailable",
            "attempts": attempts,
        }


def _amount(value: Decimal) -> float:
    return round(float(value), 2)


def _sum_items(items: list[dict[str, Any]], key: str) -> float:
    return round(sum(float(item[key]) for item in items if item[key] is not None), 2)


def _money(value: float) -> str:
    return f"¥{value:,.2f}"