dingdanquanliucheng/backend/app/services/report_service.py

"""报表服务模块

负责业绩统计报表的查询和导出。
支持按月、按季、按年三种统计维度，可导出为 CSV 或 XLSX 格式。
导出时自动关联附件管理和审计日志。

被调用方：reports 路由（业绩报表查询、报表导出接口）。
"""

from collections import defaultdict
from datetime import datetime

from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session

from backend.app.core.error_codes import ErrorCode
from backend.app.core.exceptions import AppException
from backend.app.repositories.report_repository import ReportRepository
from backend.app.services.audit_service import audit_service
from backend.app.services.export_service import export_service
from backend.app.services.file_service import file_service
from backend.app.services.storage_service import storage_service


class ReportService:
    """业绩报表业务服务。

    依赖：
    - ReportRepository：报表数据查询
    - ExportService：构建导出文件
    - StorageService：文件上传发布
    - FileService：附件记录保存
    - AuditService：操作审计日志
    """

    def __init__(self) -> None:
        self.repository = ReportRepository()

    def performance_report(self, filters: dict, session: Session | None = None) -> dict:
        """查询业绩统计报表数据。

        根据统计类型（月/季/年）和筛选条件，聚合订单行数据生成报表。

        参数：
            filters: 筛选条件字典，包含 stat_type、start_date、end_date、category_id、export_format 等。
            session: 数据库会话，不可为 None。

        返回：
            包含 stat_type 和 list（各统计周期的汇总数据）的字典。

        被调用方：reports 路由（业绩报表查询接口）。
        """
        stat_type = self._normalize_stat_type(filters.get("stat_type"))
        parsed_filters = self._parse_filters(filters, stat_type)

        if session is not None:
            try:
                rows = self.repository.list_performance_rows(session, parsed_filters)
                return {
                    "stat_type": stat_type,
                    "list": self._build_report_list(rows, stat_type),
                }
            except SQLAlchemyError as exc:
                raise AppException(code=ErrorCode.INTERNAL_ERROR, message="数据库异常", status_code=500) from exc
        raise AppException(code=ErrorCode.INTERNAL_ERROR, message="数据库连接不可用", status_code=500)

    def export_performance_report(self, filters: dict, session: Session | None = None) -> dict:
        """导出业绩统计报表为文件（CSV 或 XLSX）。

        先查询报表数据，再调用 ExportService 生成文件，最后上传到存储服务并记录附件和审计日志。

        参数：
            filters: 筛选条件字典，额外需要 export_format 字段。
            session: 数据库会话，不可为 None。

        返回：
            包含 file_url、file_name、object_key 等文件信息及报表元数据的字典。

        被调用方：reports 路由（报表导出接口）。
        """
        report = self.performance_report(filters, session)
        export_format = self._normalize_export_format(filters.get("export_format"))
        export_meta = export_service.build_performance_export(report, export_format)
        publish_meta = storage_service.publish_local_file(export_meta["file_path"], export_meta["object_key"])
        attachment_payload = {
            "biz_type": "performance_report",
            "biz_id": int(datetime.now().strftime("%Y%m%d%H%M%S")),
            "file_name": export_meta["file_name"],
            "file_url": publish_meta["file_url"],
            "file_type": export_meta["content_type"],
            "file_size": export_meta["file_size"],
        }

        attachment_result = file_service.save_attachment(attachment_payload, session) if session is not None else {
            "file_url": publish_meta["file_url"],
            "file_name": export_meta["file_name"],
            "attachment_id": 0,
            "object_key": publish_meta["object_key"],
        }

        if session is not None:
            audit_service.write_log(
                session,
                {
                    "operate_type": "report_export",
                    "biz_type": "performance_report",
                    "biz_id": attachment_result.get("attachment_id", 0),
                    "before_value": None,
                    "after_value": {
                        "file_name": export_meta["file_name"],
                        "stat_type": report["stat_type"],
                        "export_format": export_format,
                        "total_periods": len(report["list"]),
                    },
                    "remark": "导出业绩统计报表",
                },
            )
            session.commit()

        return {
            "file_url": attachment_result["file_url"],
            "file_name": export_meta["file_name"],
            "object_key": publish_meta["object_key"],
            "storage_provider": publish_meta["storage_provider"],
            "bucket_name": publish_meta["bucket_name"],
            "stat_type": report["stat_type"],
            "export_format": export_format,
            "total_periods": len(report["list"]),
            "filters": {
                "start_date": filters.get("start_date"),
                "end_date": filters.get("end_date"),
                "category_id": filters.get("category_id"),
            },
        }

    def _normalize_stat_type(self, stat_type: str | None) -> str:
        """校验并归一化统计类型参数。

        参数：
            stat_type: 原始统计类型值，支持 'month'、'quarter'、'year'，默认 'month'。

        返回：
            归一化后的统计类型字符串。

        异常：
            参数不合法时抛出 AppException（PARAM_ERROR）。
        """
        value = (stat_type or "month").strip().lower()
        if value not in {"month", "quarter", "year"}:
            raise AppException(code=ErrorCode.PARAM_ERROR, message="统计类型不支持", status_code=400)
        return value

    def _normalize_export_format(self, export_format: str | None) -> str:
        """校验并归一化导出格式参数。

        参数：
            export_format: 原始导出格式值，支持 'csv'、'xlsx'，默认 'csv'。

        返回：
            归一化后的导出格式字符串。
        """
        value = (export_format or "csv").strip().lower()
        if value not in {"csv", "xlsx"}:
            raise AppException(code=ErrorCode.PARAM_ERROR, message="导出格式仅支持 csv 或 xlsx", status_code=400)
        return value

    def _parse_filters(self, filters: dict, stat_type: str) -> dict:
        """解析前端传入的筛选条件为内部可用的过滤参数。

        参数：
            filters: 原始筛选字典。
            stat_type: 已归一化的统计类型。

        返回：
            包含 stat_type、start_date、end_date、category_id、exclude_ecommerce 的字典。
        """
        parsed = {
            "stat_type": stat_type,
            "start_date": self._parse_date(filters.get("start_date"), "开始日期格式错误"),
            "end_date": self._parse_date(filters.get("end_date"), "结束日期格式错误"),
            "category_id": filters.get("category_id"),
            "exclude_ecommerce": bool(filters.get("exclude_ecommerce")),
        }
        if parsed["start_date"] and parsed["end_date"] and parsed["start_date"] > parsed["end_date"]:
            raise AppException(code=ErrorCode.PARAM_ERROR, message="开始日期不能晚于结束日期", status_code=400)
        return parsed

    def _parse_date(self, value: str | None, error_message: str) -> datetime | None:
        """将日期字符串解析为 datetime 对象。

        参数：
            value: 日期字符串，格式为 'YYYY-MM-DD'，空值时返回 None。
            error_message: 解析失败时的错误提示信息。

        返回：
            datetime 对象或 None。
        """
        if value is None or not str(value).strip():
            return None
        try:
            return datetime.strptime(str(value).strip(), "%Y-%m-%d")
        except ValueError as exc:
            raise AppException(code=ErrorCode.PARAM_ERROR, message=error_message, status_code=400) from exc

    def _build_report_list(self, rows: list, stat_type: str) -> list[dict]:
        """将原始订单行数据按统计周期分组聚合，生成报表列表。

        参数：
            rows: 数据库查询返回的订单行列表。
            stat_type: 统计类型（month/quarter/year），决定周期分组方式。

        返回：
            按时间顺序排列的统计周期汇总列表，每个元素包含 order_count、order_amount、
            category_amounts、commission_amount 等。
        """
        grouped: dict[str, dict] = {}
        period_order: list[str] = []

        for row in rows:
            created_at = row.created_at
            if created_at is None:
                continue

            stat_period = self._format_stat_period(created_at, stat_type)
            if stat_period not in grouped:
                grouped[stat_period] = {
                    "stat_period": stat_period,
                    "order_ids": set(),
                    "order_amount": 0.0,
                    "commission_by_order": {},
                    "category_amounts": defaultdict(lambda: {"category_id": 0, "category_name": "未分类", "amount": 0.0}),
                }
                period_order.append(stat_period)

            period_data = grouped[stat_period]
            line_amount = float(row.quantity or 0) * float(row.sale_price or 0)
            period_data["order_amount"] += line_amount
            period_data["order_ids"].add(row.order_id)
            if row.order_id not in period_data["commission_by_order"]:
                period_data["commission_by_order"][row.order_id] = float(row.commission_amount or 0)

            category_key = row.category_id or 0
            category_item = period_data["category_amounts"][category_key]
            category_item["category_id"] = row.category_id or 0
            category_item["category_name"] = row.category_name or "未分类"
            category_item["amount"] += line_amount

        result: list[dict] = []
        for stat_period in period_order:
            period_data = grouped[stat_period]
            category_amounts = sorted(
                (
                    {
                        "category_id": item["category_id"],
                        "category_name": item["category_name"],
                        "amount": round(item["amount"], 2),
                    }
                    for item in period_data["category_amounts"].values()
                ),
                key=lambda item: (-item["amount"], item["category_id"]),
            )
            result.append(
                {
                    "stat_period": stat_period,
                    "order_count": len(period_data["order_ids"]),
                    "order_amount": round(period_data["order_amount"], 2),
                    "category_amounts": category_amounts,
                    "commission_amount": round(sum(period_data["commission_by_order"].values()), 2),
                }
            )
        return result

    def _format_stat_period(self, dt: datetime, stat_type: str) -> str:
        """将日期按统计类型格式化为统计周期标识字符串。

        参数：
            dt: 日期时间对象。
            stat_type: 统计类型，'year' 返回 'YYYY'，'quarter' 返回 'YYYY-QN'，'month' 返回 'YYYY-MM'。

        返回：
            统计周期标识字符串。
        """
        if stat_type == "year":
            return dt.strftime("%Y")
        if stat_type == "quarter":
            quarter = ((dt.month - 1) // 3) + 1
            return f"{dt.year}-Q{quarter}"
        return dt.strftime("%Y-%m")


report_service = ReportService()