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

"""
日志管理接口
"""
from fastapi import APIRouter, Depends, Query, Request
from fastapi.responses import StreamingResponse
from sqlalchemy.orm import Session
from sqlalchemy import or_, and_
from app.database import get_db
from app.api.deps import get_current_admin
from app.models.api_log import APILog
from app.models.user import User
from app.models.admin import Admin
from datetime import datetime, timedelta
from app.utils.timezone import beijing_now
import csv
import io

router = APIRouter()

@router.get("")
async def get_logs(
    page: int = Query(1, ge=1),
    page_size: int = Query(10, ge=1, le=100),
    keyword: str = Query(None, description="搜索关键词（接口路径）"),
    # 支持后端的 user_id，也兼容前端传 username 字段
    user_id: int = Query(None, description="用户ID筛选"),
    username: str = Query(None, description="用户名筛选（兼容前端）"),
    # 支持后端的 endpoint，也兼容前端传 action 字段
    endpoint: str = Query(None, description="接口筛选"),
    action: str = Query(None, description="接口筛选（兼容前端）"),
    method: str = Query(None, description="方法筛选"),
    # 支持后端的 response_code，也兼容前端传 level（ERROR/WARN/INFO）
    response_code: int = Query(None, description="状态码筛选"),
    level: str = Query(None, description="级别筛选（ERROR/WARN/INFO）"),
    # 支持 start_time/end_time 和 前端的 start_date/end_date（YYYY-MM-DD）
    start_time: str = Query(None, description="开始时间"),
    end_time: str = Query(None, description="结束时间"),
    start_date: str = Query(None, description="开始日期（YYYY-MM-DD，兼容前端）"),
    end_date: str = Query(None, description="结束日期（YYYY-MM-DD，兼容前端）"),
    db: Session = Depends(get_db),
    current_admin = Depends(get_current_admin)
):
    """获取日志列表"""
    offset = (page - 1) * page_size
    
    query = db.query(APILog).filter(APILog.is_deleted == 0)

    # 搜索
    if keyword:
        query = query.filter(APILog.endpoint.like(f"%{keyword}%"))

    # 兼容前端：如果前端传了 username，则解析出 user_id（先查 users，再查 admins）
    if username:
        user = db.query(User).filter(User.username == username).first()
        if user:
            user_id = user.id
        else:
            admin_user = db.query(Admin).filter(Admin.username == username).first()
            if admin_user:
                # 这里用 admin 的 id 去匹配历史日志中记录的 user_id（历史数据中可能写入了 admin.id）
                user_id = admin_user.id
            else:
                # 用户不存在，返回空结果
                return {"code": 200, "msg": "成功", "data": [], "total": 0, "page": page, "page_size": page_size}

    # 筛选（优先使用前端 action 参数）
    if user_id is not None:
        query = query.filter(APILog.user_id == user_id)
    if action:
        query = query.filter(APILog.endpoint.like(f"%{action}%"))
    elif endpoint:
        query = query.filter(APILog.endpoint.like(f"%{endpoint}%"))
    if method:
        query = query.filter(APILog.method == method)

    # 兼容前端的 level 过滤：ERROR/WARN/INFO -> response_code 区间
    if level:
        lvl = level.upper()
        if lvl == "ERROR":
            query = query.filter(APILog.response_code >= 500)
        elif lvl == "WARN":
            query = query.filter(APILog.response_code >= 400, APILog.response_code < 500)
        elif lvl == "INFO":
            query = query.filter((APILog.response_code < 400) | (APILog.response_code == None))
    elif response_code is not None:
        query = query.filter(APILog.response_code == response_code)

    # 支持前端的 start_date/end_date（YYYY-MM-DD）
    if start_date and not start_time:
        start_time = start_date
    if end_date and not end_time:
        end_time = end_date

    if start_time:
        try:
            start_dt = datetime.fromisoformat(start_time.replace('Z', '+00:00'))
            query = query.filter(APILog.created_at >= start_dt)
        except:
            pass
    if end_time:
        try:
            end_dt = datetime.fromisoformat(end_time.replace('Z', '+00:00'))
            query = query.filter(APILog.created_at <= end_dt)
        except:
            pass
    
    # 排序和分页
    total = query.count()
    logs = query.order_by(APILog.created_at.desc()).offset(offset).limit(page_size).all()
    # resolve usernames for logs with user_id and admin_id to match frontend expectations
    user_ids = list({l.user_id for l in logs if l.user_id})
    admin_ids = list({l.admin_id for l in logs if getattr(l, "admin_id", None)})
    users_map = {}
    admins_map = {}
    if user_ids:
        users = db.query(User).filter(User.id.in_(user_ids)).all()
        users_map = {u.id: u.username for u in users}
    # admins map for admin_id (explicit) and fallback for historical user_id values that were admin ids
    if admin_ids:
        admins = db.query(Admin).filter(Admin.id.in_(admin_ids)).all()
        admins_map = {a.id: a.username for a in admins}
    # also attempt to resolve any user_ids that point to admins (historical)
    missing_user_ids = [uid for uid in (user_ids or []) if uid not in users_map]
    if missing_user_ids:
        admins = db.query(Admin).filter(Admin.id.in_(missing_user_ids)).all()
        for a in admins:
            users_map[a.id] = a.username

    def map_level(response_code: int) -> str:
        try:
            if response_code is None:
                return "INFO"
            if response_code >= 500:
                return "ERROR"
            if response_code >= 400:
                return "WARN"
            return "INFO"
        except:
            return "INFO"

    return {
        "code": 200,
        "msg": "成功",
        "data": [
            {
                "id": l.id,
                "user_id": l.user_id,
                "username": users_map.get(l.user_id) if l.user_id else None,
                "admin_id": getattr(l, "admin_id", None),
                "admin_username": admins_map.get(getattr(l, "admin_id", None)) if getattr(l, "admin_id", None) else None,
                # frontend expects 'action' field for interface path
                "action": l.endpoint,
                "method": l.method,
                "ip_address": l.ip_address,
                "user_agent": l.user_agent,
                # map response/request data to 'message' for frontend display
                "message": l.response_data or l.request_data,
                "level": map_level(l.response_code),
                "response_code": l.response_code,
                "execution_time": l.execution_time,
                "created_at": l.created_at.isoformat() if l.created_at else None
            }
            for l in logs
        ],
        "total": total,
        "page": page,
        "page_size": page_size
    }

