""" 主播服务 """ from sqlalchemy.orm import Session from sqlalchemy import and_, desc, func from typing import List, Optional, Dict, Any from ..models.streamer import StreamerProfile, StreamerStatus from ..models.user import User, UserRole from ..models.game import Chest from ..schemas.streamer import ( StreamerProfileCreate, StreamerProfileUpdate, StreamerStatistics, StreamerChestList, StreamerBetRecord ) class StreamerService: """主播服务""" @staticmethod def create_streamer_profile(db: Session, streamer_data: StreamerProfileCreate) -> StreamerProfile: """创建主播配置""" # 检查用户是否存在且角色为主播 user = db.query(User).filter(User.id == streamer_data.user_id).first() if not user: raise ValueError("用户不存在") if user.role != UserRole.STREAMER: raise ValueError("用户不是主播角色") # 检查是否已存在配置 existing = db.query(StreamerProfile).filter(StreamerProfile.user_id == streamer_data.user_id).first() if existing: raise ValueError("主播配置已存在") db_streamer = StreamerProfile(**streamer_data.dict()) db.add(db_streamer) db.commit() db.refresh(db_streamer) return db_streamer @staticmethod def get_streamer_profile(db: Session, streamer_id: int) -> Optional[StreamerProfile]: """获取主播配置""" return db.query(StreamerProfile).filter(StreamerProfile.id == streamer_id).first() @staticmethod def get_streamer_by_user_id(db: Session, user_id: int) -> Optional[StreamerProfile]: """根据用户ID获取主播配置""" return db.query(StreamerProfile).filter(StreamerProfile.user_id == user_id).first() @staticmethod def update_streamer_profile( db: Session, streamer: StreamerProfile, streamer_data: StreamerProfileUpdate ) -> StreamerProfile: """更新主播配置""" update_data = streamer_data.dict(exclude_unset=True) for field, value in update_data.items(): setattr(streamer, field, value) db.commit() db.refresh(streamer) return streamer @staticmethod def get_streamer_list( db: Session, skip: int = 0, limit: int = 20, status: Optional[str] = None, sort_by: str = "created_at", order: str = "desc" ) -> List[StreamerProfile]: """获取主播列表""" query = db.query(StreamerProfile) if status: query = query.filter(StreamerProfile.status == status) # 排序 if order.lower() == "desc": query = query.order_by(desc(getattr(StreamerProfile, sort_by))) else: query = query.order_by(getattr(StreamerProfile, sort_by)) return query.offset(skip).limit(limit).all() @staticmethod def get_streamer_statistics(db: Session, streamer_id: int) -> StreamerStatistics: """获取主播统计信息""" streamer = db.query(StreamerProfile).filter(StreamerProfile.id == streamer_id).first() if not streamer: raise ValueError("主播不存在") # 统计宝箱信息 total_chests = db.query(func.count(Chest.id)).filter( Chest.streamer_id == streamer.user_id ).scalar() or 0 active_chests = db.query(func.count(Chest.id)).filter( Chest.streamer_id == streamer.user_id, Chest.status == 0 # BETTING状态 ).scalar() or 0 # 查询已结算的宝箱以计算总奖金 settled_chests = db.query(Chest).filter( Chest.streamer_id == streamer.user_id, Chest.status == 3 # FINISHED状态 ).all() # 计算总获奖金额(所有已结算宝箱的奖金池总和) total_winnings = sum(int(chest.pool_a + chest.pool_b) for chest in settled_chests) # 计算抽成收益 total_commission = int(total_winnings * float(streamer.commission_rate) / 100) # 计算平均宝箱价值 average_chest_value = int(total_winnings / total_chests) if total_chests > 0 else 0 # 计算成功率(已结算宝箱中A选项获胜的比例) settled_count = len(settled_chests) if settled_count > 0: a_win_count = sum(1 for chest in settled_chests if chest.result == "A") success_rate = a_win_count / settled_count else: # 如果没有已结算的宝箱,则使用默认值 success_rate = 0.0 return StreamerStatistics( total_chests=total_chests, active_chests=active_chests, total_winnings=total_winnings, total_commission=total_commission, average_chest_value=average_chest_value, success_rate=success_rate ) @staticmethod def get_streamer_chests( db: Session, streamer_id: int, skip: int = 0, limit: int = 20, status: Optional[str] = None ) -> List[StreamerChestList]: """获取主播宝箱列表""" streamer = db.query(StreamerProfile).filter(StreamerProfile.id == streamer_id).first() if not streamer: raise ValueError("主播不存在") query = db.query(Chest).filter(Chest.streamer_id == streamer.user_id) if status: query = query.filter(Chest.status == status) chests = query.order_by(desc(Chest.created_at)).offset(skip).limit(limit).all() return [StreamerChestList.from_orm(chest) for chest in chests] @staticmethod def suspend_streamer(db: Session, streamer_id: int, reason: str = None) -> bool: """暂停主播""" streamer = db.query(StreamerProfile).filter(StreamerProfile.id == streamer_id).first() if not streamer: raise ValueError("主播不存在") streamer.status = StreamerStatus.SUSPENDED db.commit() return True @staticmethod def activate_streamer(db: Session, streamer_id: int) -> bool: """激活主播""" streamer = db.query(StreamerProfile).filter(StreamerProfile.id == streamer_id).first() if not streamer: raise ValueError("主播不存在") streamer.status = StreamerStatus.ACTIVE db.commit() return True @staticmethod def ban_streamer(db: Session, streamer_id: int, reason: str = None) -> bool: """封禁主播""" streamer = db.query(StreamerProfile).filter(StreamerProfile.id == streamer_id).first() if not streamer: raise ValueError("主播不存在") streamer.status = StreamerStatus.BANNED db.commit() return True @staticmethod def check_streamer_chest_limit(db: Session, streamer_id: int) -> Dict[str, Any]: """检查主播宝箱数量限制""" streamer = db.query(StreamerProfile).filter(StreamerProfile.id == streamer_id).first() if not streamer: raise ValueError("主播不存在") active_chests = db.query(func.count(Chest.id)).filter( Chest.streamer_id == streamer.user_id, Chest.status == 0 # BETTING状态 ).scalar() or 0 max_chests = streamer.max_active_chests return { "active_chests": active_chests, "max_chests": max_chests, "can_create": active_chests < max_chests, "remaining": max_chests - active_chests }