harness_zbt/backend/app/domains/agent/tools.py

"""智保通第一阶段 Agent 可调用的确定性业务工具。"""

from typing import Any, Literal, Protocol

from pydantic import BaseModel, ConfigDict, Field

from app.core.errors import AppError
from app.domains.catalog.service import ProductCatalogService
from app.domains.enrollment.service import EnrollmentService
from app.harness.policy import ToolPolicy
from app.harness.schemas import (
    AgentAction,
    BusinessListBlock,
    HarnessToolResult,
    MetricItem,
    MetricsBlock,
    QuoteBlock,
    QuoteView,
)
from app.harness.tooling import ToolDefinition, ToolExecutionContext, ToolRegistry


class AttributionMetricsProvider(Protocol):
    def performance(self, admin_user_id: str | None = None) -> dict[str, Any]: ...

    def order_ids(self, admin_user_id: str | None = None) -> set[str]: ...


class EmptyAttributionMetrics:
    def performance(self, admin_user_id: str | None = None) -> dict[str, Any]:
        del admin_user_id
        return {
            "salesperson_count": 0,
            "visit_count": 0,
            "lead_count": 0,
            "order_count": 0,
            "premium_cents": 0,
            "items": [],
        }

    def order_ids(self, admin_user_id: str | None = None) -> set[str]:
        del admin_user_id
        return set()


class ToolArguments(BaseModel):
    model_config = ConfigDict(extra="forbid")


class ListProductsArgs(ToolArguments):
    category: Literal["MEDICAL", "ACCIDENT"] | None = None


class CalculateQuoteArgs(ToolArguments):
    product_code: str = Field(min_length=2, max_length=32)
    plan_code: str = Field(min_length=2, max_length=32)
    age: int = Field(ge=0, le=120)
    region_code: str = Field(default="510100", pattern=r"^\d{6}$")
    occupation_code: str = Field(default="GENERAL", min_length=2, max_length=32)
    relationship: Literal["SELF", "PARENT", "SPOUSE", "CHILD"]


class PrepareEnrollmentArgs(ToolArguments):
    product_code: str = Field(min_length=2, max_length=32)
    plan_code: str = Field(min_length=2, max_length=32)
    age: int = Field(ge=0, le=120)
    relationship: Literal["SELF", "PARENT", "SPOUSE", "CHILD"]


class EmptyArgs(ToolArguments):
    pass


class RecentItemsArgs(ToolArguments):
    limit: int = Field(default=5, ge=1, le=20)


