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.demo_store import demo_store from backend.app.services.export_service import export_service from backend.app.services.file_service import file_service class ReportService: def __init__(self) -> None: self.repository = ReportRepository() def performance_report(self, filters: dict, session: Session | None = None) -> dict: 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: pass return self._build_demo_report(stat_type) def export_performance_report(self, filters: dict, session: Session | None = None) -> dict: report = self.performance_report(filters, session) export_meta = export_service.build_performance_export(report) 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": export_meta["file_path"], "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": export_meta["file_path"], "file_name": export_meta["file_name"], "attachment_id": 0, } 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"], "total_periods": len(report["list"]), }, "remark": "导出业绩统计报表", }, ) session.commit() return { "file_url": attachment_result["file_url"], "file_name": export_meta["file_name"], "stat_type": report["stat_type"], "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: 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 _parse_filters(self, filters: dict, stat_type: str) -> dict: 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"), } 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: 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]: 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: 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") def _build_demo_report(self, stat_type: str) -> dict: period = "2026-05" if stat_type == "month" else "2026-Q2" if stat_type == "quarter" else "2026" return { "stat_type": stat_type, "list": [ { "stat_period": period, "order_count": len(demo_store.orders), "order_amount": round( sum(float(order.get("sale_price_total", 0) or 0) for order in demo_store.orders), 2, ), "category_amounts": [ { "category_id": 0, "category_name": "演示数据", "amount": round( sum(float(order.get("sale_price_total", 0) or 0) for order in demo_store.orders), 2, ), } ], "commission_amount": round( sum(float(order.get("commission_amount", 0) or 0) for order in demo_store.orders), 2, ), } ], } report_service = ReportService()