@router.get("/{log_id}")
async def get_log_detail(
    log_id: int,
    db: Session = Depends(get_db),
    current_admin = Depends(get_current_admin)
):
    """获取日志详情"""
    log = db.query(APILog).filter(APILog.id == log_id).first()
    if not log:
        return {"code": 404, "msg": "日志不存在"}
    # resolve username if available (try users then admins as fallback)
    username = None
    admin_username = None
    if log.user_id:
        user = db.query(User).filter(User.id == log.user_id).first()
        if user:
            username = user.username
        else:
            admin_user = db.query(Admin).filter(Admin.id == log.user_id).first()
            if admin_user:
                username = admin_user.username
    if getattr(log, "admin_id", None):
        adm = db.query(Admin).filter(Admin.id == log.admin_id).first()
        if adm:
            admin_username = adm.username

    def map_level(response_code: int) -> str:
        try:
            if response_code is None:
                return "INFO"
            if response_code >= 500:
                return "ERROR"
            if response_code >= 400:
                return "WARN"
            return "INFO"
        except:
            return "INFO"

    return {
        "code": 200,
        "msg": "成功",
        "data": {
            "id": log.id,
            "user_id": log.user_id,
            "username": username,
            "admin_id": getattr(log, "admin_id", None),
            "admin_username": admin_username,
            "action": log.endpoint,
            "method": log.method,
            "message": log.response_data or log.request_data,
            "request_data": log.request_data,
            "response_data": log.response_data,
            "ip_address": log.ip_address,
            "user_agent": log.user_agent,
            "level": map_level(log.response_code),
            "response_code": log.response_code,
            "execution_time": log.execution_time,
            "created_at": log.created_at.isoformat() if log.created_at else None
        }
    }

@router.delete("")
async def clear_logs(
    request: Request,
    db: Session = Depends(get_db),
    current_admin = Depends(get_current_admin)
):
    """清空日志（软删除）"""
    try:
        from app.utils.history import create_api_log_history, create_operation_log

        # 先写历史表，再改/删：查出即将被软删除的日志，逐条写入 api_logs_his
        logs_to_clear = db.query(APILog).filter(APILog.is_deleted == 0).all()
        log_count = len(logs_to_clear)
        for log in logs_to_clear:
            create_api_log_history(db, log, "DELETE")

        db.query(APILog).filter(APILog.is_deleted == 0).update({"is_deleted": 1, "updated_at": beijing_now()})

        # 记录操作日志
        client_ip = request.client.host if request.client else None
        create_operation_log(
            db=db,
            admin_id=current_admin.id,
            module="日志管理",
            action="清空日志",
            description=f"清空API日志，共删除{log_count}条日志记录",
            old_data={"log_count": log_count},
            ip_address=client_ip
        )

        # 一次性提交所有更改
        db.commit()

        return {"code": 200, "msg": "清空成功"}
    except Exception as e:
        try:
            db.rollback()
        except Exception:
            pass
        logger.exception("清空日志失败")
        return {"code": 500, "msg": "系统异常，操作未完成，请稍后重试；若持续发生请联系客服"}

