from typing import Dict, Any, List, Optional from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, func, text, and_ from app.models.user import User, Product from app.models.team import Team, TeamMember from app.models.analytics import UsageLog from app.models.customer import Customer from app.models.quotation import Quotation from app.models.system_config import SystemConfig from datetime import datetime, timedelta import logging logger = logging.getLogger(__name__) class AdminService: def __init__(self, db: AsyncSession): self.db = db async def get_dashboard(self) -> Dict[str, Any]: now = datetime.utcnow() today_start = now.replace(hour=0, minute=0, second=0, microsecond=0) user_count = await self.db.execute(select(func.count(User.id))) team_count = await self.db.execute(select(func.count(Team.id))) customer_count = await self.db.execute(select(func.count(Customer.id))) quotation_count = await self.db.execute(select(func.count(Quotation.id))) today_logs = await self.db.execute( select(func.count(UsageLog.id)).where(UsageLog.created_at >= today_start) ) total_logs = await self.db.execute(select(func.count(UsageLog.id))) recent_users_result = await self.db.execute( select(User).order_by(User.created_at.desc()).limit(5) ) recent_users = recent_users_result.scalars().all() return { "users": { "total": user_count.scalar() or 0, }, "teams": { "total": team_count.scalar() or 0, }, "customers": { "total": customer_count.scalar() or 0, }, "quotations": { "total": quotation_count.scalar() or 0, }, "usage": { "today": today_logs.scalar() or 0, "total": total_logs.scalar() or 0, }, "recent_users": [ { "id": str(u.id), "username": u.username, "tier": u.tier, "is_active": u.is_active, "created_at": u.created_at.isoformat() if u.created_at else None, } for u in recent_users ], } def _user_to_dict(self, u: User) -> Dict[str, Any]: return { "id": str(u.id), "username": u.username, "phone": u.phone, "email": u.email, "tier": u.tier, "role": u.role, "is_active": u.is_active, "last_login_at": u.last_login_at.isoformat() if u.last_login_at else None, "login_count": u.login_count or 0, "created_at": u.created_at.isoformat() if u.created_at else None, "settings": u.settings, } async def list_users(self, page: int = 1, size: int = 20, role: Optional[str] = None) -> Dict[str, Any]: base_query = select(User) count_query = select(func.count(User.id)) if role: base_query = base_query.where(User.role == role) count_query = count_query.where(User.role == role) query = base_query.order_by(User.created_at.desc()).offset((page - 1) * size).limit(size) total = await self.db.execute(count_query) result = await self.db.execute(query) users = result.scalars().all() return { "items": [self._user_to_dict(u) for u in users], "total": total.scalar(), "page": page, "size": size, } async def update_user_tier(self, user_id: str, tier: str) -> bool: result = await self.db.execute(select(User).where(User.id == user_id)) user = result.scalar_one_or_none() if not user: return False user.tier = tier await self.db.flush() return True async def toggle_user_active(self, user_id: str) -> bool: result = await self.db.execute(select(User).where(User.id == user_id)) user = result.scalar_one_or_none() if not user: return False user.is_active = not user.is_active await self.db.flush() return True async def update_user_role(self, user_id: str, role: str) -> Optional[Dict[str, Any]]: if role not in ("user", "admin"): raise ValueError("Invalid role") result = await self.db.execute(select(User).where(User.id == user_id)) user = result.scalar_one_or_none() if not user: return None user.role = role await self.db.flush() return self._user_to_dict(user) async def get_user_detail(self, user_id: str) -> Optional[Dict[str, Any]]: result = await self.db.execute(select(User).where(User.id == user_id)) user = result.scalar_one_or_none() if not user: return None now = datetime.utcnow() today_start = now.replace(hour=0, minute=0, second=0, microsecond=0) today_start_date = today_start.date() product_count = await self.db.execute(select(func.count()).select_from(User.__table__).where(User.id == user_id)) prod_count = await self.db.execute( select(func.count(Product.id)).where(Product.user_id == user_id) ) cust_count = await self.db.execute( select(func.count(Customer.id)).where(Customer.user_id == user_id) ) quot_count = await self.db.execute( select(func.count(Quotation.id)).where(Quotation.user_id == user_id) ) usage_today = await self.db.execute( select(func.count(UsageLog.id)).where( UsageLog.user_id == user_id, UsageLog.created_at >= today_start ) ) usage_total = await self.db.execute( select(func.count(UsageLog.id)).where(UsageLog.user_id == user_id) ) return { **self._user_to_dict(user), "stats": { "products": prod_count.scalar() or 0, "customers": cust_count.scalar() or 0, "quotations": quot_count.scalar() or 0, "usage_today": usage_today.scalar() or 0, "usage_total": usage_total.scalar() or 0, }, } async def get_system_health(self) -> Dict[str, Any]: return { "status": "healthy", "version": "1.0.0", "timestamp": datetime.utcnow().isoformat(), } async def log_usage(self, user_id: str, action: str, detail: Dict = None, ip: str = None, ua: str = None): try: log = UsageLog( user_id=user_id, action=action, detail=detail or {}, ip_address=ip, user_agent=ua, ) self.db.add(log) await self.db.flush() except Exception as e: logger.warning(f"Failed to log usage: {e}") async def get_usage_stats(self) -> Dict[str, Any]: now = datetime.utcnow() today_start = now.replace(hour=0, minute=0, second=0, microsecond=0) week_ago = now - timedelta(days=7) by_action = await self.db.execute( select(UsageLog.action, func.count(UsageLog.id)) .where(UsageLog.created_at >= today_start) .group_by(UsageLog.action) .order_by(func.count(UsageLog.id).desc()) ) daily_result = await self.db.execute( text( "SELECT date_trunc('day', created_at) AS day, count(id) " "FROM usage_logs WHERE created_at >= :week_ago " "GROUP BY date_trunc('day', created_at) " "ORDER BY date_trunc('day', created_at)" ), {"week_ago": week_ago}, ) today_logs = await self.db.execute( select(func.count(UsageLog.id)).where(UsageLog.created_at >= today_start) ) dau = await self.db.execute( select(func.count(func.distinct(UsageLog.user_id))) .where(UsageLog.created_at >= today_start) ) total_users = await self.db.execute(select(func.count(User.id))) return { "today_total": today_logs.scalar() or 0, "dau": dau.scalar() or 0, "total_users": total_users.scalar() or 0, "by_action": {row[0]: row[1] for row in by_action.all()}, "daily_trend": [ {"date": str(row[0].date()), "count": row[1]} for row in daily_result.all() ], } async def get_logs( self, page: int = 1, size: int = 50, action: Optional[str] = None, user_id: Optional[str] = None, date_from: Optional[datetime] = None, date_to: Optional[datetime] = None, ) -> Dict[str, Any]: filters = [] if action: filters.append(UsageLog.action == action) if user_id: filters.append(UsageLog.user_id == user_id) if date_from: filters.append(UsageLog.created_at >= date_from) if date_to: filters.append(UsageLog.created_at <= date_to) count_query = select(func.count(UsageLog.id)) data_query = select(UsageLog).order_by(UsageLog.created_at.desc()) if filters: where_clause = and_(*filters) count_query = count_query.where(where_clause) data_query = data_query.where(where_clause) total = await self.db.execute(count_query) result = await self.db.execute( data_query.offset((page - 1) * size).limit(size) ) logs = result.scalars().all() return { "items": [ { "id": str(log.id), "user_id": str(log.user_id), "action": log.action, "detail": log.detail, "ip_address": log.ip_address, "user_agent": log.user_agent, "created_at": log.created_at.isoformat() if log.created_at else None, } for log in logs ], "total": total.scalar(), "page": page, "size": size, } async def _seed_default_configs(self): defaults = [ SystemConfig(key="ai_provider_translate", value={"primary": "sensenova", "fallback": ["openai", "local"]}, description="翻译任务 AI 模型选择"), SystemConfig(key="ai_provider_reply", value={"primary": "sensenova", "fallback": ["anthropic", "local"]}, description="回复建议 AI 模型选择"), SystemConfig(key="ai_provider_marketing", value={"primary": "sensenova", "fallback": ["openai", "local"]}, description="营销文案 AI 模型选择"), SystemConfig(key="ai_provider_extract", value={"primary": "sensenova", "fallback": ["openai"]}, description="信息提取 AI 模型选择"), SystemConfig(key="ai_provider_quotation", value={"primary": "sensenova", "fallback": ["openai"]}, description="报价单 AI 模型选择"), SystemConfig(key="feature_guest_mode", value={"enabled": True}, description="游客模式开关"), SystemConfig(key="feature_wechat_login", value={"enabled": False}, description="微信登录开关"), SystemConfig(key="feature_registration", value={"enabled": True}, description="新用户注册开关"), SystemConfig(key="free_daily_limits", value={"translate_chars": 5000, "replies": 20, "marketing": 5, "customers": 5, "products": 1, "quotations": 3}, description="免费版每日配额"), SystemConfig(key="pro_daily_limits", value={"translate_chars": 50000, "replies": 200, "marketing": 50, "customers": 100, "products": 20, "quotations": 30}, description="Pro 版每日配额"), ] for cfg in defaults: self.db.add(cfg) await self.db.flush() async def list_config(self) -> List[Dict[str, Any]]: result = await self.db.execute( select(func.count(SystemConfig.id)) ) if result.scalar() == 0: await self._seed_default_configs() result = await self.db.execute( select(SystemConfig).order_by(SystemConfig.key) ) configs = result.scalars().all() return [ { "key": c.key, "value": c.value, "description": c.description, "updated_at": c.updated_at.isoformat() if c.updated_at else None, } for c in configs ] async def update_config(self, key: str, value: Any) -> Optional[Dict[str, Any]]: result = await self.db.execute( select(SystemConfig).where(SystemConfig.key == key) ) config = result.scalar_one_or_none() if not config: return None config.value = value config.updated_at = datetime.utcnow() await self.db.flush() return { "key": config.key, "value": config.value, "description": config.description, "updated_at": config.updated_at.isoformat() if config.updated_at else None, } async def search_users(self, query: str) -> List[Dict[str, Any]]: result = await self.db.execute( select(User) .where( (User.username.ilike(f"%{query}%")) | (User.phone.ilike(f"%{query}%")) ) .order_by(User.created_at.desc()) .limit(20) ) users = result.scalars().all() return [ { "id": str(u.id), "username": u.username, "phone": u.phone, "tier": u.tier, "role": u.role, "is_active": u.is_active, "created_at": u.created_at.isoformat() if u.created_at else None, "settings": u.settings, } for u in users ]