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

from collections.abc import Callable
from dataclasses import replace
from datetime import datetime
from typing import Any

from app.core.errors import AppError
from app.core.identifiers import new_ulid
from app.domains.attribution.models import LeadAttribution, OrderAttribution
from app.domains.attribution.repository import AttributionRepository
from app.domains.enrollment.models import EnrollmentOrder
from app.domains.identity.models import H5User


class AttributionService:
    """第一阶段使用首次有效触点归因，订单创建时固化归因快照。"""

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

    def capture_visit(
        self,
        user: H5User,
        referral_code: str,
    ) -> dict[str, Any]:
        code = self._repository.get_promotion_code(referral_code.strip().upper())
        if code is None or code.status != "ACTIVE":
            raise AppError("REFERRAL_CODE_INVALID", "推广码无效或已停用", 422)
        now = self._clock()
        existing = self._repository.get_lead_by_user(user.id)
        if existing is None:
            lead = LeadAttribution(
                id=new_ulid(),
                h5_user_id=user.id,
                salesperson_id=code.salesperson_id,
                promotion_code_id=code.id,
                source_code=code.code,
                visit_count=1,
                first_touch_at=now,
                last_touch_at=now,
            )
        else:
            lead = replace(
                existing,
                visit_count=existing.visit_count + 1,
                last_touch_at=now,
            )
        self._repository.save_lead(lead)
        return {
            "source_code": lead.source_code,
            "attributed": True,
            "first_touch": existing is None,
        }

    def record_order(self, order: EnrollmentOrder) -> None:
        if self._repository.get_order_attribution(order.id) is not None:
            return
        lead = self._repository.get_lead_by_user(order.user_id)
        if lead is None:
            return
        self._repository.save_order_attribution(
            OrderAttribution(
                id=new_ulid(),
                order_id=order.id,
                h5_user_id=order.user_id,
                salesperson_id=lead.salesperson_id,
                promotion_code_id=lead.promotion_code_id,
                source_code=lead.source_code,
                amount_cents=order.amount_cents,
                attributed_at=self._clock(),
            )
        )

    def performance(self, admin_user_id: str | None = None) -> dict[str, Any]:
        salespersons = [
            item
            for item in self._repository.list_salespersons()
            if item.status == "ACTIVE"
            and (admin_user_id is None or item.admin_user_id == admin_user_id)
        ]
        salesperson_ids = {item.id for item in salespersons}
        leads = [
            item for item in self._repository.list_leads() if item.salesperson_id in salesperson_ids
        ]
        orders = [
            item
            for item in self._repository.list_order_attributions()
            if item.salesperson_id in salesperson_ids
        ]
        items = []
        for salesperson in salespersons:
            owned_leads = [item for item in leads if item.salesperson_id == salesperson.id]
            owned_orders = [item for item in orders if item.salesperson_id == salesperson.id]
            visit_count = sum(item.visit_count for item in owned_leads)
            lead_count = len(owned_leads)
            order_count = len(owned_orders)
            premium_cents = sum(item.amount_cents for item in owned_orders)
            items.append(
                {
                    "salesperson_id": salesperson.id,
                    "admin_user_id": salesperson.admin_user_id,
                    "salesperson_code": salesperson.code,
                    "salesperson_name": salesperson.name,
                    "visit_count": visit_count,
                    "lead_count": lead_count,
                    "order_count": order_count,
                    "premium_cents": premium_cents,
                    "visit_conversion_rate": round(lead_count / visit_count * 100, 1)
                    if visit_count
                    else 0.0,
                    "lead_conversion_rate": round(order_count / lead_count * 100, 1)
                    if lead_count
                    else 0.0,
                    "average_premium_cents": premium_cents // order_count if order_count else 0,
                }
            )
        items.sort(
            key=lambda item: (item["premium_cents"], item["order_count"]),
            reverse=True,
        )
        visit_count = sum(item.visit_count for item in leads)
        lead_count = len(leads)
        order_count = len(orders)
        premium_cents = sum(item.amount_cents for item in orders)
        return {
            "salesperson_count": len(salespersons),
            "visit_count": visit_count,
            "lead_count": lead_count,
            "order_count": order_count,
            "premium_cents": premium_cents,
            "visit_conversion_rate": round(lead_count / visit_count * 100, 1)
            if visit_count
            else 0.0,
            "lead_conversion_rate": round(order_count / lead_count * 100, 1) if lead_count else 0.0,
            "average_premium_cents": premium_cents // order_count if order_count else 0,
            "items": items,
        }

    def order_ids(self, admin_user_id: str | None = None) -> set[str]:
        salesperson_ids = {
            item.id
            for item in self._repository.list_salespersons()
            if admin_user_id is None or item.admin_user_id == admin_user_id
        }
        return {
            item.order_id
            for item in self._repository.list_order_attributions()
            if item.salesperson_id in salesperson_ids
        }

    def list_codes(self, admin_user_id: str | None = None) -> dict[str, Any]:
        salespersons = {
            item.id: item
            for item in self._repository.list_salespersons()
            if admin_user_id is None or item.admin_user_id == admin_user_id
        }
        items = [
            {
                "code": item.code,
                "channel": item.channel,
                "status": item.status,
                "salesperson_id": salesperson.id,
                "salesperson_code": salesperson.code,
                "salesperson_name": salesperson.name,
                "created_at": item.created_at,
            }
            for item in self._repository.list_promotion_codes()
            if (salesperson := salespersons.get(item.salesperson_id)) is not None
        ]
        return {"items": items, "total": len(items)}