languard-servers-manager/backend/core/servers/bans_router.py

"""Ban management endpoints — create, list, and revoke bans."""
from __future__ import annotations

import logging
from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, field_validator
from sqlalchemy.engine import Connection

from adapters.arma3.ban_manager import Arma3BanManager
from core.dal.ban_repository import BanRepository
from core.servers.service import ServerService
from database import get_db
from dependencies import get_current_user, require_admin

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/servers/{server_id}/bans", tags=["bans"])


def _ok(data):
    return {"success": True, "data": data, "error": None}


class CreateBanRequest(BaseModel):
    player_uid: str
    ban_type: str = "GUID"
    reason: str = ""
    duration_minutes: int = 0  # 0 = permanent

    @field_validator("ban_type")
    @classmethod
    def validate_ban_type(cls, v: str) -> str:
        if v not in ("GUID", "IP"):
            raise ValueError("ban_type must be 'GUID' or 'IP'")
        return v

    @field_validator("duration_minutes")
    @classmethod
    def validate_duration(cls, v: int) -> int:
        if v < 0:
            raise ValueError("duration_minutes cannot be negative")
        return v


@router.get("")
def list_bans(
    server_id: int,
    db: Annotated[Connection, Depends(get_db)],
    _user: Annotated[dict, Depends(get_current_user)],
) -> dict:
    """List all active bans for the server."""
    ServerService(db).get_server(server_id)  # raises 404 if not found
    ban_repo = BanRepository(db)
    bans = ban_repo.get_all(server_id=server_id)
    return _ok(bans)


@router.post("", status_code=status.HTTP_201_CREATED)
def create_ban(
    server_id: int,
    body: CreateBanRequest,
    db: Annotated[Connection, Depends(get_db)],
    _admin: Annotated[dict, Depends(require_admin)],
) -> dict:
    """Create a new ban. Writes to DB and syncs to bans.txt."""
    ServerService(db).get_server(server_id)  # raises 404 if not found
    ban_repo = BanRepository(db)

    # Calculate expires_at if duration is set
    expires_at = None
    if body.duration_minutes > 0:
        from datetime import datetime, timezone, timedelta
        expires_at = (
            datetime.now(timezone.utc) + timedelta(minutes=body.duration_minutes)
        ).isoformat()

    ban_id = ban_repo.create(
        server_id=server_id,
        guid=body.player_uid if body.ban_type == "GUID" else None,
        name=None,
        reason=body.reason,
        banned_by=_admin["username"],
        expires_at=expires_at,
        game_data={"ban_type": body.ban_type, "duration_minutes": body.duration_minutes},
    )
    db.commit()

    ban = ban_repo.get_by_id(ban_id)

    # Sync to bans.txt (non-blocking — log error but don't fail request)
    _sync_ban_to_file(server_id, body.player_uid, body.ban_type, body.reason, body.duration_minutes)

    return _ok(ban)


@router.delete("/{ban_id}")
def revoke_ban(
    server_id: int,
    ban_id: int,
    db: Annotated[Connection, Depends(get_db)],
    _admin: Annotated[dict, Depends(require_admin)],
) -> dict:
    """Revoke a ban (marks as inactive in DB, removes from bans.txt)."""
    ServerService(db).get_server(server_id)  # raises 404 if not found
    ban_repo = BanRepository(db)
    ban = ban_repo.get_by_id(ban_id)
    if ban is None or ban["server_id"] != server_id:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail={"code": "NOT_FOUND", "message": "Ban not found"},
        )
    ban_repo.deactivate(ban_id)
    db.commit()

    # Remove from bans.txt
    _remove_ban_from_file(server_id, ban.get("guid") or "")

    return _ok({"message": f"Ban {ban_id} revoked"})


# ── File sync helpers ──

def _sync_ban_to_file(
    server_id: int, identifier: str, ban_type: str, reason: str, duration_minutes: int
) -> None:
    """Write ban to bans.txt. Log error but don't fail the request."""
    try:
        mgr = Arma3BanManager(server_id)
        mgr.add_ban(identifier, ban_type, reason, duration_minutes)
    except Exception as exc:
        logger.error("Failed to sync ban to bans.txt for server %d: %s", server_id, exc)


def _remove_ban_from_file(server_id: int, identifier: str) -> None:
    """Remove ban from bans.txt. Log error but don't fail the request."""
    try:
        mgr = Arma3BanManager(server_id)
        mgr.remove_ban(identifier)
    except Exception as exc:
        logger.error("Failed to remove ban from bans.txt for server %d: %s", server_id, exc)