def build_agent_tool_registry(
    catalog: ProductCatalogService,
    enrollment: EnrollmentService,
    attribution: AttributionMetricsProvider | None = None,
) -> ToolRegistry:
    registry = ToolRegistry()
    metrics = attribution or EmptyAttributionMetrics()

    def list_products(
        context: ToolExecutionContext,
        arguments: BaseModel,
    ) -> HarnessToolResult:
        del context
        args = _arguments(arguments, ListProductsArgs)
        products = catalog.list_available(category=args.category)
        return HarnessToolResult(
            summary=f"当前共有{len(products)}款符合条件的在售产品",
            data={"items": products, "total": len(products)},
        )

    def calculate_quote(
        context: ToolExecutionContext,
        arguments: BaseModel,
    ) -> HarnessToolResult:
        args = _arguments(arguments, CalculateQuoteArgs)
        user = context.h5_user
        if user is None:
            raise AppError("AGENT_CUSTOMER_REQUIRED", "保费测算需要H5用户身份", 401)
        product, plan = _resolve_product_plan(
            catalog,
            args.product_code,
            args.plan_code,
        )
        try:
            quote = enrollment.create_quote(
                user,
                product_id=str(product["product_id"]),
                plan_id=str(plan["id"]),
                age=args.age,
                region_code=args.region_code,
                occupation_code=args.occupation_code,
                relationship=args.relationship,
            )
        except AppError as error:
            if error.code != "ELIGIBILITY_REJECTED":
                raise
            block = QuoteBlock(
                quote=QuoteView(
                    eligible=False,
                    product_name=str(product["name"]),
                    plan_name=str(plan["name"]),
                    reason=error.message,
                )
            )
            return HarnessToolResult(
                summary=f"资格校验未通过：{error.message}",
                data={
                    "eligible": False,
                    "reason_code": (error.details or {}).get("reason_code"),
                },
                blocks=[block],
            )
        block = QuoteBlock(
            quote=QuoteView(
                eligible=True,
                quote_id=str(quote["quote_id"]),
                product_id=str(quote["product_id"]),
                product_name=str(product["name"]),
                plan_id=str(quote["plan_id"]),
                plan_name=str(plan["name"]),
                premium_cents=int(quote["premium_cents"]),
                currency=str(quote["currency"]),
                expires_at=str(quote["expires_at"]),
            )
        )
        return HarnessToolResult(
            summary=(
                f"{product['name']}{plan['name']}测算成功，"
                f"年保费{int(quote['premium_cents']) / 100:.2f}元"
            ),
            data=quote,
            blocks=[block],
            actions=[
                AgentAction(
                    type="open_enrollment",
                    label="填写投保信息",
                    payload={
                        "product_code": args.product_code,
                        "plan_code": args.plan_code,
                        "quote_id": quote["quote_id"],
                        "age": args.age,
                        "relationship": args.relationship,
                    },
                )
            ],
        )

    def prepare_enrollment(
        context: ToolExecutionContext,
        arguments: BaseModel,
    ) -> HarnessToolResult:
        if context.h5_user is None:
            raise AppError("AGENT_CUSTOMER_REQUIRED", "投保引导需要H5用户身份", 401)
        args = _arguments(arguments, PrepareEnrollmentArgs)
        product, plan = _resolve_product_plan(
            catalog,
            args.product_code,
            args.plan_code,
        )
        action = AgentAction(
            type="open_enrollment",
            label="立即投保",
            payload={
                "product_id": product["product_id"],
                "product_code": args.product_code,
                "plan_id": plan["id"],
                "plan_code": args.plan_code,
                "age": args.age,
                "relationship": args.relationship,
            },
        )
        return HarnessToolResult(
            summary="已准备安全投保表单入口",
            data={"product": product, "plan": plan},
            actions=[action],
        )

    def list_my_orders(
        context: ToolExecutionContext,
        arguments: BaseModel,
    ) -> HarnessToolResult:
        del arguments
        user = context.h5_user
        if user is None:
            raise AppError("AGENT_CUSTOMER_REQUIRED", "订单查询需要H5用户身份", 401)
        result = enrollment.list_orders(user.id)
        items = [_safe_order_item(item) for item in result["items"]]
        return HarnessToolResult(
            summary=f"查到{len(items)}笔本人投保订单",
            data={"items": items, "total": len(items)},
            blocks=[
                BusinessListBlock(
                    title="我的投保订单",
                    entity="order",
                    items=items[:5],
                    total=len(items),
                )
            ],
            actions=[AgentAction(type="open_orders", label="查看我的订单")],
        )

    def list_my_policies(
        context: ToolExecutionContext,
        arguments: BaseModel,
    ) -> HarnessToolResult:
        del arguments
        user = context.h5_user
        if user is None:
            raise AppError("AGENT_CUSTOMER_REQUIRED", "保单查询需要H5用户身份", 401)
        result = enrollment.list_policies(user)
        items = [_safe_policy_item(item) for item in result["items"]]
        return HarnessToolResult(
            summary=f"查到{len(items)}张本人保单",
            data={"items": items, "total": len(items)},
            blocks=[
                BusinessListBlock(
                    title="我的保障",
                    entity="policy",
                    items=items[:5],
                    total=len(items),
                )
            ],
            actions=[AgentAction(type="open_policies", label="查看我的保障")],
        )

    def overview(
        context: ToolExecutionContext,
        arguments: BaseModel,
    ) -> HarnessToolResult:
        del arguments
        orders = enrollment.list_orders()
        policies = enrollment.list_all_policies()
        if context.admin_user and context.admin_user.data_scope == "SELF":
            allowed = metrics.order_ids(context.admin_user.id)
            orders = _filter_business_result(orders, allowed, "order_id")
            policies = _filter_business_result(policies, allowed, "order_id")
        active = [item for item in policies["items"] if item["status"] == "ACTIVE"]
        premium_cents = sum(int(item["premium_cents"]) for item in active)
        items = [
            MetricItem(label="订单总量", value=int(orders["total"]), unit="笔"),
            MetricItem(label="有效保单", value=len(active), unit="张"),
            MetricItem(label="累计保费", value=premium_cents / 100, unit="元"),
        ]
        return HarnessToolResult(
            summary=(
                f"当前订单{orders['total']}笔、有效保单{len(active)}张、"
                f"累计保费{premium_cents / 100:.2f}元"
            ),
            data={
                "order_count": orders["total"],
                "active_policy_count": len(active),
                "premium_cents": premium_cents,
            },
            blocks=[MetricsBlock(title="经营总览", items=items)],
        )

    def recent_orders(
        context: ToolExecutionContext,
        arguments: BaseModel,
    ) -> HarnessToolResult:
        args = _arguments(arguments, RecentItemsArgs)
        result = enrollment.list_orders()
        if context.admin_user and context.admin_user.data_scope == "SELF":
            result = _filter_business_result(
                result,
                metrics.order_ids(context.admin_user.id),
                "order_id",
            )
        items = [_safe_order_item(item) for item in result["items"][: args.limit]]
        return HarnessToolResult(
            summary=f"返回最近{len(items)}笔订单",
            data={"items": items, "total": result["total"]},
            blocks=[
                BusinessListBlock(
                    title="近期订单",
                    entity="order",
                    items=items,
                    total=int(result["total"]),
                )
            ],
            actions=[AgentAction(type="open_admin_orders", label="进入订单中心")],
        )

    def recent_policies(
        context: ToolExecutionContext,
        arguments: BaseModel,
    ) -> HarnessToolResult:
        args = _arguments(arguments, RecentItemsArgs)
        result = enrollment.list_all_policies()
        if context.admin_user and context.admin_user.data_scope == "SELF":
            result = _filter_business_result(
                result,
                metrics.order_ids(context.admin_user.id),
                "order_id",
            )
        items = [_safe_policy_item(item) for item in result["items"][: args.limit]]
        return HarnessToolResult(
            summary=f"返回最近{len(items)}张保单",
            data={"items": items, "total": result["total"]},
            blocks=[
                BusinessListBlock(
                    title="近期保单",
                    entity="policy",
                    items=items,
                    total=int(result["total"]),
                )
            ],
            actions=[AgentAction(type="open_admin_policies", label="进入保单中心")],
        )

    def attribution_performance(
        context: ToolExecutionContext,
        arguments: BaseModel,
    ) -> HarnessToolResult:
        del arguments
        admin = context.admin_user
        if admin is None:
            raise AppError("AGENT_ADMIN_REQUIRED", "推广分析需要后台身份", 401)
        owner_id = admin.id if admin.data_scope == "SELF" else None
        result = metrics.performance(owner_id)
        block = MetricsBlock(
            title="推广业绩",
            items=[
                MetricItem(
                    label="推广员",
                    value=int(result["salesperson_count"]),
                    unit="人",
                ),
                MetricItem(label="访问", value=int(result["visit_count"]), unit="次"),
                MetricItem(label="线索", value=int(result["lead_count"]), unit="人"),
                MetricItem(label="归因订单", value=int(result["order_count"]), unit="笔"),
                MetricItem(
                    label="归因保费",
                    value=int(result["premium_cents"]) / 100,
                    unit="元",
                ),
            ],
        )
        return HarnessToolResult(
            summary=(
                f"推广访问{result['visit_count']}次、线索{result['lead_count']}人、"
                f"归因订单{result['order_count']}笔"
            ),
            data=result,
            blocks=[block],
            actions=[AgentAction(type="open_attribution", label="查看推广明细")],
        )

    definitions = (
        ToolDefinition(
            name="list_available_products",
            description="查询当前真实在售保险产品和计划；推荐产品前必须调用。",
            arguments=ListProductsArgs,
            policy=ToolPolicy(personas=("customer", "operation"), effect="read"),
            handler=list_products,
        ),
        ToolDefinition(
            name="calculate_insurance_quote",
            description="按真实产品、年龄、地区、职业和关系执行确定性资格校验与保费测算。",
            arguments=CalculateQuoteArgs,
            policy=ToolPolicy(personas=("customer",), effect="draft"),
            handler=calculate_quote,
        ),
        ToolDefinition(
            name="prepare_enrollment",
            description="为已明确的产品和计划生成安全投保表单入口，不采集身份证号。",
            arguments=PrepareEnrollmentArgs,
            policy=ToolPolicy(personas=("customer",), effect="draft"),
            handler=prepare_enrollment,
        ),
        ToolDefinition(
            name="list_my_orders",
            description="查询当前登录客户自己的投保订单。",
            arguments=EmptyArgs,
            policy=ToolPolicy(personas=("customer",), effect="read"),
            handler=list_my_orders,
        ),
        ToolDefinition(
            name="list_my_policies",
            description="查询当前登录客户自己的电子保单。",
            arguments=EmptyArgs,
            policy=ToolPolicy(personas=("customer",), effect="read"),
            handler=list_my_policies,
        ),
        ToolDefinition(
            name="get_operation_overview",
            description="查询订单、有效保单和累计保费经营指标。",
            arguments=EmptyArgs,
            policy=ToolPolicy(
                personas=("operation",),
                effect="read",
                required_permissions=("dashboard:read",),
            ),
            handler=overview,
        ),
        ToolDefinition(
            name="list_recent_orders",
            description="按当前管理员数据范围查询近期订单，只返回脱敏字段。",
            arguments=RecentItemsArgs,
            policy=ToolPolicy(
                personas=("operation",),
                effect="read",
                required_permissions=("order:read",),
            ),
            handler=recent_orders,
        ),
        ToolDefinition(
            name="list_recent_policies",
            description="按当前管理员数据范围查询近期保单，只返回脱敏字段。",
            arguments=RecentItemsArgs,
            policy=ToolPolicy(
                personas=("operation",),
                effect="read",
                required_permissions=("policy:read",),
            ),
            handler=recent_policies,
        ),
        ToolDefinition(
            name="get_attribution_performance",
            description="查询当前管理员数据范围内的推广访问、线索、订单和保费。",
            arguments=EmptyArgs,
            policy=ToolPolicy(
                personas=("operation",),
                effect="read",
                required_permissions=("attribution:read",),
            ),
            handler=attribution_performance,
        ),
    )
    for definition in definitions:
        registry.register(definition)
    return registry


