harness_zbt/backend/app/domains/catalog/service.py

"""产品目录业务逻辑。

这一层不关心数据来自 MySQL 还是测试内存，只通过 CatalogRepository 接口取数据。
"""

import re
from collections.abc import Callable
from dataclasses import replace
from datetime import datetime, timedelta
from typing import Any

from app.core.errors import AppError
from app.core.identifiers import new_ulid
from app.domains.catalog.models import Plan, Product, ProductChangeLog, ProductVersion
from app.domains.catalog.repository import CatalogRepository


class ProductCatalogService:
    def __init__(
        self,
        repository: CatalogRepository,
        clock: Callable[[], datetime],
    ) -> None:
        self._repository = repository
        self._clock = clock

    def list_available(self, *, category: str | None = None) -> list[dict[str, Any]]:
        """返回当前时间有效的在售产品，可选按产品类别过滤。"""
        now = self._clock()
        items: list[dict[str, Any]] = []
        for product in self._repository.list_products():
            # continue 会跳过本轮循环，继续检查下一个产品。
            if product.status != "ACTIVE" or (category and product.category != category):
                continue

            # 列表推导式：从产品的所有版本中筛出当前生效的版本。
            versions = [
                version
                for version in self._repository.list_versions(product.id)
                if version.is_available_at(now)
            ]
            if not versions:
                continue
            # 若多个版本同时有效，选择生效时间最晚的版本作为当前版本。
            current = max(versions, key=lambda version: version.effective_from)
            items.append(
                {
                    "product_id": product.id,
                    "product_code": product.product_code,
                    "name": product.name,
                    "category": product.category,
                    "summary": product.summary,
                    "product_version_id": current.id,
                    "version_no": current.version_no,
                    "plans": [
                        {
                            "id": plan.id,
                            "code": plan.code,
                            "name": plan.name,
                            "summary": plan.summary,
                            "premium_cents": plan.premium_cents,
                            "coverage_amount_cents": plan.coverage_amount_cents,
                            "min_age": plan.min_age,
                            "max_age": plan.max_age,
                        }
                        for plan in current.plans
                        if plan.status == "ACTIVE"
                    ],
                }
            )
        # lambda 是一个简短的匿名函数，这里告诉 sorted 按产品编码排序。
        return sorted(items, key=lambda item: str(item["product_code"]))

    def list_managed(self) -> dict[str, Any]:
        items = []
        for product in self._repository.list_products():
            versions = self._repository.list_versions(product.id)
            items.append(
                {
                    "product_id": product.id,
                    "product_code": product.product_code,
                    "name": product.name,
                    "category": product.category,
                    "summary": product.summary,
                    "status": product.status,
                    "versions": [
                        {
                            "id": version.id,
                            "version_no": version.version_no,
                            "status": version.status,
                            "effective_from": version.effective_from,
                            "effective_to": version.effective_to,
                            "rule_version": version.rule_version,
                            "rate_version": version.rate_version,
                            "terms_summary": version.terms_summary,
                            "plan_count": len(version.plans),
                            "plans": [
                                {
                                    "id": plan.id,
                                    "code": plan.code,
                                    "name": plan.name,
                                    "summary": plan.summary,
                                    "status": plan.status,
                                    "premium_cents": plan.premium_cents,
                                    "coverage_amount_cents": plan.coverage_amount_cents,
                                    "min_age": plan.min_age,
                                    "max_age": plan.max_age,
                                }
                                for plan in version.plans
                            ],
                        }
                        for version in sorted(
                            versions,
                            key=lambda item: item.effective_from,
                            reverse=True,
                        )
                    ],
                    "recent_logs": [
                        self._serialize_log(log)
                        for log in self._repository.list_change_logs(product.id)[:8]
                    ],
                }
            )
        sorted_items = sorted(items, key=lambda item: str(item["product_code"]))
        return {"items": sorted_items, "total": len(sorted_items)}

    def update_product_status(
        self,
        product_id: str,
        status: str,
        *,
        actor_id: str = "SYSTEM",
        actor_name: str = "系统",
    ) -> dict[str, Any]:
        if status not in {"ACTIVE", "INACTIVE"}:
            raise AppError("PRODUCT_STATUS_INVALID", "产品状态无效", 422)
        product = next(
            (item for item in self._repository.list_products() if item.id == product_id),
            None,
        )
        if product is None:
            raise AppError("PRODUCT_NOT_FOUND", "未找到产品", 404)
        updated = replace(product, status=status)
        self._repository.save_product(updated)
        self._write_log(
            product_id,
            None,
            "PRODUCT_STATUS_CHANGED",
            actor_id,
            actor_name,
            {"from": product.status, "to": status},
        )
        return {
            "product_id": updated.id,
            "product_code": updated.product_code,
            "name": updated.name,
            "status": updated.status,
        }

    def update_product(
        self,
        product_id: str,
        *,
        name: str,
        category: str,
        summary: str,
        actor_id: str,
        actor_name: str,
    ) -> dict[str, Any]:
        product = self._get_product(product_id)
        normalized_name = name.strip()
        normalized_summary = summary.strip()
        if not normalized_name or not normalized_summary:
            raise AppError("PRODUCT_CONTENT_REQUIRED", "产品名称和产品简介不能为空", 422)
        if category not in {"MEDICAL", "ACCIDENT"}:
            raise AppError("PRODUCT_CATEGORY_INVALID", "产品分类无效", 422)
        updated = replace(
            product,
            name=normalized_name,
            category=category,
            summary=normalized_summary,
        )
        self._repository.save_product(updated)
        self._write_log(
            product_id,
            None,
            "PRODUCT_UPDATED",
            actor_id,
            actor_name,
            {"name": updated.name, "category": updated.category},
        )
        return self._serialize_product(updated)

    def create_product(
        self,
        *,
        product_code: str,
        name: str,
        category: str,
        summary: str,
        version_no: str,
        effective_from: datetime,
        rule_version: str,
        rate_version: str,
        terms_summary: str,
        plans: list[dict[str, Any]],
        actor_id: str,
        actor_name: str,
    ) -> dict[str, Any]:
        normalized_code = product_code.strip().upper()
        normalized_name = name.strip()
        normalized_summary = summary.strip()
        normalized_version = version_no.strip()
        if not re.fullmatch(r"[A-Z][A-Z0-9-]{2,31}", normalized_code):
            raise AppError(
                "PRODUCT_CODE_INVALID",
                "产品编码须以字母开头，仅包含大写字母、数字和连字符",
                422,
            )
        if any(
            product.product_code.upper() == normalized_code
            for product in self._repository.list_products()
        ):
            raise AppError("PRODUCT_CODE_DUPLICATED", "产品编码已存在", 409)
        if not normalized_name or not normalized_summary:
            raise AppError("PRODUCT_CONTENT_REQUIRED", "产品名称和产品简介不能为空", 422)
        if category not in {"MEDICAL", "ACCIDENT"}:
            raise AppError("PRODUCT_CATEGORY_INVALID", "产品分类无效", 422)
        if not normalized_version:
            raise AppError("VERSION_NO_REQUIRED", "首个版本号不能为空", 422)
        product = Product(
            id=new_ulid(),
            product_code=normalized_code,
            name=normalized_name,
            category=category,
            summary=normalized_summary,
            status="INACTIVE",
        )
        initial_version = ProductVersion(
            id=new_ulid(),
            product_id=product.id,
            version_no=normalized_version,
            status="DRAFT",
            effective_from=self._require_aware(effective_from),
            effective_to=None,
            plans=self._build_plans(plans, ()),
            rule_version=rule_version.strip() or "eligibility-v1",
            rate_version=rate_version.strip() or "rate-v1",
            terms_summary=terms_summary.strip(),
        )
        self._repository.save_product(product)
        self._repository.save_version(initial_version)
        self._write_log(
            product.id,
            initial_version.id,
            "PRODUCT_CREATED",
            actor_id,
            actor_name,
            {
                "product_code": product.product_code,
                "version_no": initial_version.version_no,
            },
        )
        return {
            "product": self._serialize_product(product),
            "initial_version": self._serialize_version(initial_version),
        }

    def create_draft_version(
        self,
        product_id: str,
        *,
        version_no: str,
        effective_from: datetime,
        source_version_id: str | None,
        actor_id: str,
        actor_name: str,
    ) -> dict[str, Any]:
        self._get_product(product_id)
        normalized_version = version_no.strip()
        versions = self._repository.list_versions(product_id)
        if not normalized_version:
            raise AppError("VERSION_NO_REQUIRED", "版本号不能为空", 422)
        if any(version.version_no == normalized_version for version in versions):
            raise AppError("VERSION_NO_DUPLICATED", "该产品版本号已存在", 409)
        source = None
        if source_version_id:
            source = next(
                (version for version in versions if version.id == source_version_id),
                None,
            )
            if source is None:
                raise AppError("SOURCE_VERSION_NOT_FOUND", "未找到要复制的产品版本", 404)
        elif versions:
            source = max(versions, key=lambda version: version.effective_from)
        plans = tuple(
            replace(plan, id=new_ulid()) for plan in source.plans
        ) if source else ()
        draft = ProductVersion(
            id=new_ulid(),
            product_id=product_id,
            version_no=normalized_version,
            status="DRAFT",
            effective_from=self._require_aware(effective_from),
            effective_to=None,
            plans=plans,
            rule_version=source.rule_version if source else "eligibility-v1",
            rate_version=source.rate_version if source else "rate-v1",
            terms_summary=source.terms_summary if source else "",
        )
        self._repository.save_version(draft)
        self._write_log(
            product_id,
            draft.id,
            "VERSION_DRAFT_CREATED",
            actor_id,
            actor_name,
            {"version_no": draft.version_no, "copied_from": source_version_id},
        )
        return self._serialize_version(draft)

    def update_draft_version(
        self,
        product_id: str,
        version_id: str,
        *,
        effective_from: datetime,
        effective_to: datetime | None,
        rule_version: str,
        rate_version: str,
        terms_summary: str,
        plans: list[dict[str, Any]],
        actor_id: str,
        actor_name: str,
    ) -> dict[str, Any]:
        version = self._get_version(product_id, version_id)
        if version.status != "DRAFT":
            raise AppError("VERSION_IMMUTABLE", "只有草稿版本可以编辑", 409)
        start = self._require_aware(effective_from)
        end = self._require_aware(effective_to) if effective_to else None
        if end is not None and end <= start:
            raise AppError("VERSION_PERIOD_INVALID", "销售截止时间必须晚于生效时间", 422)
        updated = replace(
            version,
            effective_from=start,
            effective_to=end,
            rule_version=rule_version.strip() or "eligibility-v1",
            rate_version=rate_version.strip() or "rate-v1",
            terms_summary=terms_summary.strip(),
            plans=self._build_plans(plans, version.plans),
        )
        self._repository.save_version(updated)
        self._write_log(
            product_id,
            version_id,
            "VERSION_DRAFT_UPDATED",
            actor_id,
            actor_name,
            {"version_no": updated.version_no, "plan_count": len(updated.plans)},
        )
        return self._serialize_version(updated)

    def publish_version(
        self,
        product_id: str,
        version_id: str,
        *,
        actor_id: str,
        actor_name: str,
    ) -> dict[str, Any]:
        version = self._get_version(product_id, version_id)
        if version.status != "DRAFT":
            raise AppError("VERSION_NOT_DRAFT", "只有草稿版本可以发布", 409)
        if not version.terms_summary:
            raise AppError("VERSION_TERMS_REQUIRED", "发布前请填写投保须知与条款摘要", 422)
        active_plans = [plan for plan in version.plans if plan.status == "ACTIVE"]
        if not active_plans:
            raise AppError("VERSION_PLAN_REQUIRED", "发布前至少需要一个有效保障计划", 422)
        if any(plan.premium_cents <= 0 or plan.coverage_amount_cents <= 0 for plan in active_plans):
            raise AppError("VERSION_PLAN_INVALID", "有效计划的保费和保额必须大于零", 422)
        for existing in self._repository.list_versions(product_id):
            if (
                existing.status == "PUBLISHED"
                and existing.effective_from < version.effective_from
                and (
                    existing.effective_to is None
                    or existing.effective_to >= version.effective_from
                )
            ):
                self._repository.save_version(
                    replace(
                        existing,
                        effective_to=version.effective_from - timedelta(microseconds=1),
                    )
                )
        published = replace(version, status="PUBLISHED")
        self._repository.save_version(published)
        self._write_log(
            product_id,
            version_id,
            "VERSION_PUBLISHED",
            actor_id,
            actor_name,
            {"version_no": published.version_no},
        )
        return self._serialize_version(published)

    def retire_version(
        self,
        product_id: str,
        version_id: str,
        *,
        actor_id: str,
        actor_name: str,
    ) -> dict[str, Any]:
        version = self._get_version(product_id, version_id)
        if version.status != "PUBLISHED":
            raise AppError("VERSION_NOT_PUBLISHED", "只有已发布版本可以终止", 409)
        retired = replace(version, status="RETIRED", effective_to=self._clock())
        self._repository.save_version(retired)
        self._write_log(
            product_id,
            version_id,
            "VERSION_RETIRED",
            actor_id,
            actor_name,
            {"version_no": retired.version_no},
        )
        return self._serialize_version(retired)

    def delete_draft_version(
        self,
        product_id: str,
        version_id: str,
        *,
        actor_id: str,
        actor_name: str,
    ) -> dict[str, Any]:
        version = self._get_version(product_id, version_id)
        if version.status != "DRAFT":
            raise AppError("VERSION_DELETE_DENIED", "只有草稿版本可以删除", 409)
        self._repository.delete_version(version_id)
        self._write_log(
            product_id,
            version_id,
            "VERSION_DRAFT_DELETED",
            actor_id,
            actor_name,
            {"version_no": version.version_no},
        )
        return {"version_id": version_id, "deleted": True}

    def list_change_logs(self, product_id: str) -> dict[str, Any]:
        self._get_product(product_id)
        items = [
            self._serialize_log(log)
            for log in self._repository.list_change_logs(product_id)
        ]
        return {"items": items, "total": len(items)}

    def _get_product(self, product_id: str) -> Product:
        product = next(
            (item for item in self._repository.list_products() if item.id == product_id),
            None,
        )
        if product is None:
            raise AppError("PRODUCT_NOT_FOUND", "未找到产品", 404)
        return product

    def _get_version(self, product_id: str, version_id: str) -> ProductVersion:
        self._get_product(product_id)
        version = next(
            (
                item
                for item in self._repository.list_versions(product_id)
                if item.id == version_id
            ),
            None,
        )
        if version is None:
            raise AppError("VERSION_NOT_FOUND", "未找到产品版本", 404)
        return version

    @staticmethod
    def _require_aware(value: datetime) -> datetime:
        if value.tzinfo is None or value.utcoffset() is None:
            raise AppError("DATETIME_TIMEZONE_REQUIRED", "时间必须包含时区", 422)
        return value

    @staticmethod
    def _build_plans(
        items: list[dict[str, Any]],
        existing_plans: tuple[Plan, ...],
    ) -> tuple[Plan, ...]:
        if not items:
            raise AppError("VERSION_PLAN_REQUIRED", "至少需要一个保障计划", 422)
        existing_by_id = {plan.id: plan for plan in existing_plans}
        codes: set[str] = set()
        plans: list[Plan] = []
        for item in items:
            code = str(item.get("code", "")).strip().upper()
            name = str(item.get("name", "")).strip()
            if not code or not name:
                raise AppError("PLAN_CONTENT_REQUIRED", "计划编码和名称不能为空", 422)
            if code in codes:
                raise AppError("PLAN_CODE_DUPLICATED", "同一版本的计划编码不能重复", 422)
            codes.add(code)
            min_age = int(item.get("min_age", 0))
            max_age = int(item.get("max_age", 100))
            premium_cents = int(item.get("premium_cents", 0))
            coverage_amount_cents = int(item.get("coverage_amount_cents", 0))
            if min_age < 0 or max_age > 120 or min_age > max_age:
                raise AppError("PLAN_AGE_INVALID", "计划年龄范围无效", 422)
            if premium_cents < 0 or coverage_amount_cents < 0:
                raise AppError("PLAN_AMOUNT_INVALID", "计划保费和保额不能小于零", 422)
            requested_id = str(item.get("id", ""))
            plan_id = requested_id if requested_id in existing_by_id else new_ulid()
            plans.append(
                Plan(
                    id=plan_id,
                    code=code,
                    name=name,
                    summary=str(item.get("summary", "")).strip(),
                    status=(
                        str(item.get("status", "ACTIVE"))
                        if str(item.get("status", "ACTIVE")) in {"ACTIVE", "INACTIVE"}
                        else "ACTIVE"
                    ),
                    premium_cents=premium_cents,
                    coverage_amount_cents=coverage_amount_cents,
                    min_age=min_age,
                    max_age=max_age,
                )
            )
        return tuple(plans)

    def _write_log(
        self,
        product_id: str,
        version_id: str | None,
        action: str,
        actor_id: str,
        actor_name: str,
        detail: dict[str, Any],
    ) -> None:
        self._repository.save_change_log(
            ProductChangeLog(
                id=new_ulid(),
                product_id=product_id,
                version_id=version_id,
                action=action,
                actor_id=actor_id,
                actor_name=actor_name,
                detail=detail,
                created_at=self._clock(),
            )
        )

    @staticmethod
    def _serialize_product(product: Any) -> dict[str, Any]:
        return {
            "product_id": product.id,
            "product_code": product.product_code,
            "name": product.name,
            "category": product.category,
            "summary": product.summary,
            "status": product.status,
        }

    @staticmethod
    def _serialize_version(version: ProductVersion) -> dict[str, Any]:
        return {
            "id": version.id,
            "version_no": version.version_no,
            "status": version.status,
            "effective_from": version.effective_from,
            "effective_to": version.effective_to,
            "rule_version": version.rule_version,
            "rate_version": version.rate_version,
            "terms_summary": version.terms_summary,
            "plan_count": len(version.plans),
            "plans": [
                {
                    "id": plan.id,
                    "code": plan.code,
                    "name": plan.name,
                    "summary": plan.summary,
                    "status": plan.status,
                    "premium_cents": plan.premium_cents,
                    "coverage_amount_cents": plan.coverage_amount_cents,
                    "min_age": plan.min_age,
                    "max_age": plan.max_age,
                }
                for plan in version.plans
            ],
        }

    @staticmethod
    def _serialize_log(log: ProductChangeLog) -> dict[str, Any]:
        return {
            "id": log.id,
            "version_id": log.version_id,
            "action": log.action,
            "actor_id": log.actor_id,
            "actor_name": log.actor_name,
            "detail": log.detail,
            "created_at": log.created_at,
        }