@router.get("/export/csv")
async def export_logs_csv(
    request: Request,
    keyword: str = Query(None),
    user_id: int = Query(None),
    endpoint: str = Query(None),
    method: str = Query(None),
    response_code: int = Query(None),
    start_time: str = Query(None),
    end_time: str = Query(None),
    db: Session = Depends(get_db),
    current_admin = Depends(get_current_admin)
):
    """导出日志为CSV"""
    query = db.query(APILog)

    # 应用相同的筛选条件
    if keyword:
        query = query.filter(APILog.endpoint.like(f"%{keyword}%"))
    if user_id is not None:
        query = query.filter(APILog.user_id == user_id)
    if endpoint:
        query = query.filter(APILog.endpoint.like(f"%{endpoint}%"))
    if method:
        query = query.filter(APILog.method == method)
    if response_code is not None:
        query = query.filter(APILog.response_code == response_code)
    if start_time:
        try:
            start_dt = datetime.fromisoformat(start_time.replace('Z', '+00:00'))
            query = query.filter(APILog.created_at >= start_dt)
        except:
            pass
    if end_time:
        try:
            end_dt = datetime.fromisoformat(end_time.replace('Z', '+00:00'))
            query = query.filter(APILog.created_at <= end_dt)
        except:
            pass

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

    # 记录操作日志
    from app.utils.history import create_operation_log
    client_ip = request.client.host if request.client else None
    create_operation_log(
        db=db,
        admin_id=current_admin.id,
        module="日志管理",
        action="导出日志CSV",
        description=f"导出API日志为CSV，共导出{len(logs)}条日志记录",
        new_data={
            "export_count": len(logs),
            "export_format": "CSV",
            "filters": {
                "keyword": keyword,
                "user_id": user_id,
                "endpoint": endpoint,
                "method": method,
                "response_code": response_code,
                "start_time": start_time,
                "end_time": end_time
            }
        },
        ip_address=client_ip
    )
    db.commit()

    # 生成CSV
    output = io.StringIO()
    writer = csv.writer(output)
    writer.writerow(['ID', '用户ID', '接口', '方法', 'IP地址', '用户代理', '状态码', '执行时间(ms)', '创建时间'])

    for log in logs:
        # 清理字段中的换行符，保留其他字符（不要手动对引号做二次转义，交给csv模块处理）
        def clean_field(value):
            if value is None:
                return ''
            # 转换为字符串，清理换行符
            return str(value).replace('\n', ' ').replace('\r', ' ')

        writer.writerow([
            log.id,
            log.user_id or '',
            log.endpoint,
            log.method,
            log.ip_address or '',
            clean_field(log.user_agent),
            log.response_code or '',
            log.execution_time or '',
            log.created_at.isoformat() if log.created_at else ''
        ])

    output.seek(0)
    # 为了让 Excel 正确识别 UTF-8 编码并按列分割，添加 UTF-8 BOM 并以 bytes 返回
    csv_bytes = output.getvalue().encode('utf-8-sig')
    return StreamingResponse(
        iter([csv_bytes]),
        media_type="text/csv; charset=utf-8",
        headers={"Content-Disposition": f"attachment; filename=logs_{beijing_now().strftime('%Y%m%d_%H%M%S')}.csv"}
    )

@router.get("/export/excel")
async def export_logs(
    request: Request,
    keyword: str = Query(None),
    user_id: int = Query(None),
    endpoint: str = Query(None),
    method: str = Query(None),
    response_code: int = Query(None),
    start_time: str = Query(None),
    end_time: str = Query(None),
    db: Session = Depends(get_db),
    current_admin = Depends(get_current_admin)
):
    """导出日志为Excel"""
    # 记录操作日志（在调用CSV导出前记录，因为CSV导出函数中也会记录）
    from app.utils.history import create_operation_log
    client_ip = request.client.host if request.client else None

    # 统计导出数量
    query = db.query(APILog)
    if keyword:
        query = query.filter(APILog.endpoint.like(f"%{keyword}%"))
    if user_id is not None:
        query = query.filter(APILog.user_id == user_id)
    if endpoint:
        query = query.filter(APILog.endpoint.like(f"%{endpoint}%"))
    if method:
        query = query.filter(APILog.method == method)
    if response_code is not None:
        query = query.filter(APILog.response_code == response_code)
    if start_time:
        try:
            start_dt = datetime.fromisoformat(start_time.replace('Z', '+00:00'))
            query = query.filter(APILog.created_at >= start_dt)
        except:
            pass
    if end_time:
        try:
            end_dt = datetime.fromisoformat(end_time.replace('Z', '+00:00'))
            query = query.filter(APILog.created_at <= end_dt)
        except:
            pass

    export_count = query.count()

    create_operation_log(
        db=db,
        admin_id=current_admin.id,
        module="日志管理",
        action="导出日志Excel",
        description=f"导出API日志为Excel，共导出{export_count}条日志记录",
        new_data={
            "export_count": export_count,
            "export_format": "Excel",
            "filters": {
                "keyword": keyword,
                "user_id": user_id,
                "endpoint": endpoint,
                "method": method,
                "response_code": response_code,
                "start_time": start_time,
                "end_time": end_time
            }
        },
        ip_address=client_ip
    )
    db.commit()

    # 暂时使用CSV格式，如果需要Excel格式可以安装openpyxl
    return await export_logs_csv(
        request=request,
        keyword=keyword,
        user_id=user_id,
        endpoint=endpoint,
        method=method,
        response_code=response_code,
        start_time=start_time,
        end_time=end_time,
        db=db,
        current_admin=current_admin
    )