qihuaiflow/backend/python/app/api/admin/logs.py

"""
全平台日志与风控 API（管理员）
关键约束: 操作日志、Token日志、审计日志三个实体绝对不能写DELETE操作

增强功能:
- 审计日志: 按时间范围、用户、操作类型多条件过滤
- 操作日志: 关键词搜索和导出
- API调用统计: 端点调用次数、平均响应时间、错误率
- 慢请求记录: 响应时间超过阈值自动标记
- 统计概览API: /app/api/admin/logs/stats
"""
from fastapi import APIRouter, Depends, HTTPException, Query
from fastapi.responses import StreamingResponse
from sqlalchemy.orm import Session
from sqlalchemy import func, text
from ...database import get_db, SessionLocal
from ...models import (
    OperationLog, TokenUsageLog, Order, AuditLog, RiskRule,
    ApiCallStat, SlowRequestLog, User
)
from ...auth import get_current_admin
from ...middleware.log_service import LogService
import uuid
import csv
import io
from datetime import datetime, timedelta

router = APIRouter()


# ============================================================
# 日志概览统计 API
# ============================================================

@router.get("/stats")
async def get_log_stats(
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """获取日志概览数据（今日调用量、错误数、活跃用户等）"""
    today_start = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)

    # 今日API调用总量
    today_calls = db.query(func.sum(ApiCallStat.call_count)).filter(
        ApiCallStat.updated_at >= today_start
    ).scalar() or 0

    # 今日错误数
    today_errors = db.query(func.sum(ApiCallStat.error_count)).filter(
        ApiCallStat.updated_at >= today_start
    ).scalar() or 0

    # 今日操作日志数
    today_ops = db.query(OperationLog).filter(
        OperationLog.created_at >= today_start
    ).count()

    # 今日审计日志数
    today_audits = db.query(AuditLog).filter(
        AuditLog.created_at >= today_start
    ).count()

    # 今日活跃用户数（有操作记录的用户）
    today_active_users = db.query(func.count(func.distinct(OperationLog.user_id))).filter(
        OperationLog.created_at >= today_start
    ).scalar() or 0

    # 今日Token消耗总量
    today_tokens = db.query(func.sum(TokenUsageLog.total_tokens)).filter(
        TokenUsageLog.created_at >= today_start
    ).scalar() or 0

    # 慢请求数（今日）
    today_slow = db.query(SlowRequestLog).filter(
        SlowRequestLog.created_at >= today_start
    ).count()

    # 总用户数
    total_users = db.query(User).count()

    # 错误率
    error_rate = round(today_errors / max(today_calls, 1) * 100, 2)

    return {
        "code": 0,
        "data": {
            "today": {
                "api_calls": int(today_calls),
                "errors": int(today_errors),
                "error_rate": error_rate,
                "operation_logs": today_ops,
                "audit_logs": today_audits,
                "active_users": today_active_users,
                "total_tokens": int(today_tokens),
                "slow_requests": today_slow,
            },
            "total": {
                "users": total_users,
            },
        },
        "message": "ok",
    }


# ============================================================
# API调用统计
# ============================================================