def _arguments(value: BaseModel, expected: type[ToolArguments]) -> Any:
    if not isinstance(value, expected):
        raise AppError("AGENT_TOOL_ARGUMENTS_INVALID", "工具参数类型错误", 500)
    return value


def _resolve_product_plan(
    catalog: ProductCatalogService,
    product_code: str,
    plan_code: str,
) -> tuple[dict[str, Any], dict[str, Any]]:
    product = next(
        (item for item in catalog.list_available() if item["product_code"] == product_code),
        None,
    )
    if product is None:
        raise AppError("PRODUCT_NOT_AVAILABLE", "产品当前不可投保", 404)
    plan = next(
        (item for item in product["plans"] if item["code"] == plan_code),
        None,
    )
    if plan is None:
        raise AppError("PLAN_NOT_AVAILABLE", "保障计划当前不可投保", 404)
    return product, plan


def _safe_order_item(item: dict[str, Any]) -> dict[str, Any]:
    return {
        "order_id": item["order_id"],
        "order_no": item["order_no"],
        "product_name": item["product_name"],
        "plan_name": item["plan_name"],
        "relationship": item["relationship"],
        "status": item["status"],
        "amount_cents": item["amount_cents"],
        "currency": item["currency"],
        "policy_no": item["policy_no"],
        "created_at": str(item["created_at"]),
    }


def _safe_policy_item(item: dict[str, Any]) -> dict[str, Any]:
    return {
        "policy_id": item["policy_id"],
        "policy_no": item["policy_no"],
        "order_no": item["order_no"],
        "product_name": item["product_name"],
        "plan_name": item["plan_name"],
        "relationship": item["relationship"],
        "status": item["status"],
        "premium_cents": item["premium_cents"],
        "currency": item["currency"],
        "coverage_start": str(item["coverage_start"]),
        "coverage_end": str(item["coverage_end"]),
        "issued_at": str(item["issued_at"]),
    }


def _filter_business_result(
    result: dict[str, Any],
    allowed_ids: set[str],
    id_field: str,
) -> dict[str, Any]:
    items = [item for item in result["items"] if str(item.get(id_field, "")) in allowed_ids]
    return {"items": items, "total": len(items)}