buff3.0/backend-admin/app/api/admin/dashboard.py

"""
仪表盘统计接口
"""
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from sqlalchemy import func, and_
from app.database import get_db
from app.api.deps import get_current_admin
from app.models.user import User
from app.models.software import Software
from app.models.card import Card
from app.models.api_log import APILog
from app.models.session import Session as UserSession
from app.background_tasks import get_task_manager
from datetime import datetime, timedelta
from app.utils.timezone import beijing_now

router = APIRouter()

@router.get("/stats")
async def get_dashboard_stats(
    db: Session = Depends(get_db),
    current_admin = Depends(get_current_admin)
):
    """获取仪表盘统计数据"""
    
    # 用户总数
    total_users = db.query(func.count(User.id)).scalar() or 0
    
    # 今日新增用户
    today_start = beijing_now().replace(hour=0, minute=0, second=0, microsecond=0)
    today_new_users = db.query(func.count(User.id)).filter(
        User.created_at >= today_start
    ).scalar() or 0
    
    # 软件数量
    total_software = db.query(func.count(Software.id)).scalar() or 0
    
    # 卡密总数
    total_cards = db.query(func.count(Card.id)).scalar() or 0
    
    # 已激活卡密数
    activated_cards = db.query(func.count(Card.id)).filter(
        Card.status == 1
    ).scalar() or 0
    
    # 未激活卡密数
    unactivated_cards = db.query(func.count(Card.id)).filter(
        Card.status == 0
    ).scalar() or 0
    
    # 日志总数
    total_logs = db.query(func.count(APILog.id)).scalar() or 0
    
    # 今日日志数
    today_logs = db.query(func.count(APILog.id)).filter(
        APILog.created_at >= today_start
    ).scalar() or 0
    
    # 有效用户数（未过期）
    active_users = db.query(func.count(User.id)).filter(
        and_(
            User.expire_time.isnot(None),
            User.expire_time > beijing_now()
        )
    ).scalar() or 0
    
    # 过期用户数
    expired_users = db.query(func.count(User.id)).filter(
        and_(
            User.expire_time.isnot(None),
            User.expire_time <= beijing_now()
        )
    ).scalar() or 0
    
    # 最近7天新增用户趋势
    seven_days_ago = beijing_now() - timedelta(days=7)
    recent_users = db.query(
        func.date(User.created_at).label('date'),
        func.count(User.id).label('count')
    ).filter(
        User.created_at >= seven_days_ago
    ).group_by(func.date(User.created_at)).all()
    
    user_trend = [
        {
            "date": str(row.date),
            "count": row.count
        }
        for row in recent_users
    ]
    
    # 最近7天日志趋势
    recent_logs = db.query(
        func.date(APILog.created_at).label('date'),
        func.count(APILog.id).label('count')
    ).filter(
        APILog.created_at >= seven_days_ago
    ).group_by(func.date(APILog.created_at)).all()
    
    log_trend = [
        {
            "date": str(row.date),
            "count": row.count
        }
        for row in recent_logs
    ]

    # 今日心跳/活跃人数（按会话 last_activity 去重 user_id）
    today_start_naive = beijing_now().replace(tzinfo=None, hour=0, minute=0, second=0, microsecond=0)
    heartbeat_today_users = db.query(
        func.count(func.distinct(UserSession.user_id))
    ).filter(
        and_(
            UserSession.is_active == 1,
            UserSession.last_activity >= today_start_naive
        )
    ).scalar() or 0

    heartbeat_today_sessions = db.query(func.count(UserSession.id)).filter(
        and_(
            UserSession.is_active == 1,
            UserSession.last_activity >= today_start_naive
        )
    ).scalar() or 0
    
    # 获取后台任务状态
    task_manager = get_task_manager()
    background_tasks_status = {
        "running": task_manager.running,
        "total_tasks": len(task_manager.tasks),
        "active_tasks": sum(1 for task in task_manager.tasks if not task.done()),
        "completed_tasks": sum(1 for task in task_manager.tasks if task.done()),
    }

    return {
        "code": 200,
        "msg": "成功",
        "data": {
            "heartbeat": {
                "today_users": heartbeat_today_users,
                "today_sessions": heartbeat_today_sessions
            },
            "users": {
                "total": total_users,
                "today_new": today_new_users,
                "active": active_users,
                "expired": expired_users,
                "trend": user_trend
            },
            "software": {
                "total": total_software
            },
            "cards": {
                "total": total_cards,
                "activated": activated_cards,
                "unactivated": unactivated_cards
            },
            "logs": {
                "total": total_logs,
                "today": today_logs,
                "trend": log_trend
            },
            "background_tasks": background_tasks_status
        }
    }


@router.get("/background-tasks/status")
async def get_background_tasks_status(
    db: Session = Depends(get_db),
    current_admin = Depends(get_current_admin)
):
    """获取后台任务状态详情"""
    task_manager = get_task_manager()

    # 获取各个任务的详细信息
    task_details = []
    for i, task in enumerate(task_manager.tasks):
        task_info = {
            "id": i + 1,
            "name": f"后台任务-{i + 1}",
            "status": "running" if not task.done() else "completed",
            "done": task.done(),
            "cancelled": task.cancelled(),
        }

        # 如果任务已完成，获取结果或异常信息
        if task.done():
            try:
                result = task.result()
                task_info["result"] = str(result) if result is not None else None
            except Exception as e:
                task_info["exception"] = str(e)

        task_details.append(task_info)

    return {
        "code": 200,
        "msg": "成功",
        "data": {
            "manager_running": task_manager.running,
            "total_tasks": len(task_manager.tasks),
            "tasks": task_details
        }
    }