@router.get("/api-stats")
async def get_api_call_stats(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    sort_by: str = Query("call_count", description="排序字段: call_count / avg_time / error_rate"),
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """获取API调用统计：各端点调用次数、平均响应时间、错误率"""
    query = db.query(ApiCallStat)

    # 排序
    if sort_by == "avg_time":
        query = query.order_by(ApiCallStat.total_response_time / func.greatest(ApiCallStat.call_count, 1))
    elif sort_by == "error_rate":
        query = query.order_by(
            (ApiCallStat.error_count * 1.0 / func.greatest(ApiCallStat.call_count, 1)).desc()
        )
    else:
        query = query.order_by(ApiCallStat.call_count.desc())

    total = query.count()
    stats = query.offset((page - 1) * page_size).limit(page_size).all()

    return {
        "code": 0,
        "data": {
            "total": total,
            "page": page,
            "page_size": page_size,
            "items": [
                {
                    "id": s.id,
                    "endpoint": s.endpoint,
                    "method": s.method,
                    "call_count": s.call_count,
                    "avg_response_time": round(s.total_response_time / max(s.call_count, 1), 4),
                    "error_count": s.error_count,
                    "error_rate": round(s.error_count / max(s.call_count, 1) * 100, 2),
                    "last_called_at": s.last_called_at.isoformat() if s.last_called_at else None,
                }
                for s in stats
            ],
        },
        "message": "ok",
    }


# ============================================================
# 慢请求记录
# ============================================================

@router.get("/slow-requests")
async def get_slow_requests(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    min_response_time: float = Query(None, description="最小响应时间过滤"),
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """获取慢请求记录列表"""
    query = db.query(SlowRequestLog)

    if min_response_time is not None:
        query = query.filter(SlowRequestLog.response_time >= min_response_time)

    total = query.count()
    logs = query.order_by(SlowRequestLog.created_at.desc()).offset(
        (page - 1) * page_size
    ).limit(page_size).all()

    return {
        "code": 0,
        "data": {
            "total": total,
            "page": page,
            "page_size": page_size,
            "items": [
                {
                    "id": log.id,
                    "request_id": log.request_id,
                    "method": log.method,
                    "endpoint": log.endpoint,
                    "status_code": log.status_code,
                    "response_time": log.response_time,
                    "threshold": log.threshold,
                    "created_at": log.created_at.isoformat() if log.created_at else None,
                }
                for log in logs
            ],
        },
        "message": "ok",
    }


# ============================================================
# 操作日志（增强: 关键词搜索 + 导出）
# ============================================================

@router.get("/operations")
async def get_operation_logs(
    user_id: str = Query(None, description="按用户ID筛选"),
    username: str = Query(None, description="按用户名筛选"),
    action: str = Query(None, description="按操作类型筛选"),
    keyword: str = Query(None, description="关键词搜索（匹配目标、详情）"),
    start_date: str = Query(None, description="开始日期 (YYYY-MM-DD)"),
    end_date: str = Query(None, description="结束日期 (YYYY-MM-DD)"),
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """获取操作日志列表（增强：支持关键词搜索和日期范围过滤）"""
    query = db.query(OperationLog)

    if user_id:
        query = query.filter(OperationLog.user_id == user_id)
    if username:
        query = query.filter(OperationLog.username.like(f"%{username}%"))
    if action:
        query = query.filter(OperationLog.action == action)
    if keyword:
        query = query.filter(
            (OperationLog.target.like(f"%{keyword}%")) |
            (OperationLog.detail.like(f"%{keyword}%"))
        )
    if start_date:
        try:
            sd = datetime.strptime(start_date, "%Y-%m-%d")
            query = query.filter(OperationLog.created_at >= sd)
        except ValueError:
            pass
    if end_date:
        try:
            ed = datetime.strptime(end_date, "%Y-%m-%d") + timedelta(days=1)
            query = query.filter(OperationLog.created_at < ed)
        except ValueError:
            pass

    total = query.count()
    logs = query.order_by(OperationLog.created_at.desc()).offset(
        (page - 1) * page_size
    ).limit(page_size).all()

    return {
        "code": 0,
        "data": {
            "total": total,
            "page": page,
            "page_size": page_size,
            "items": [
                {
                    "id": log.id,
                    "user_id": log.user_id,
                    "username": log.username,
                    "action": log.action,
                    "target": log.target,
                    "target_id": log.target_id,
                    "detail": log.detail,
                    "ip": log.ip,
                    "created_at": log.created_at.isoformat() if log.created_at else None,
                }
                for log in logs
            ],
        },
        "message": "ok",
    }


# ============================================================
# Token调用日志
# ============================================================

@router.get("/token-logs")
async def get_token_logs(
    user_id: str = Query(None, description="按用户ID筛选"),
    model_id: str = Query(None, description="按模型ID筛选"),
    keyword: str = Query(None, description="关键词搜索（匹配模型名称/应用名称）"),
    start_date: str = Query(None, description="开始日期 (YYYY-MM-DD)"),
    end_date: str = Query(None, description="结束日期 (YYYY-MM-DD)"),
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """获取Token调用日志列表"""
    query = db.query(TokenUsageLog)

    if user_id:
        query = query.filter(TokenUsageLog.user_id == user_id)
    if model_id:
        query = query.filter(TokenUsageLog.model_id == model_id)
    if keyword:
        query = query.filter(
            (TokenUsageLog.model_name.like(f"%{keyword}%")) |
            (TokenUsageLog.app_name.like(f"%{keyword}%"))
        )
    if start_date:
        try:
            sd = datetime.strptime(start_date, "%Y-%m-%d")
            query = query.filter(TokenUsageLog.created_at >= sd)
        except ValueError:
            pass
    if end_date:
        try:
            ed = datetime.strptime(end_date, "%Y-%m-%d") + timedelta(days=1)
            query = query.filter(TokenUsageLog.created_at < ed)
        except ValueError:
            pass

    total = query.count()
    logs = query.order_by(TokenUsageLog.created_at.desc()).offset(
        (page - 1) * page_size
    ).limit(page_size).all()

    return {
        "code": 0,
        "data": {
            "total": total,
            "page": page,
            "page_size": page_size,
            "items": [
                {
                    "id": log.id,
                    "user_id": log.user_id,
                    "model_id": log.model_id,
                    "model_name": log.model_name,
                    "app_id": log.app_id,
                    "app_name": log.app_name,
                    "conversation_id": log.conversation_id,
                    "prompt_tokens": log.prompt_tokens,
                    "completion_tokens": log.completion_tokens,
                    "total_tokens": log.total_tokens,
                    "cost": float(log.cost) if log.cost else 0,
                    "created_at": log.created_at.isoformat() if log.created_at else None,
                }
                for log in logs
            ],
        },
        "message": "ok",
    }


# ============================================================
# 付费订单日志
# ============================================================

@router.get("/payment-logs")
async def get_payment_logs(
    user_id: str = Query(None, description="按用户ID筛选"),
    status: str = Query(None, description="按状态筛选"),
    keyword: str = Query(None, description="关键词搜索（匹配项目名称）"),
    start_date: str = Query(None, description="开始日期 (YYYY-MM-DD)"),
    end_date: str = Query(None, description="结束日期 (YYYY-MM-DD)"),
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """获取付费订单日志列表"""
    query = db.query(Order)

    if user_id:
        query = query.filter(Order.user_id == user_id)
    if status:
        query = query.filter(Order.status == status)
    if keyword:
        query = query.filter(Order.item_name.like(f"%{keyword}%"))
    if start_date:
        try:
            sd = datetime.strptime(start_date, "%Y-%m-%d")
            query = query.filter(Order.created_at >= sd)
        except ValueError:
            pass
    if end_date:
        try:
            ed = datetime.strptime(end_date, "%Y-%m-%d") + timedelta(days=1)
            query = query.filter(Order.created_at < ed)
        except ValueError:
            pass

    total = query.count()
    orders = query.order_by(Order.created_at.desc()).offset(
        (page - 1) * page_size
    ).limit(page_size).all()

    return {
        "code": 0,
        "data": {
            "total": total,
            "page": page,
            "page_size": page_size,
            "items": [
                {
                    "id": o.id,
                    "user_id": o.user_id,
                    "order_type": o.order_type,
                    "item_id": o.item_id,
                    "item_name": o.item_name,
                    "amount": float(o.amount) if o.amount else 0,
                    "coupon_id": o.coupon_id,
                    "discount_amount": float(o.discount_amount) if o.discount_amount else 0,
                    "payment_method": o.payment_method,
                    "payment_id": o.payment_id,
                    "status": o.status,
                    "created_at": o.created_at.isoformat() if o.created_at else None,
                    "updated_at": o.updated_at.isoformat() if o.updated_at else None,
                }
                for o in orders
            ],
        },
        "message": "ok",
    }


# ============================================================
# 审计日志（增强：多条件过滤）
# ============================================================

@router.get("/audit-logs")
async def get_audit_logs(
    admin_id: str = Query(None, description="按管理员ID筛选"),
    admin_name: str = Query(None, description="按管理员名称筛选"),
    action: str = Query(None, description="按操作类型筛选"),
    target_type: str = Query(None, description="按目标类型筛选"),
    start_date: str = Query(None, description="开始日期 (YYYY-MM-DD)"),
    end_date: str = Query(None, description="结束日期 (YYYY-MM-DD)"),
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """获取审计日志列表（增强：支持按管理员、操作类型、目标类型、日期范围过滤）"""
    query = db.query(AuditLog)

    if admin_id:
        query = query.filter(AuditLog.admin_id == admin_id)
    if admin_name:
        query = query.filter(AuditLog.admin_name.like(f"%{admin_name}%"))
    if action:
        query = query.filter(AuditLog.action == action)
    if target_type:
        query = query.filter(AuditLog.target_type == target_type)
    if start_date:
        try:
            sd = datetime.strptime(start_date, "%Y-%m-%d")
            query = query.filter(AuditLog.created_at >= sd)
        except ValueError:
            pass
    if end_date:
        try:
            ed = datetime.strptime(end_date, "%Y-%m-%d") + timedelta(days=1)
            query = query.filter(AuditLog.created_at < ed)
        except ValueError:
            pass

    total = query.count()
    logs = query.order_by(AuditLog.created_at.desc()).offset(
        (page - 1) * page_size
    ).limit(page_size).all()

    return {
        "code": 0,
        "data": {
            "total": total,
            "page": page,
            "page_size": page_size,
            "items": [
                {
                    "id": log.id,
                    "admin_id": log.admin_id,
                    "admin_name": log.admin_name,
                    "action": log.action,
                    "target_type": log.target_type,
                    "target_id": log.target_id,
                    "detail": log.detail,
                    "created_at": log.created_at.isoformat() if log.created_at else None,
                }
                for log in logs
            ],
        },
        "message": "ok",
    }


# ============================================================
# 日志导出
# ============================================================

@router.get("/export/operations")
async def export_operations(
    user_id: str = Query(None),
    action: str = Query(None),
    keyword: str = Query(None),
    start_date: str = Query(None),
    end_date: str = Query(None),
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """导出操作日志（CSV）- 增强：支持与列表相同的过滤条件"""
    query = db.query(OperationLog)
    if user_id:
        query = query.filter(OperationLog.user_id == user_id)
    if action:
        query = query.filter(OperationLog.action == action)
    if keyword:
        query = query.filter(
            (OperationLog.target.like(f"%{keyword}%")) |
            (OperationLog.detail.like(f"%{keyword}%"))
        )
    if start_date:
        try:
            sd = datetime.strptime(start_date, "%Y-%m-%d")
            query = query.filter(OperationLog.created_at >= sd)
        except ValueError:
            pass
    if end_date:
        try:
            ed = datetime.strptime(end_date, "%Y-%m-%d") + timedelta(days=1)
            query = query.filter(OperationLog.created_at < ed)
        except ValueError:
            pass

    logs = query.order_by(OperationLog.created_at.desc()).all()

    output = io.StringIO()
    writer = csv.writer(output)
    writer.writerow(["ID", "用户ID", "用户名", "操作", "目标", "目标ID", "详情", "IP", "时间"])

    for log in logs:
        writer.writerow([
            log.id, log.user_id, log.username, log.action,
            log.target, log.target_id, log.detail, log.ip,
            log.created_at.isoformat() if log.created_at else "",
        ])

    output.seek(0)
    return StreamingResponse(
        iter([output.getvalue()]),
        media_type="text/csv",
        headers={
            "Content-Disposition": f"attachment; filename=operations_export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
        },
    )


@router.get("/export/token-logs")
async def export_token_logs(
    user_id: str = Query(None),
    keyword: str = Query(None),
    start_date: str = Query(None),
    end_date: str = Query(None),
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """导出Token日志（CSV）"""
    query = db.query(TokenUsageLog)
    if user_id:
        query = query.filter(TokenUsageLog.user_id == user_id)
    if keyword:
        query = query.filter(
            (TokenUsageLog.model_name.like(f"%{keyword}%")) |
            (TokenUsageLog.app_name.like(f"%{keyword}%"))
        )
    if start_date:
        try:
            sd = datetime.strptime(start_date, "%Y-%m-%d")
            query = query.filter(TokenUsageLog.created_at >= sd)
        except ValueError:
            pass
    if end_date:
        try:
            ed = datetime.strptime(end_date, "%Y-%m-%d") + timedelta(days=1)
            query = query.filter(TokenUsageLog.created_at < ed)
        except ValueError:
            pass

    logs = query.order_by(TokenUsageLog.created_at.desc()).all()

    output = io.StringIO()
    writer = csv.writer(output)
    writer.writerow([
        "ID", "用户ID", "模型ID", "模型名称", "应用ID", "应用名称",
        "会话ID", "Prompt Tokens", "Completion Tokens", "Total Tokens", "费用", "时间"
    ])

    for log in logs:
        writer.writerow([
            log.id, log.user_id, log.model_id, log.model_name,
            log.app_id, log.app_name, log.conversation_id,
            log.prompt_tokens, log.completion_tokens, log.total_tokens,
            float(log.cost) if log.cost else 0,
            log.created_at.isoformat() if log.created_at else "",
        ])

    output.seek(0)
    return StreamingResponse(
        iter([output.getvalue()]),
        media_type="text/csv",
        headers={
            "Content-Disposition": f"attachment; filename=token_logs_export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
        },
    )


@router.get("/export/audit-logs")
async def export_audit_logs(
    admin_id: str = Query(None),
    action: str = Query(None),
    target_type: str = Query(None),
    start_date: str = Query(None),
    end_date: str = Query(None),
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """导出审计日志（CSV）"""
    query = db.query(AuditLog)
    if admin_id:
        query = query.filter(AuditLog.admin_id == admin_id)
    if action:
        query = query.filter(AuditLog.action == action)
    if target_type:
        query = query.filter(AuditLog.target_type == target_type)
    if start_date:
        try:
            sd = datetime.strptime(start_date, "%Y-%m-%d")
            query = query.filter(AuditLog.created_at >= sd)
        except ValueError:
            pass
    if end_date:
        try:
            ed = datetime.strptime(end_date, "%Y-%m-%d") + timedelta(days=1)
            query = query.filter(AuditLog.created_at < ed)
        except ValueError:
            pass

    logs = query.order_by(AuditLog.created_at.desc()).all()

    output = io.StringIO()
    writer = csv.writer(output)
    writer.writerow(["ID", "管理员ID", "管理员名称", "操作", "目标类型", "目标ID", "详情", "时间"])

    for log in logs:
        writer.writerow([
            log.id, log.admin_id, log.admin_name, log.action,
            log.target_type, log.target_id, log.detail,
            log.created_at.isoformat() if log.created_at else "",
        ])

    output.seek(0)
    return StreamingResponse(
        iter([output.getvalue()]),
        media_type="text/csv",
        headers={
            "Content-Disposition": f"attachment; filename=audit_logs_export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
        },
    )


# ============================================================
# 操作类型枚举
# ============================================================

@router.get("/action-types")
async def get_action_types(
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """获取所有操作类型枚举值（用于筛选下拉）"""
    ops = db.query(OperationLog.action).distinct().all()
    audits = db.query(AuditLog.action).distinct().all()
    audit_targets = db.query(AuditLog.target_type).distinct().all()

    return {
        "operation_actions": [r[0] for r in ops if r[0]],
        "audit_actions": [r[0] for r in audits if r[0]],
        "audit_target_types": [r[0] for r in audit_targets if r[0]],
    }


# ============================================================
# 风控规则管理
# ============================================================

@router.get("/risk-rules")
async def get_risk_rules(
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """获取风控规则列表"""
    rules = db.query(RiskRule).order_by(RiskRule.created_at.desc()).all()
    return [
        {
            "id": r.id,
            "name": r.name,
            "type": r.type,
            "threshold": r.threshold,
            "duration": r.duration,
            "is_enabled": r.is_enabled,
            "created_at": r.created_at.isoformat() if r.created_at else None,
            "updated_at": r.updated_at.isoformat() if r.updated_at else None,
        }
        for r in rules
    ]


@router.post("/risk-rules")
async def create_risk_rule(
    name: str,
    rule_type: str,
    threshold: int = 0,
    duration: int = 3600,
    is_enabled: bool = True,
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """创建风控规则"""
    rule = RiskRule(
        id=str(uuid.uuid4()),
        name=name,
        type=rule_type,
        threshold=threshold,
        duration=duration,
        is_enabled=is_enabled,
    )
    db.add(rule)
    db.commit()

    LogService.log_audit(
        admin_id=admin_user.id,
        admin_name=admin_user.username,
        action="create_risk_rule",
        target_type="risk_rule",
        target_id=rule.id,
        detail=f"创建风控规则 {name}",
    )

    return {"message": "风控规则已创建", "id": rule.id}


@router.put("/risk-rules/{id}")
async def update_risk_rule(
    id: str,
    name: str = None,
    threshold: int = None,
    duration: int = None,
    is_enabled: bool = None,
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """更新风控规则"""
    rule = db.query(RiskRule).filter(RiskRule.id == id).first()
    if not rule:
        raise HTTPException(status_code=404, detail="风控规则不存在")

    if name is not None:
        rule.name = name
    if threshold is not None:
        rule.threshold = threshold
    if duration is not None:
        rule.duration = duration
    if is_enabled is not None:
        rule.is_enabled = is_enabled
    rule.updated_at = datetime.now()
    db.commit()

    LogService.log_audit(
        admin_id=admin_user.id,
        admin_name=admin_user.username,
        action="update_risk_rule",
        target_type="risk_rule",
        target_id=id,
        detail=f"更新风控规则 {rule.name}",
    )

    return {"message": "风控规则已更新"}


@router.delete("/risk-rules/{id}")
async def delete_risk_rule(
    id: str,
    admin_user=Depends(get_current_admin),
    db: Session = Depends(get_db),
):
    """删除风控规则"""
    rule = db.query(RiskRule).filter(RiskRule.id == id).first()
    if not rule:
        raise HTTPException(status_code=404, detail="风控规则不存在")

    db.delete(rule)
    db.commit()

    LogService.log_audit(
        admin_id=admin_user.id,
        admin_name=admin_user.username,
        action="delete_risk_rule",
        target_type="risk_rule",
        target_id=id,
        detail=f"删除风控规则 {rule.name}",
    )

    return {"message": "风控规则已删除"}