baoxiang/backend/app/services/game_service.py

"""
游戏服务
"""
from typing import List
from sqlalchemy.orm import Session
from sqlalchemy import and_, update
from ..models.user import User
from ..models.game import Chest, Bet, ChestStatus
from ..schemas.game import ChestCreate, BetCreate
from ..utils.redis import acquire_lock, release_lock, update_pool_cache, get_pool_cache
from datetime import datetime


class GameService:
    """游戏服务"""

    @staticmethod
    def create_chest(db: Session, streamer: User, chest_data: ChestCreate) -> Chest:
        """
        创建宝箱
        """
        # 检查主播同时进行的宝箱数量（最多3个）
        active_chests = db.query(Chest).filter(
            and_(
                Chest.streamer_id == streamer.id,
                Chest.status.in_([ChestStatus.BETTING, ChestStatus.LOCKED, ChestStatus.SETTLING])
            )
        ).count()

        if active_chests >= 3:
            raise ValueError("同时进行的宝箱数量已达上限（3个）")

        # 创建宝箱
        db_chest = Chest(
            streamer_id=streamer.id,
            title=chest_data.title,
            option_a=chest_data.option_a,
            option_b=chest_data.option_b,
            countdown_seconds=chest_data.countdown_seconds
        )
        db.add(db_chest)
        db.commit()
        db.refresh(db_chest)

        # 初始化缓存
        update_pool_cache(db_chest.id, 0, 0)

        return db_chest

    @staticmethod
    def get_active_chests(db: Session, streamer_id: int = None) -> List[Chest]:
        """
        获取活跃宝箱
        """
        query = db.query(Chest).filter(Chest.status == ChestStatus.BETTING)
        if streamer_id:
            query = query.filter(Chest.streamer_id == streamer_id)
        return query.order_by(Chest.created_at.desc()).all()

    @staticmethod
    def get_chests_by_streamer(db: Session, streamer_id: int) -> List[Chest]:
        """
        获取指定主播的所有宝箱（包括历史宝箱）
        """
        return db.query(Chest).filter(Chest.streamer_id == streamer_id).order_by(Chest.created_at.desc()).all()

    @staticmethod
    def get_chest_by_id(db: Session, chest_id: int) -> Chest:
        """
        根据ID获取宝箱
        """
        return db.query(Chest).filter(Chest.id == chest_id).first()
    @staticmethod
    def place_bet(db: Session, user: User, bet_data: BetCreate) -> Bet:
        """
        下注
        """
        lock_key = f"lock:user:{user.id}:bet"
        lock_value = f"{user.id}:{int(datetime.now().timestamp())}"

        # 获取分布式锁
        if not acquire_lock(lock_key, lock_value, expire_seconds=10):
            raise ValueError("操作过于频繁，请稍后重试")

        try:
            # 验证宝箱状态
            chest = GameService.get_chest_by_id(db, bet_data.chest_id)
            if not chest or chest.status != ChestStatus.BETTING:
                raise ValueError("宝箱不可下注")

            # 检查余额
            if user.balance < bet_data.amount:
                raise ValueError("余额不足")

            # 检查用户是否已在该宝箱下注同一选项
            existing_bet = db.query(Bet).filter(
                and_(
                    Bet.user_id == user.id,
                    Bet.chest_id == bet_data.chest_id,
                    Bet.option == bet_data.option,
                    Bet.status == "PENDING"
                )
            ).first()

            # 如果用户已经对该选项下注，则累加金额而不是拒绝下注
            if existing_bet:
                # 更新现有下注金额
                existing_bet.amount += bet_data.amount
                db.commit()
                
                # 更新奖池
                if bet_data.option == "A":
                    chest.pool_a += bet_data.amount
                else:
                    chest.pool_b += bet_data.amount
                
                chest.total_bets += 1
                db.commit()
                
                # 更新缓存
                update_pool_cache(chest.id, chest.pool_a, chest.pool_b)
                
                # 扣款
                result = db.execute(
                    update(User).where(
                        and_(
                            User.id == user.id,
                            User.balance >= bet_data.amount,
                            User.version == user.version
                        )
                    ).values(
                        balance=User.balance - bet_data.amount,
                        version=User.version + 1
                    )
                )
                
                if result.rowcount == 0:
                    # 回滚之前的更改
                    existing_bet.amount -= bet_data.amount
                    db.commit()
                    raise ValueError("余额不足或并发冲突，请重试")
                
                # 刷新用户信息
                db.refresh(user)
                
                # 通知用户余额更新
                import asyncio
                try:
                    # 尝试获取当前事件循环
                    loop = asyncio.get_running_loop()
                    # 局部导入避免循环导入
                    from ..routers.websocket import notify_user_balance_update
                    loop.create_task(notify_user_balance_update(user.id, user.balance))
                except RuntimeError:
                    # 如果没有运行中的事件循环，则在新线程中处理
                    import threading
                    def run_async():
                        async def _run():
                            # 局部导入避免循环导入
                            from ..routers.websocket import notify_user_balance_update
                            await notify_user_balance_update(user.id, user.balance)
                        asyncio.run(_run())
                    thread = threading.Thread(target=run_async, daemon=True)
                    thread.start()
                
                # 记录交易
                from ..services.user_service import UserService
                UserService.create_transaction(
                    db=db,
                    user_id=user.id,
                    transaction_type="下注",
                    amount=-bet_data.amount,
                    balance_after=user.balance,
                    related_id=chest.id,
                    description=f"追加下注{chest.title}({bet_data.option}边)"
                )
                
                return existing_bet
            # 使用乐观锁扣款
            result = db.execute(
                update(User).where(
                    and_(
                        User.id == user.id,
                        User.balance >= bet_data.amount,
                        User.version == user.version
                    )
                ).values(
                    balance=User.balance - bet_data.amount,
                    version=User.version + 1
                )
            )

            if result.rowcount == 0:
                raise ValueError("余额不足或并发冲突，请重试")

            # 刷新用户信息
            db.refresh(user)
            
            # 通知用户余额更新
            import asyncio
            try:
                # 尝试获取当前事件循环
                loop = asyncio.get_running_loop()
                # 局部导入避免循环导入
                from ..routers.websocket import notify_user_balance_update
                loop.create_task(notify_user_balance_update(user.id, user.balance))
            except RuntimeError:
                # 如果没有运行中的事件循环，则在新线程中处理
                import threading
                def run_async():
                    async def _run():
                        # 局部导入避免循环导入
                        from ..routers.websocket import notify_user_balance_update
                        await notify_user_balance_update(user.id, user.balance)
                    asyncio.run(_run())
                thread = threading.Thread(target=run_async, daemon=True)
                thread.start()

            # 更新奖池
            if bet_data.option == "A":
                chest.pool_a += bet_data.amount
            else:
                chest.pool_b += bet_data.amount

            chest.total_bets += 1
            db.commit()

            # 更新缓存
            update_pool_cache(chest.id, chest.pool_a, chest.pool_b)

            # 创建下注记录
            db_bet = Bet(
                user_id=user.id,
                chest_id=chest.id,
                option=bet_data.option,
                amount=bet_data.amount,
                status="PENDING"
            )
            db.add(db_bet)
            db.commit()
            db.refresh(db_bet)

            # 记录交易
            from ..services.user_service import UserService
            UserService.create_transaction(
                db=db,
                user_id=user.id,
                transaction_type="下注",
                amount=-bet_data.amount,
                balance_after=user.balance,
                related_id=chest.id,
                description=f"下注{chest.title}({bet_data.option}边)"
            )

            return db_bet

        finally:
            # 释放锁
            release_lock(lock_key, lock_value)

    @staticmethod
    def lock_chest(db: Session, chest: Chest, streamer: User) -> bool:
        """
        封盘
        """
        if chest.streamer_id != streamer.id and streamer.role != "admin":
            raise ValueError("无权操作此宝箱")

        if chest.status != ChestStatus.BETTING:
            raise ValueError("宝箱状态不正确")

        chest.status = ChestStatus.LOCKED
        chest.locked_at = datetime.now()
        db.commit()

        return True

    @staticmethod
    def lock_expired_chest(db: Session, chest_id: int) -> Chest:
        """
        自动锁定过期的宝箱
        """
        chest = db.query(Chest).filter(Chest.id == chest_id).first()
        if not chest:
            return None

        if chest.status != ChestStatus.BETTING:
            return chest

        # 检查是否真的过期（增加一个小的容差值以确保准确性）
        # 使用本地时间进行计算（与数据库func.now()保持一致）
        from datetime import datetime
        elapsed = (datetime.now() - chest.created_at).total_seconds()
        if elapsed >= chest.countdown_seconds - 0.1:  # 0.1秒的容差
            chest.status = ChestStatus.LOCKED
            chest.locked_at = datetime.now()
            db.commit()
            db.refresh(chest)
            return chest

        return chest
    @staticmethod
    def settle_chest(db: Session, chest: Chest, result: str, streamer: User) -> bool:
        """
        结算宝箱
        """
        if chest.streamer_id != streamer.id and streamer.role != "admin":
            raise ValueError("无权操作此宝箱")

        if chest.status not in [ChestStatus.LOCKED, ChestStatus.SETTLING]:
            raise ValueError("宝箱状态不正确")

        # 锁定状态防止重复执行
        chest.status = ChestStatus.SETTLING
        db.commit()

        try:
            if result == "REFUND":
                # 流局退款
                GameService._refund_chest(db, chest)
            else:
                # 正常结算
                GameService._settle_winner(db, chest, result)

            chest.status = ChestStatus.FINISHED
            chest.result = result
            chest.settled_at = datetime.now()
            db.commit()

            return True

        except Exception as e:
            db.rollback()
            raise e

    @staticmethod
    def _settle_winner(db: Session, chest: Chest, winner: str) -> None:
        """
        结算获胜方
        """
        # 获取所有下注
        bets = db.query(Bet).filter(
            Bet.chest_id == chest.id,
            Bet.status == "PENDING"
        ).all()

        if not bets:
            return

        # 计算总奖池和抽水
        total_pool = chest.pool_a + chest.pool_b
        # 从数据库动态读取抽水比例
        from ..services.system_service import ConfigManager
        streamer_share = ConfigManager.get_streamer_share(db)
        platform_share = ConfigManager.get_platform_share(db)
        streamer_income = int(total_pool * streamer_share)
        platform_income = int(total_pool * platform_share)
        distributable = total_pool - streamer_income - platform_income

        # 主播分润
        streamer = db.query(User).filter(User.id == chest.streamer_id).first()
        if streamer:
            streamer.balance += streamer_income
            db.commit()

            from ..services.user_service import UserService
            UserService.create_transaction(
                db=db,
                user_id=streamer.id,
                transaction_type="抽水分润",
                amount=streamer_income,
                balance_after=streamer.balance,
                related_id=chest.id,
                description=f"宝箱《{chest.title}》抽水分润"
            )

        # 平台抽水 - 直接到管理员账户
        try:
            admin_user = UserService.get_admin_user(db)
            if admin_user and platform_income > 0:
                # 记录转账前的余额
                old_balance = admin_user.balance
                admin_user.balance += platform_income
                db.commit()

                try:
                    UserService.create_transaction(
                        db=db,
                        user_id=admin_user.id,
                        transaction_type="平台抽水",
                        amount=platform_income,
                        balance_after=admin_user.balance,
                        related_id=chest.id,
                        description=f"宝箱《{chest.title}》平台抽水收益"
                    )
                except Exception as e:
                    print(f"创建平台抽水交易记录失败: {str(e)}")
                    # 即使交易记录创建失败，也不应回滚余额变更
                    # 可以在这里添加告警通知或其他补偿措施
                
                # 通知管理员余额更新
                import asyncio
                try:
                    # 尝试获取当前事件循环
                    loop = asyncio.get_running_loop()
                    # 局部导入避免循环导入
                    from ..routers.websocket import notify_user_balance_update
                    loop.create_task(notify_user_balance_update(admin_user.id, admin_user.balance))
                except RuntimeError:
                    # 如果没有运行中的事件循环，则在新线程中处理
                    import threading
                    def run_async():
                        async def _run():
                            # 局部导入避免循环导入
                            from ..routers.websocket import notify_user_balance_update
                            await notify_user_balance_update(admin_user.id, admin_user.balance)
                        asyncio.run(_run())
                    thread = threading.Thread(target=run_async, daemon=True)
                    thread.start()
                
                print(f"平台抽水转账成功: {platform_income} 分至管理员账户 {admin_user.id}")
        except Exception as e:
            # 如果转账失败，记录错误并回滚
            db.rollback()
            print(f"平台抽水转账失败: {str(e)}")
            # 可以在这里添加更详细的错误处理逻辑，比如发送告警通知

        # 计算赔率 - 修改为只对获胜方抽水
        winner_pool = chest.pool_a if winner == "A" else chest.pool_b
        loser_pool = chest.pool_b if winner == "A" else chest.pool_a
        if winner_pool > 0:
            # 赔率计算改为：(获胜方奖池 + 失败方奖池 * 0.9) / 获胜方奖池
            odds = (winner_pool + loser_pool * 0.9) / winner_pool

            # 结算给获胜方
            for bet in bets:
                if bet.option == winner:
                    payout = int(bet.amount * odds)
                    bet.payout = payout
                    bet.status = "WON"

                    # 给用户加钱
                    user = db.query(User).filter(User.id == bet.user_id).first()
                    if user:
                        user.balance += payout
                        db.commit()
                        
                        # 通知用户余额更新
                        import asyncio
                        try:
                            # 尝试获取当前事件循环
                            loop = asyncio.get_running_loop()
                            # 局部导入避免循环导入
                            from ..routers.websocket import notify_user_balance_update
                            loop.create_task(notify_user_balance_update(user.id, user.balance))
                        except RuntimeError:
                            # 如果没有运行中的事件循环，则在新线程中处理
                            import threading
                            def run_async():
                                async def _run():
                                    # 局部导入避免循环导入
                                    from ..routers.websocket import notify_user_balance_update
                                    await notify_user_balance_update(user.id, user.balance)
                                asyncio.run(_run())
                            thread = threading.Thread(target=run_async, daemon=True)
                            thread.start()

                        # 记录交易
                        UserService.create_transaction(
                            db=db,
                            user_id=user.id,
                            transaction_type="获胜",
                            amount=payout,
                            balance_after=user.balance,
                            related_id=chest.id,
                            description=f"宝箱《{chest.title}》获胜奖金"
                        )
                else:
                    bet.status = "LOST"

            db.commit()

    @staticmethod
    def _refund_chest(db: Session, chest: Chest) -> None:
        """
        流局退款
        """
        bets = db.query(Bet).filter(
            Bet.chest_id == chest.id,
            Bet.status == "PENDING"
        ).all()

        for bet in bets:
            # 退款给用户
            user = db.query(User).filter(User.id == bet.user_id).first()
            if user:
                user.balance += bet.amount
                db.commit()
                
                # 通知用户余额更新
                import asyncio
                try:
                    # 尝试获取当前事件循环
                    loop = asyncio.get_running_loop()
                    # 局部导入避免循环导入
                    from ..routers.websocket import notify_user_balance_update
                    loop.create_task(notify_user_balance_update(user.id, user.balance))
                except RuntimeError:
                    # 如果没有运行中的事件循环，则在新线程中处理
                    import threading
                    def run_async():
                        async def _run():
                            # 局部导入避免循环导入
                            from ..routers.websocket import notify_user_balance_update
                            await notify_user_balance_update(user.id, user.balance)
                        asyncio.run(_run())
                    thread = threading.Thread(target=run_async, daemon=True)
                    thread.start()

                # 记录交易
                from ..services.user_service import UserService
                UserService.create_transaction(
                    db=db,
                    user_id=user.id,
                    transaction_type="退款",
                    amount=bet.amount,
                    balance_after=user.balance,
                    related_id=chest.id,
                    description=f"宝箱《{chest.title}》流局退款"
                )

            bet.status = "REFUNDED"

        # 重置奖池
        chest.pool_a = 0
        chest.pool_b = 0
        db.commit()

        # 更新缓存
        update_pool_cache(chest.id, 0, 0)

    @staticmethod
    def get_chest_bets(db: Session, chest_id: int) -> List[Bet]:
        """
        获取宝箱下注记录
        """
        return db.query(Bet).filter(Bet.chest_id == chest_id).all()