Nexus/server/api/terminal.py

"""Nexus — Terminal Quick Commands API
CRUD for terminal quick commands, persisted in MySQL.
Built-in commands (is_builtin=True) are seeded on startup and cannot be deleted/edited.
"""

from fastapi import APIRouter, Depends, HTTPException, Request
from pydantic import BaseModel, Field
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from server.api.dependencies import get_db
from server.api.auth_jwt import get_current_admin
from server.domain.models import QuickCommand, Admin, AuditLog
from server.infrastructure.database.audit_log_repo import AuditLogRepositoryImpl

router = APIRouter(prefix="/api/terminal", tags=["terminal"])


# ── Schemas ──

class QuickCommandCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=100, description="命令名称")
    cmd: str = Field(..., min_length=1, max_length=500, description="命令内容")


class QuickCommandUpdate(BaseModel):
    name: str | None = Field(None, min_length=1, max_length=100, description="命令名称")
    cmd: str | None = Field(None, min_length=1, max_length=500, description="命令内容")


# ── Built-in commands seed data ──

BUILTIN_COMMANDS = [
    {"name": "系统状态", "cmd": "systemctl status --no-pager 2>&1 | head -20\r", "sort_order": 1},
    {"name": "系统日志", "cmd": "journalctl -n 30 --no-pager 2>&1\r", "sort_order": 2},
    {"name": "磁盘使用", "cmd": "df -h\r", "sort_order": 3},
    {"name": "内存使用", "cmd": "free -h\r", "sort_order": 4},
    {"name": "进程列表", "cmd": "ps aux --sort=-%mem 2>&1 | head -15\r", "sort_order": 5},
]


async def seed_builtin_commands(db: AsyncSession) -> None:
    """Ensure built-in commands exist in DB. Called on app startup."""
    result = await db.execute(
        select(QuickCommand).where(QuickCommand.is_builtin.is_(True))
    )
    existing = {qc.name for qc in result.scalars().all()}

    for builtin in BUILTIN_COMMANDS:
        if builtin["name"] not in existing:
            qc = QuickCommand(
                name=builtin["name"],
                cmd=builtin["cmd"],
                is_builtin=True,
                sort_order=builtin["sort_order"],
            )
            db.add(qc)

    if len(existing) < len(BUILTIN_COMMANDS):
        await db.commit()


# ── Routes ──

@router.get("/quick-commands", response_model=list)
async def list_quick_commands(
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """List all quick commands (built-in + custom), sorted by sort_order then id"""
    result = await db.execute(
        select(QuickCommand)
        .order_by(QuickCommand.sort_order.asc(), QuickCommand.id.asc())
    )
    commands = result.scalars().all()
    return [_qc_to_dict(qc) for qc in commands]


@router.post("/quick-commands", response_model=dict, status_code=201)
async def create_quick_command(
    payload: QuickCommandCreate,
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Create a custom quick command"""
    # Get max sort_order for custom commands to append at end
    result = await db.execute(
        select(QuickCommand)
        .where(QuickCommand.is_builtin.is_(False))
        .order_by(QuickCommand.sort_order.desc())
        .limit(1)
    )
    last = result.scalar_one_or_none()
    next_order = (last.sort_order + 1) if last else (len(BUILTIN_COMMANDS) + 1)

    qc = QuickCommand(
        name=payload.name,
        cmd=payload.cmd,
        is_builtin=False,
        sort_order=next_order,
        created_by=admin.username,
    )
    db.add(qc)
    await db.commit()
    await db.refresh(qc)

    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username, action="create_quick_command",
        target_type="quick_command", target_id=qc.id,
        detail=f"名称={qc.name}", ip_address=request.client.host if request.client else "",
    ))

    return _qc_to_dict(qc)


@router.put("/quick-commands/{id}", response_model=dict)
async def update_quick_command(
    id: int,
    payload: QuickCommandUpdate,
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Update a custom quick command (built-in commands cannot be edited)"""
    result = await db.execute(select(QuickCommand).where(QuickCommand.id == id))
    qc = result.scalar_one_or_none()
    if not qc:
        raise HTTPException(status_code=404, detail="快捷命令不存在")
    if qc.is_builtin:
        raise HTTPException(status_code=403, detail="系统内置命令不可编辑")

    if payload.name is not None:
        qc.name = payload.name
    if payload.cmd is not None:
        qc.cmd = payload.cmd

    await db.commit()
    await db.refresh(qc)

    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username, action="update_quick_command",
        target_type="quick_command", target_id=id,
        detail=f"名称={qc.name}", ip_address=request.client.host if request.client else "",
    ))

    return _qc_to_dict(qc)


@router.delete("/quick-commands/{id}", status_code=204)
async def delete_quick_command(
    id: int,
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Delete a custom quick command (built-in commands cannot be deleted)"""
    result = await db.execute(select(QuickCommand).where(QuickCommand.id == id))
    qc = result.scalar_one_or_none()
    if not qc:
        raise HTTPException(status_code=404, detail="快捷命令不存在")
    if qc.is_builtin:
        raise HTTPException(status_code=403, detail="系统内置命令不可删除")

    await db.delete(qc)
    await db.commit()

    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username, action="delete_quick_command",
        target_type="quick_command", target_id=id,
        detail=f"名称={qc.name}", ip_address=request.client.host if request.client else "",
    ))


# ── Helper ──

def _qc_to_dict(qc: QuickCommand) -> dict:
    return {
        "id": qc.id,
        "name": qc.name,
        "cmd": qc.cmd,
        "is_builtin": qc.is_builtin,
        "sort_order": qc.sort_order,
        "created_by": qc.created_by,
        "created_at": str(qc.created_at) if qc.created_at else None,
    }