dingdanquanliucheng/backend/app/services/customer_service.py

"""客户管理服务层。

负责客户的增删改查、批量导入、权限过滤等业务逻辑。
依赖 CustomerRepository 进行数据持久化，依赖 audit_service 记录操作审计日志，
依赖 storage_service 解析导入文件的存储路径。
"""

import csv
from pathlib import Path

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

from backend.app.core.cache import cache_delete_pattern, cache_get, cache_set, make_cache_key
from backend.app.core.error_codes import ErrorCode
from backend.app.core.exceptions import AppException
from backend.app.repositories.customer_repository import CustomerRepository
from backend.app.services.audit_service import audit_service
from backend.app.services.storage_service import storage_service


class CustomerService:
    """客户服务，封装客户信息的 CRUD、导入和权限控制。

    依赖:
        CustomerRepository - 客户数据访问
        audit_service - 操作审计日志
        storage_service - 文件存储路径解析
    """

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

    def list_customers(self, session: Session | None = None, filters: dict | None = None) -> dict:
        """查询客户列表。

        Args:
            session: 数据库会话
            filters: 查询过滤条件

        Returns:
            包含 total、page_no、page_size、list 的分页字典

        被调用路由: customers.py - GET /customers
        """
        if session is None:
            raise AppException(code=ErrorCode.PARAM_ERROR, message="数据库会话未初始化", status_code=500)
        # 读缓存
        cache_key = make_cache_key("customer:list", **(filters or {}))
        cached = cache_get(cache_key)
        if cached is not None:
            return cached
        try:
            customers = self.repository.list_customers_by_filters(session, filters or {})
            result = {
                "total": len(customers),
                "page_no": 1,
                "page_size": 20,
                "list": [
                    {
                        "customer_id": item.id,
                        "customer_name": item.customer_name,
                        "mobile": item.mobile,
                        "address": item.address,
                        "settlement_type": item.settlement_type,
                        "settlement_days": item.settlement_days,
                        "customer_type": item.customer_type,
                        "salesman_id": item.salesman_id,
                        "salesman_name": "",
                        "arrears_amount": 0,
                        "status": 1,
                    }
                    for item in customers
                ],
            }
            cache_set(cache_key, result, ttl=60)
            return result
        except SQLAlchemyError as exc:
            raise AppException(code=ErrorCode.SYSTEM_ERROR, message="数据库操作失败", status_code=500) from exc

    def create_customer(
        self,
        session: Session | None = None,
        payload: dict | None = None,
        current_user: dict | None = None,
    ) -> dict:
        """创建新客户。

        校验客户姓名和手机号非空，并通过姓名+手机号判断是否重复。
        自动记录审计日志。

        Args:
            session: 数据库会话
            payload: 包含 customer_name、mobile 等字段的请求体
            current_user: 当前登录用户信息，用于确定业务员归属

        Returns:
            包含 customer_id、customer_name、mobile 的结果字典

        被调用路由: customers.py - POST /customers
        """
        if session is None or payload is None:
            raise AppException(code=ErrorCode.PARAM_ERROR, message="数据库会话或参数未初始化", status_code=500)
        try:
            if not payload["customer_name"].strip():
                raise AppException(code=ErrorCode.PARAM_ERROR, message="客户姓名不能为空", status_code=400)
            if not payload["mobile"].strip():
                raise AppException(code=ErrorCode.PARAM_ERROR, message="客户手机号不能为空", status_code=400)

            existed = self.repository.find_by_name_and_mobile(
                session,
                payload["customer_name"],
                payload["mobile"],
            )
            if existed is not None:
                raise AppException(code=ErrorCode.DUPLICATE, message="客户已存在", status_code=400)

            customer = self.repository.create_customer(
                session,
                {
                    "customer_name": payload["customer_name"],
                    "mobile": payload["mobile"],
                    "address": payload.get("address"),
                    "settlement_type": payload.get("settlement_type"),
                    "settlement_days": payload.get("settlement_days", 0),
                    "customer_type": payload.get("customer_type"),
                    "salesman_id": self._resolve_salesman_id(payload, current_user),
                    "credit_limit": payload.get("credit_limit", 0),
                    "remark": payload.get("remark"),
                },
            )
            audit_service.write_log(
                session,
                {
                    "operate_type": "customer_create",
                    "biz_type": "customer",
                    "biz_id": customer.id,
                    "before_value": None,
                    "after_value": {
                        "customer_name": customer.customer_name,
                        "mobile": customer.mobile,
                    },
                    "remark": f"新增客户 {customer.customer_name}",
                },
            )
            session.commit()
            cache_delete_pattern("customer:list:*")
            return {
                "customer_id": customer.id,
                "customer_name": customer.customer_name,
                "mobile": customer.mobile,
            }
        except AppException:
            session.rollback()
            raise
        except SQLAlchemyError as exc:
            session.rollback()
            raise AppException(code=ErrorCode.SYSTEM_ERROR, message="数据库操作失败", status_code=500) from exc

    def update_customer(
        self,
        customer_id: int,
        session: Session | None = None,
        payload: dict | None = None,
        current_user: dict | None = None,
    ) -> dict | None:
        """更新客户信息。

        校验客户姓名+手机号唯一性，执行权限检查。
        自动记录审计日志。

        Args:
            customer_id: 客户 ID
            session: 数据库会话
            payload: 包含更新字段的请求体
            current_user: 当前登录用户信息

        Returns:
            更新结果字典，客户不存在时返回 None

        被调用路由: customers.py - PUT /customers/{customer_id}
        """
        if session is None or payload is None:
            return None
        try:
            customer = self.repository.get_customer(session, customer_id)
            if customer is None:
                return None
            self._ensure_customer_access(customer, current_user)
            if not payload["customer_name"].strip():
                raise AppException(code=ErrorCode.PARAM_ERROR, message="客户姓名不能为空", status_code=400)
            if not payload["mobile"].strip():
                raise AppException(code=ErrorCode.PARAM_ERROR, message="客户手机号不能为空", status_code=400)
            existed = self.repository.find_by_name_and_mobile(session, payload["customer_name"], payload["mobile"])
            if existed is not None and existed.id != customer_id:
                raise AppException(code=ErrorCode.DUPLICATE, message="客户姓名+手机号已存在", status_code=400)
            self.repository.update_customer(
                session,
                customer,
                {
                    "customer_name": payload["customer_name"],
                    "mobile": payload["mobile"],
                    "address": payload.get("address"),
                    "settlement_type": payload.get("settlement_type"),
                    "settlement_days": payload.get("settlement_days", 0),
                    "customer_type": payload.get("customer_type"),
                    "salesman_id": payload.get("salesman_id"),
                    "credit_limit": payload.get("credit_limit", 0),
                    "remark": payload.get("remark"),
                },
            )
            audit_service.write_log(
                session,
                {
                    "operate_type": "customer_update",
                    "biz_type": "customer",
                    "biz_id": customer.id,
                    "before_value": None,
                    "after_value": {"customer_name": customer.customer_name, "mobile": customer.mobile},
                    "remark": f"更新客户 {customer.customer_name}",
                },
            )
            session.commit()
            cache_delete_pattern("customer:list:*")
            return {"customer_id": customer.id, "customer_name": customer.customer_name, "mobile": customer.mobile}
        except AppException:
            session.rollback()
            raise
        except SQLAlchemyError as exc:
            session.rollback()
            raise AppException(code=ErrorCode.INTERNAL_ERROR, message="数据库异常", status_code=500) from exc

    def get_customer(
        self,
        customer_id: int,
        session: Session | None = None,
        current_user: dict | None = None,
    ) -> dict:
        """查询单个客户详情。

        Args:
            customer_id: 客户 ID
            session: 数据库会话
            current_user: 当前登录用户信息，用于权限校验

        Returns:
            客户详情字典，包含所有字段

        被调用路由: customers.py - GET /customers/{customer_id}
        """
        if session is None:
            raise AppException(code=ErrorCode.PARAM_ERROR, message="数据库会话未初始化", status_code=500)
        try:
            customer = self.repository.get_customer(session, customer_id)
            if customer is None:
                raise AppException(code=ErrorCode.NOT_FOUND, message="客户不存在", status_code=404)
            self._ensure_customer_access(customer, current_user)
            return {
                "customer_id": customer.id,
                "customer_name": customer.customer_name,
                "mobile": customer.mobile,
                "address": customer.address,
                "settlement_type": customer.settlement_type,
                "settlement_days": customer.settlement_days,
                "customer_type": customer.customer_type,
                "salesman_id": customer.salesman_id,
                "salesman_name": "",
                "arrears_amount": 0,
                "status": 1,
                "credit_limit": float(customer.credit_limit or 0),
                "remark": customer.remark,
            }
        except AppException:
            raise
        except SQLAlchemyError as exc:
            raise AppException(code=ErrorCode.SYSTEM_ERROR, message="数据库操作失败", status_code=500) from exc

    def import_customers_from_file(self, session: Session | None = None, payload: dict | None = None) -> dict:
        """从文件批量导入客户。

        支持 csv、xlsx、xls 格式，支持跳过重复或覆盖重复两种导入模式。
        自动记录审计日志。

        Args:
            session: 数据库会话
            payload: 包含 file_url、import_mode 的请求体

        Returns:
            包含 total_count、success_count、duplicate_count、fail_count、fail_list 的导入结果

        被调用路由: customers.py - POST /customers/import
        """
        if payload is None:
            raise AppException(code=ErrorCode.PARAM_ERROR, message="导入参数不能为空", status_code=400)

        file_url = (payload.get("file_url") or "").strip()
        import_mode = (payload.get("import_mode") or "skip_duplicate").strip()
        if not file_url:
            raise AppException(code=ErrorCode.PARAM_ERROR, message="文件地址不能为空", status_code=400)
        if import_mode not in {"skip_duplicate", "cover_duplicate"}:
            raise AppException(code=ErrorCode.PARAM_ERROR, message="导入模式不正确", status_code=400)

        if session is None:
            raise AppException(code=ErrorCode.PARAM_ERROR, message="数据库会话未初始化", status_code=500)
        try:
            rows = self._load_import_rows(file_url)
            result = self._import_customer_rows(session, rows, import_mode, file_url)
            audit_service.write_log(
                session,
                {
                    "operate_type": "customer_import",
                    "biz_type": "customer",
                    "biz_id": 0,
                    "before_value": None,
                    "after_value": {
                        "total_count": result["total_count"],
                        "success_count": result["success_count"],
                        "duplicate_count": result["duplicate_count"],
                        "fail_count": result["fail_count"],
                        "file_url": file_url,
                        "import_mode": import_mode,
                    },
                    "remark": f"批量导入客户，成功 {result['success_count']} 条",
                },
            )
            session.commit()
            cache_delete_pattern("customer:list:*")
            return result
        except AppException:
            session.rollback()
            raise
        except SQLAlchemyError as exc:
            session.rollback()
            raise AppException(code=ErrorCode.SYSTEM_ERROR, message="数据库操作失败", status_code=500) from exc

    def _load_import_rows(self, file_url: str) -> list[dict]:
        """加载导入文件内容并解析为行数据列表。

        支持 csv（UTF-8-sig 编码）和 xlsx/xls 格式。

        Args:
            file_url: 文件地址（本地路径或 OSS URL）

        Returns:
            解析后的行数据字典列表

        Raises:
            AppException: 文件不存在或格式不支持时抛出
        """
        path = self._resolve_import_path(file_url)
        if not path.exists() or not path.is_file():
            raise AppException(code=ErrorCode.PARAM_ERROR, message="导入文件不存在", status_code=400)

        suffix = path.suffix.lower()
        if suffix == ".csv":
            with path.open("r", encoding="utf-8-sig", newline="") as file:
                return list(csv.DictReader(file))

        if suffix in {".xlsx", ".xls"}:
            try:
                from openpyxl import load_workbook
            except ImportError as exc:
                raise AppException(code=ErrorCode.SYSTEM_ERROR, message="缺少 Excel 解析依赖 openpyxl", status_code=500) from exc

            workbook = load_workbook(path, read_only=True, data_only=True)
            sheet = workbook.active
            rows = list(sheet.iter_rows(values_only=True))
            if not rows:
                return []
            headers = [str(item).strip() if item is not None else "" for item in rows[0]]
            return [
                {headers[index]: value for index, value in enumerate(row) if index < len(headers)}
                for row in rows[1:]
            ]

        raise AppException(code=ErrorCode.PARAM_ERROR, message="仅支持导入 csv、xlsx、xls 文件", status_code=400)

    def _resolve_import_path(self, file_url: str) -> Path:
        """将导入文件 URL 转换为本地文件路径。

        如果是 OSS URL，从 OSS 下载到本地临时文件；
        如果是本地路径，直接返回。

        Args:
            file_url: 文件地址

        Returns:
            本地文件 Path 对象
        """
        import tempfile

        normalized = file_url.replace("\\", "/")
        # 如果是 OSS URL，从 OSS 下载到本地临时文件
        if normalized.startswith(("http://", "https://")):
            object_key = storage_service.extract_object_key(file_url)
            suffix = Path(object_key).suffix
            with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
                storage_service.bucket.get_object_to_file(object_key, tmp.name)
                return Path(tmp.name)
        path = Path(file_url)
        if not path.is_absolute():
            path = Path.cwd() / path
        return path

    def _import_customer_rows(self, session: Session, rows: list[dict], import_mode: str, file_url: str) -> dict:
        """逐行导入客户数据，处理重复判断和错误收集。

        Args:
            session: 数据库会话
            rows: 解析后的行数据列表
            import_mode: 导入模式，skip_duplicate 或 cover_duplicate
            file_url: 导入文件地址（用于备注）

        Returns:
            包含统计信息和失败明细的导入结果字典
        """
        fail_list: list[dict] = []
        success_count = 0
        duplicate_count = 0

        for index, raw_row in enumerate(rows, start=2):
            try:
                payload = self._normalize_import_row(raw_row, file_url)
                existed = self.repository.find_by_name_and_mobile(session, payload["customer_name"], payload["mobile"])
                if existed is not None:
                    duplicate_count += 1
                    if import_mode == "skip_duplicate":
                        continue
                    self.repository.update_customer(session, existed, payload)
                    success_count += 1
                    continue

                self.repository.create_customer(session, payload)
                success_count += 1
            except AppException as exc:
                fail_list.append({"row_no": index, "reason": exc.message})

        return {
            "total_count": len(rows),
            "success_count": success_count,
            "duplicate_count": duplicate_count,
            "fail_count": len(fail_list),
            "fail_list": fail_list,
        }

    def _normalize_import_row(self, raw_row: dict, file_url: str) -> dict:
        """将导入行的原始数据标准化为客户创建所需字段。

        支持中英文混合列名匹配，如"客户姓名"和"customer_name"均可识别。

        Args:
            raw_row: 原始行数据字典
            file_url: 导入文件地址

        Returns:
            标准化后的客户字段字典
        """
        customer_name = self._pick_import_value(raw_row, ["customer_name", "客户姓名", "姓名"])
        mobile = self._pick_import_value(raw_row, ["mobile", "手机号", "手机号码"])
        if not customer_name:
            raise AppException(code=ErrorCode.PARAM_ERROR, message="客户姓名不能为空", status_code=400)
        if not mobile:
            raise AppException(code=ErrorCode.PARAM_ERROR, message="手机号不能为空", status_code=400)

        settlement_days = self._parse_int(self._pick_import_value(raw_row, ["settlement_days", "账期天数"]), 0)
        salesman_id = self._parse_int(self._pick_import_value(raw_row, ["salesman_id", "业务员ID"]), None)
        credit_limit = self._parse_float(self._pick_import_value(raw_row, ["credit_limit", "信用额度"]), 0)

        return {
            "customer_name": customer_name,
            "mobile": mobile,
            "address": self._pick_import_value(raw_row, ["address", "客户地址", "地址"]) or None,
            "settlement_type": self._pick_import_value(raw_row, ["settlement_type", "结算方式"]) or "monthly",
            "settlement_days": settlement_days,
            "customer_type": self._pick_import_value(raw_row, ["customer_type", "客户类型"]) or "channel",
            "salesman_id": salesman_id,
            "credit_limit": credit_limit,
            "remark": self._pick_import_value(raw_row, ["remark", "备注"]) or f"导入文件：{file_url}",
        }

    def _pick_import_value(self, raw_row: dict, keys: list[str]) -> str:
        """从原始行数据中按优先级尝试多个列名取值。

        Args:
            raw_row: 原始行数据字典
            keys: 候选列名列表，按优先级排列

        Returns:
            第一个非空的值字符串，全部为空时返回空字符串
        """
        for key in keys:
            value = raw_row.get(key)
            if value is None:
                continue
            text = str(value).strip()
            if text:
                return text
        return ""

    def _parse_int(self, value: str, default: int | None) -> int | None:
        """安全解析整数，空字符串返回默认值，格式错误抛出异常。

        Args:
            value: 待解析的字符串
            default: 空值时的默认返回值

        Returns:
            解析后的整数或默认值
        """
        if value == "":
            return default
        try:
            return int(float(value))
        except ValueError as exc:
            raise AppException(code=ErrorCode.PARAM_ERROR, message="数字字段格式不正确", status_code=400) from exc

    def _parse_float(self, value: str, default: float) -> float:
        """安全解析浮点数，空字符串返回默认值，格式错误抛出异常。

        Args:
            value: 待解析的字符串
            default: 空值时的默认返回值

        Returns:
            解析后的浮点数或默认值
        """
        if value == "":
            return default
        try:
            return float(value)
        except ValueError as exc:
            raise AppException(code=ErrorCode.PARAM_ERROR, message="金额字段格式不正确", status_code=400) from exc

    def normalize_list_filters(self, filters: dict | None, current_user: dict | None = None) -> dict:
        """标准化查询过滤条件。

        对业务员角色自动注入 salesman_id 过滤条件，确保只能查看自己的客户。

        Args:
            filters: 原始过滤条件
            current_user: 当前登录用户信息

        Returns:
            标准化后的过滤条件字典

        被调用路由: customers.py - GET /customers
        """
        normalized = dict(filters or {})
        if current_user and current_user.get("role_code") == "salesman":
            normalized["salesman_id"] = current_user.get("user_id")
        return normalized

    def _resolve_salesman_id(self, payload: dict, current_user: dict | None) -> int | None:
        """根据当前用户角色确定业务员 ID。

        业务员角色自动使用自身 user_id，其他角色从 payload 中取值。

        Args:
            payload: 请求参数
            current_user: 当前登录用户信息

        Returns:
            业务员 ID 或 None
        """
        if current_user and current_user.get("role_code") == "salesman":
            return current_user.get("user_id")
        return payload.get("salesman_id")

    def _ensure_customer_access(self, customer: object, current_user: dict | None) -> None:
        """校验当前用户是否有权访问指定客户。

        管理员和经理可访问所有客户，业务员只能访问自己负责的客户。

        Args:
            customer: 客户 ORM 对象
            current_user: 当前登录用户信息

        Raises:
            AppException: 无权限时抛出 FORBIDDEN
        """
        if not current_user:
            return
        role_code = current_user.get("role_code")
        if role_code in {"admin", "manager"}:
            return
        if role_code == "salesman":
            if customer.salesman_id != current_user.get("user_id"):
                raise AppException(code=ErrorCode.FORBIDDEN, message="无权限访问他人客户", status_code=403)
            return
        raise AppException(code=ErrorCode.FORBIDDEN, message="无权限访问", status_code=403)


customer_service = CustomerService()