Tolan/backend/services/energy_stone_service.py

"""
能量石发放与事件记录服务
"""
import math
import uuid
import logging
import time
from datetime import datetime, timezone
from typing import Any, Dict, List, Literal, Optional, Tuple

from sqlalchemy import select, func, text, update
from sqlalchemy.exc import IntegrityError

from database import db_manager
from models_db import (
    MissionEnergyStone,
    ConversationEnergyStone,
    MissionType,
    User,
)
from config import settings

logger = logging.getLogger(__name__)

StoneType = Literal["mission", "conversation"]


def _floor_non_negative_energy_stones(raw: object) -> int:
    """将 DECIMAL/浮点花费转为非负整数块数（向下取整）。"""
    if raw is None:
        return 0
    try:
        value = float(raw)
    except (TypeError, ValueError):
        return 0
    if value <= 0:
        return 0
    return int(math.floor(value))


class EnergyStoneService:
    """能量石生成/状态同步"""

    def __init__(self):
        self.db = db_manager
        raw_ttl = settings.ENERGY_STONE_TTL_SECONDS
        # 兼容被误配置为毫秒的场景（例如 20000/30000），自动转换为秒
        if raw_ttl > 600:  # 超过 10 分钟视为可能传入了毫秒
            converted = max(1, int(raw_ttl / 1000))
            logger.warning(
                "ENERGY_STONE_TTL_SECONDS=%s 看起来是毫秒，已按秒自动转换为 %s",
                raw_ttl,
                converted,
            )
            self.ttl_seconds = converted
        else:
            self.ttl_seconds = raw_ttl
        self.conversation_interval_ms = settings.CONVERSATION_STONE_INTERVAL_SECONDS * 1000
        self._mission_type_cache: Dict[str, Tuple[float, Optional[MissionType]]] = {}
        self._mission_type_cache_ttl = max(5, settings.MISSION_TYPE_CACHE_SECONDS)

    @staticmethod
    def _as_utc(dt: Optional[datetime]) -> Optional[datetime]:
        """将数据库取出的 naive 时间视为 UTC，避免本地时区偏移导致过长 TTL。"""
        if dt is None:
            return None
        if dt.tzinfo is None:
            return dt.replace(tzinfo=timezone.utc)
        return dt.astimezone(timezone.utc)

    async def set_db_manager(self, manager):
        """延迟注入数据库管理器"""
        self.db = manager

    async def create_mission_stone_if_needed(
        self,
        user_id: str,
        request_id: Optional[str],
        mission_type: str = "first_dialog",
        appear_reason: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None,
    ) -> Optional[Dict[str, Any]]:
        """首次对话奖励石，单用户仅发放一次"""
        if not user_id:
            return None

        try:
            async with self.db.get_session() as session:
                mission_key = mission_type or "first_dialog"
                reason = appear_reason or mission_key
                mission_config: Optional[MissionType] = None
                is_repeatable = False

                async with session.begin():
                    mission_config = await self._get_mission_type_config(session, mission_key)

                    if mission_config is not None and mission_config.is_active is False:
                        logger.debug(
                            "任务类型已禁用，跳过发放: user=%s mission=%s", user_id, mission_key
                        )
                        return None

                    is_repeatable = bool(mission_config.is_repeatable) if mission_config else False

                    result = await session.execute(
                        select(MissionEnergyStone.stone_id)
                        .where(
                            MissionEnergyStone.user_id == user_id,
                            MissionEnergyStone.mission_type == mission_key,
                        )
                        .with_for_update()
                    )
                    existing = result.scalars().first()
                    if existing and not is_repeatable:
                        return None

                    stone = MissionEnergyStone(
                        stone_id=str(uuid.uuid4()),
                        user_id=user_id,
                        request_id=request_id,
                        appear_reason=reason,
                        mission_type=mission_key,
                        metadata=metadata or {},
                    )
                    session.add(stone)
                    await session.execute(
                        update(User)
                        .where(User.user_id == user_id)
                        .values(crystal_stone_total=User.crystal_stone_total + 1)
                    )

                await session.refresh(stone)

                spawn_meta = {
                    "appear_reason": reason,
                    "mission_type": mission_key,
                    "is_repeatable": is_repeatable,
                }
                if mission_config:
                    spawn_meta["mission_display_name"] = mission_config.display_name

                return self._to_spawn_payload(
                    stone=stone,
                    stone_type="mission",
                    extra=spawn_meta,
                )
        except Exception as exc:
            logger.warning("创建任务能量石失败: user=%s error=%s", user_id, exc)
            return None

    async def create_conversation_stones_if_needed(
        self,
        user_id: str,
        request_id: Optional[str],
        total_duration_ms: int,
        duration_increment_ms: int,
    ) -> List[Dict[str, Any]]:
        """按累计说话时长发放对话能量石"""
        if not user_id or self.conversation_interval_ms <= 0:
            return []

        if total_duration_ms < self.conversation_interval_ms:
            return []

        target_index = total_duration_ms // self.conversation_interval_ms
        if target_index <= 0:
            return []

        stones: List[Dict[str, Any]] = []
        try:
            async with self.db.get_session() as session:
                result = await session.execute(
                    select(func.max(ConversationEnergyStone.threshold_index)).where(
                        ConversationEnergyStone.user_id == user_id
                    )
                )
                last_index = result.scalar() or 0
                next_index = last_index + 1

                if next_index > target_index:
                    return []

                created_instances: List[ConversationEnergyStone] = []
                for idx in range(next_index, target_index + 1):
                    instance = ConversationEnergyStone(
                        stone_id=str(uuid.uuid4()),
                        user_id=user_id,
                        request_id=request_id,
                        threshold_index=idx,
                        total_duration_ms=total_duration_ms,
                        duration_increment_ms=duration_increment_ms,
                        metadata={
                            "target_total_ms": total_duration_ms,
                            "interval_ms": self.conversation_interval_ms,
                        },
                    )
                    session.add(instance)
                    created_instances.append(instance)

                if not created_instances:
                    return []

                n_new = len(created_instances)
                await session.execute(
                    update(User)
                    .where(User.user_id == user_id)
                    .values(
                        conversation_energy_stone_total=User.conversation_energy_stone_total
                        + n_new,
                    )
                )
                await session.commit()
                for instance in created_instances:
                    await session.refresh(instance)
                    stones.append(
                        self._to_spawn_payload(
                            stone=instance,
                            stone_type="conversation",
                            extra={
                                "threshold_index": instance.threshold_index,
                                "total_duration_ms": instance.total_duration_ms,
                                "duration_increment_ms": instance.duration_increment_ms,
                            },
                        )
                    )
            return stones
        except IntegrityError:
            # 并发情况下可能已创建，忽略
            logger.debug("对话能量石已存在，忽略重复创建: user=%s target=%s", user_id, target_index)
            return []
        except Exception as exc:
            logger.warning("创建对话能量石失败: user=%s error=%s", user_id, exc)
            return []

    async def mark_event(
        self,
        stone_id: str,
        event: Literal["click", "timeout"],
        metadata: Optional[Dict[str, Any]] = None,
    ) -> bool:
        """记录点击或自动消失"""
        if not stone_id:
            return False

        try:
            async with self.db.get_session() as session:
                stone = await self._find_stone(session, stone_id)
                if not stone:
                    return False

                if metadata:
                    merged_meta = stone.metadata or {}
                    merged_meta.update(metadata)
                    stone.metadata = merged_meta

                now = datetime.now(timezone.utc)
                if event == "click":
                    stone.clicked_at = now
                    stone.auto_removed = False
                else:
                    stone.auto_removed = True
                    if not stone.clicked_at:
                        stone.clicked_at = now

                await session.commit()
                return True
        except Exception as exc:
            logger.warning("记录能量石事件失败: stone=%s event=%s error=%s", stone_id, event, exc)
            return False

    async def _find_stone(self, session, stone_id: str):
        result = await session.execute(
            select(MissionEnergyStone).where(MissionEnergyStone.stone_id == stone_id)
        )
        stone = result.scalars().first()
        if stone:
            return stone

        result = await session.execute(
            select(ConversationEnergyStone).where(ConversationEnergyStone.stone_id == stone_id)
        )
        return result.scalars().first()

    def _to_spawn_payload(
        self,
        stone,
        stone_type: StoneType,
        extra: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        appear_at = stone.appear_at or datetime.utcnow()
        appear_at_utc = self._as_utc(appear_at) or datetime.now(timezone.utc)
        ttl_seconds = max(1, int(self.ttl_seconds))
        payload = {
            "stone_id": stone.stone_id,
            "stone_type": stone_type,
            "ttl_seconds": ttl_seconds,
            "appear_at": int(appear_at_utc.timestamp() * 1000),
            "metadata": extra or {},
        }
        return payload

    async def _sum_conversation_energy_spent_shopping(self, session, user_id: str) -> int:
        """
        购物已消耗的对话能量石（块数，向下取整）。

        优先汇总 asset_purchase_logs.cost_energy（流水明细）；
        若表不存在（未部署商城流水），再尝试 purchase_orders.total_cost_energy（已完成订单）。
        user_id 兼容 users.id 外键或 UUID 字符串两种存法（见项目文档 chatbot_mysql5_建表.sql）。
        """
        # 1) 购买流水：按笔累计能量石花费
        try:
            result = await session.execute(
                text(
                    """
                    SELECT COALESCE(SUM(apl.cost_energy), 0)
                    FROM asset_purchase_logs apl
                    WHERE apl.user_id = (SELECT id FROM users WHERE user_id = :uuid LIMIT 1)
                       OR apl.user_id = :uuid
                    """
                ),
                {"uuid": user_id},
            )
            spent = _floor_non_negative_energy_stones(result.scalar_one())
            return spent
        except Exception as exc:
            logger.debug(
                "对话能量石购物消耗汇总(asset_purchase_logs)不可用，将尝试订单表: user=%s err=%s",
                user_id,
                exc,
            )

        # 2) 订单主表：仅统计已完成订单的总能量石花费（表或流水未建时使用）
        try:
            result = await session.execute(
                text(
                    """
                    SELECT COALESCE(SUM(po.total_cost_energy), 0)
                    FROM purchase_orders po
                    WHERE po.status = 'completed'
                      AND (
                        po.user_id = (SELECT id FROM users WHERE user_id = :uuid LIMIT 1)
                        OR po.user_id = :uuid
                      )
                    """
                ),
                {"uuid": user_id},
            )
            return _floor_non_negative_energy_stones(result.scalar_one())
        except Exception as exc:
            logger.debug("对话能量石购物消耗汇总(purchase_orders)不可用: user=%s err=%s", user_id, exc)
            return 0

    async def get_stone_counts(self, user_id: str) -> Dict[str, int]:
        """从 users 表读取展示余额（与发放/赠送/购物扣减同事务维护）。"""
        if not user_id:
            return {"mission_count": 0, "conversation_count": 0}

        try:
            async with self.db.get_session() as session:
                row = await session.execute(
                    select(User.crystal_stone_total, User.conversation_energy_stone_total).where(
                        User.user_id == user_id
                    )
                )
                tup = row.one_or_none()
                if not tup:
                    return {"mission_count": 0, "conversation_count": 0}
                crystal, conv = tup[0], tup[1]
                return {
                    "mission_count": max(0, int(crystal or 0)),
                    "conversation_count": max(0, int(conv or 0)),
                }
        except Exception as exc:
            logger.warning("获取能量石数量失败: user=%s error=%s", user_id, exc)
            return {"mission_count": 0, "conversation_count": 0}

    async def apply_shopping_deduction(
        self,
        user_id: str,
        energy_blocks: int = 0,
        crystal_blocks: int = 0,
    ) -> None:
        """
        购物流水写入成功后调用：扣减 users 表上的对话能量石/水晶石余额（不低于 0）。
        energy_blocks / crystal_blocks 为整颗数（与 cost_energy / cost_crystal 向下取整后一致）。
        """
        if not user_id:
            return
        e = max(0, int(energy_blocks))
        c = max(0, int(crystal_blocks))
        if e == 0 and c == 0:
            return
        try:
            async with self.db.get_session() as session:
                async with session.begin():
                    if e > 0:
                        await session.execute(
                            update(User)
                            .where(User.user_id == user_id)
                            .values(
                                conversation_energy_stone_total=func.greatest(
                                    0, User.conversation_energy_stone_total - e
                                ),
                            )
                        )
                    if c > 0:
                        await session.execute(
                            update(User)
                            .where(User.user_id == user_id)
                            .values(
                                crystal_stone_total=func.greatest(0, User.crystal_stone_total - c),
                            )
                        )
        except Exception as exc:
            logger.warning(
                "购物扣减用户能量石余额失败: user=%s energy=%s crystal=%s error=%s",
                user_id,
                e,
                c,
                exc,
            )
            raise

    async def get_active_stones(self, user_id: str) -> List[Dict[str, Any]]:
        """获取用户未过期的能量石（用于页面恢复）"""
        if not user_id:
            return []

        try:
            async with self.db.get_session() as session:
                now = datetime.utcnow()
                active_stones: List[Dict[str, Any]] = []

                # 查询未过期的任务能量石
                # 条件：clicked_at IS NULL 且 appear_at + ttl_seconds > now
                mission_result = await session.execute(
                    select(MissionEnergyStone).where(
                        MissionEnergyStone.user_id == user_id,
                        MissionEnergyStone.clicked_at.is_(None),
                    )
                )
                for stone in mission_result.scalars().all():
                    appear_at = self._as_utc(stone.appear_at or stone.created_at)
                    if appear_at:
                        expire_at = appear_at.timestamp() + self.ttl_seconds
                        now_ts = now.timestamp()
                        if expire_at > now_ts:
                            # 计算剩余 TTL（秒），并对异常值打日志
                            remaining_ttl = int(expire_at - now_ts)
                            if remaining_ttl > self.ttl_seconds * 2:
                                logger.warning(
                                    "Mission stone TTL too large, correcting: stone=%s appear_at=%s remaining=%s expected~%s",
                                    stone.stone_id,
                                    appear_at.isoformat(),
                                    remaining_ttl,
                                    self.ttl_seconds,
                                )
                                remaining_ttl = self.ttl_seconds
                            payload = self._to_spawn_payload(
                                stone=stone,
                                stone_type="mission",
                                extra={
                                    "appear_reason": stone.appear_reason,
                                    "mission_type": stone.mission_type,
                                },
                            )
                            # 更新 TTL 为剩余时间
                            payload["ttl_seconds"] = remaining_ttl
                            active_stones.append(payload)

                # 查询未过期的对话能量石
                conversation_result = await session.execute(
                    select(ConversationEnergyStone).where(
                        ConversationEnergyStone.user_id == user_id,
                        ConversationEnergyStone.clicked_at.is_(None),
                    )
                )
                for stone in conversation_result.scalars().all():
                    appear_at = self._as_utc(stone.appear_at or stone.created_at)
                    if appear_at:
                        expire_at = appear_at.timestamp() + self.ttl_seconds
                        now_ts = now.timestamp()
                        if expire_at > now_ts:
                            # 计算剩余 TTL（秒），并对异常值打日志
                            remaining_ttl = int(expire_at - now_ts)
                            if remaining_ttl > self.ttl_seconds * 2:
                                logger.warning(
                                    "Conversation stone TTL too large, correcting: stone=%s appear_at=%s remaining=%s expected~%s",
                                    stone.stone_id,
                                    appear_at.isoformat(),
                                    remaining_ttl,
                                    self.ttl_seconds,
                                )
                                remaining_ttl = self.ttl_seconds
                            payload = self._to_spawn_payload(
                                stone=stone,
                                stone_type="conversation",
                                extra={
                                    "threshold_index": stone.threshold_index,
                                    "total_duration_ms": stone.total_duration_ms,
                                    "duration_increment_ms": stone.duration_increment_ms,
                                },
                            )
                            # 更新 TTL 为剩余时间
                            payload["ttl_seconds"] = remaining_ttl
                            active_stones.append(payload)

                return active_stones
        except Exception as exc:
            logger.warning("获取未过期能量石失败: user=%s error=%s", user_id, exc)
            return []

    async def _get_mission_type_config(
        self, session, mission_type_key: str
    ) -> Optional[MissionType]:
        """获取任务类型配置，带简单缓存"""
        now = time.monotonic()
        cached = self._mission_type_cache.get(mission_type_key)
        if cached:
            cached_ts, cached_value = cached
            if now - cached_ts < self._mission_type_cache_ttl:
                return cached_value

        result = await session.execute(
            select(MissionType).where(
                MissionType.type_key == mission_type_key,
                MissionType.is_active.is_(True),
            )
        )
        mission_type = result.scalars().first()
        self._mission_type_cache[mission_type_key] = (now, mission_type)
        return mission_type


energy_stone_service = EnergyStoneService()