TelegramBotAdmin/wbb/admin/api.py

from __future__ import annotations

import csv
import io
import tempfile
import traceback
from contextlib import suppress
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any

from aiohttp import ClientError, ClientSession, ClientTimeout, web

import wbb
from wbb import MESSAGE_DUMP_CHAT
from wbb import app as telegram_app
from wbb.admin.bot_config import (
    BotConfigError,
    create_bot_profile,
    create_bot_role,
    delete_bot_profile,
    delete_bot_role,
    get_bot_profile_secrets,
    store_bot_identity,
    telegram_config_status,
    test_bot_token,
    update_bot_profile,
    update_bot_role,
    update_telegram_config,
)
from wbb.admin.security import (
    generate_csrf_token,
    generate_session_token,
    hash_password,
    hash_token,
    validate_new_password,
    verify_password,
)
from wbb.services.bot_permissions import api_permission, has_permission
from wbb.services.chat_management import (
    ChatManagementError,
    apply_automation_settings,
    create_invite_link,
    ensure_permission,
    execute_member_action,
    get_automation_settings,
    get_chat_overview,
    get_invite_links,
    list_accessible_chats,
    list_chat_admins,
    list_recent_members,
    revoke_invite_link,
    search_chat_members,
    send_announcement,
    update_chat_permissions,
    update_chat_profile,
)
from wbb.services.giveaways import (
    GiveawayServiceError,
    cancel_and_refund_giveaway,
    create_and_publish_giveaway,
    finish_and_publish_giveaway,
    remove_and_optionally_refund_participant,
    reroll_giveaway,
)
from wbb.services.point_settings import apply_point_rules
from wbb.utils.dbadmin import (
    create_admin_session,
    dashboard_counts,
    ensure_default_admin,
    get_admin_session,
    get_admin_user,
    list_audit_logs,
    list_member_identity_changes,
    record_audit,
    record_login_failure,
    record_login_success,
    revoke_admin_session,
    update_admin_password,
)
from wbb.utils.dbfunctions import get_rules, set_chat_rules
from wbb.utils.dbgiveaway import (
    add_giveaway_ban,
    get_giveaway,
    list_giveaway_bans,
    list_giveaways_page,
    list_participants_page,
    remove_giveaway_ban,
)
from wbb.utils.dbpoints import (
    SOURCE_ADMIN,
    InsufficientPoints,
    PointsError,
    adjust_points,
    get_point_rules,
    list_point_accounts,
    list_point_transactions,
    set_points,
)

API_PREFIX = "/api/admin/v1"
SESSION_COOKIE = "wbb_admin_session"
UNSAFE_METHODS = {"POST", "PUT", "PATCH", "DELETE"}
PUBLIC_API_PATHS = {f"{API_PREFIX}/auth/login", f"{API_PREFIX}/health"}
BOT_SCOPED_PREFIXES = (
    f"{API_PREFIX}/chats",
    f"{API_PREFIX}/giveaways",
    f"{API_PREFIX}/points",
    f"{API_PREFIX}/media",
)
TELEGRAM_ID_KEYS = {
    "chat_id",
    "user_id",
    "actor_id",
    "creator_id",
    "moderator_id",
    "target_id",
    "message_id",
    "removed_by",
}


class ApiProblem(RuntimeError):
    def __init__(
        self,
        code: str,
        message: str,
        *,
        status: int = 400,
        details: Any = None,
    ):
        super().__init__(message)
        self.code = code
        self.status = status
        self.details = details


def as_utc(value: datetime) -> datetime:
    if value.tzinfo is None:
        return value.replace(tzinfo=UTC)
    return value.astimezone(UTC)


def jsonable(value: Any, *, key: str = "") -> Any:
    if isinstance(value, datetime):
        return as_utc(value).isoformat().replace("+00:00", "Z")
    if isinstance(value, dict):
        return {
            item_key: jsonable(item_value, key=item_key)
            for item_key, item_value in value.items()
            if item_key != "_id"
        }
    if isinstance(value, (list, tuple)):
        return [jsonable(item) for item in value]
    if isinstance(value, int) and (key in TELEGRAM_ID_KEYS or key.endswith("_telegram_id")):
        return str(value)
    return value


def success(data: Any, *, status: int = 200) -> web.Response:
    return web.json_response({"data": jsonable(data)}, status=status)


def error_response(problem: ApiProblem) -> web.Response:
    return web.json_response(
        {
            "error": {
                "code": problem.code,
                "message": str(problem),
                "details": jsonable(problem.details),
            }
        },
        status=problem.status,
    )


def parse_int(value: Any, name: str, *, minimum: int | None = None) -> int:
    try:
        parsed = int(value)
    except (TypeError, ValueError) as exc:
        raise ApiProblem("invalid_parameter", f"{name} 必须是整数。") from exc
    if minimum is not None and parsed < minimum:
        raise ApiProblem("invalid_parameter", f"{name} 不能小于 {minimum}。")
    return parsed


def parse_datetime(value: Any, name: str) -> datetime:
    if not isinstance(value, str) or not value.strip():
        raise ApiProblem("invalid_parameter", f"{name} 必须是 UTC ISO 8601 时间。")
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError as exc:
        raise ApiProblem("invalid_parameter", f"{name} 不是有效时间。") from exc
    if parsed.tzinfo is None:
        raise ApiProblem("invalid_parameter", f"{name} 必须包含时区。")
    return parsed.astimezone(UTC)


async def json_body(request: web.Request) -> dict[str, Any]:
    try:
        body = await request.json()
    except Exception as exc:
        raise ApiProblem("invalid_json", "请求体必须是 JSON。") from exc
    if not isinstance(body, dict):
        raise ApiProblem("invalid_json", "请求体必须是 JSON 对象。")
    return body


def require_confirmation(body: dict[str, Any]) -> None:
    if body.get("confirm") is not True:
        raise ApiProblem(
            "confirmation_required", "该操作需要二次确认。", status=409
        )


def page_params(request: web.Request) -> tuple[int, int]:
    return (
        parse_int(request.query.get("page", 1), "page", minimum=1),
        min(parse_int(request.query.get("page_size", 20), "page_size", minimum=1), 100),
    )


def chat_id_param(request: web.Request) -> int:
    return parse_int(request.match_info["chat_id"], "chatId")


def set_audit(
    request: web.Request,
    action: str,
    *,
    chat_id: int | None = None,
    target_id: int | str | None = None,
    summary: str = "",
    metadata: dict[str, Any] | None = None,
) -> None:
    request["audit_context"] = {
        "action": action,
        "chat_id": chat_id,
        "target_id": target_id,
        "summary": summary,
        "metadata": metadata or {},
    }


@web.middleware
async def api_error_middleware(request: web.Request, handler):
    try:
        response = await handler(request)
    except ApiProblem as exc:
        await _record_request_audit(request, success_state=False, error=str(exc))
        return error_response(exc)
    except (ChatManagementError, GiveawayServiceError) as exc:
        problem = ApiProblem(
            exc.code, str(exc), status=getattr(exc, "status", 400)
        )
        await _record_request_audit(request, success_state=False, error=str(exc))
        return error_response(problem)
    except (PointsError, InsufficientPoints) as exc:
        problem = ApiProblem("points_error", str(exc), status=409)
        await _record_request_audit(request, success_state=False, error=str(exc))
        return error_response(problem)
    except BotConfigError as exc:
        status = {
            "bot_not_found": 404,
            "role_not_found": 404,
            "role_in_use": 409,
            "builtin_role_immutable": 409,
        }.get(exc.code, 400)
        problem = ApiProblem(exc.code, str(exc), status=status)
        await _record_request_audit(request, success_state=False, error=str(exc))
        return error_response(problem)
    except web.HTTPException:
        raise
    except Exception as exc:
        await _record_request_audit(request, success_state=False, error=str(exc))
        wbb.log.error(
            f"Admin API {request.method} {request.path} failed: {exc}\n"
            f"{traceback.format_exc()}"
        )
        return error_response(
            ApiProblem("internal_error", "服务器处理请求失败。", status=500)
        )
    await _record_request_audit(request, success_state=response.status < 400)
    return response


async def _record_request_audit(
    request: web.Request, *, success_state: bool, error: str = ""
) -> None:
    context = request.get("audit_context")
    if not context or request.get("audit_recorded"):
        return
    request["audit_recorded"] = True
    admin = request.get("admin") or {}
    with suppress(Exception):
        await record_audit(
            source="web",
            actor_id=admin.get("username", "anonymous"),
            actor_name=admin.get("username", "anonymous"),
            success=success_state,
            error=error,
            **context,
        )


@web.middleware
async def authentication_middleware(request: web.Request, handler):
    if not request.path.startswith(API_PREFIX) or request.path in PUBLIC_API_PATHS:
        return await handler(request)
    raw_token = request.cookies.get(SESSION_COOKIE, "")
    session = await get_admin_session(hash_token(raw_token)) if raw_token else None
    if not session:
        raise ApiProblem("unauthenticated", "登录已失效，请重新登录。", status=401)
    user = session["user"]
    request["admin"] = {
        "username": user["username"],
        "must_change_password": bool(user.get("must_change_password")),
        "csrf_token": session["csrf_token"],
        "session_token_hash": session["token_hash"],
    }
    allowed_during_password_change = {
        f"{API_PREFIX}/auth/me",
        f"{API_PREFIX}/auth/password",
        f"{API_PREFIX}/auth/logout",
    }
    if user.get("must_change_password") and request.path not in allowed_during_password_change:
        raise ApiProblem(
            "password_change_required", "首次登录必须修改初始密码。", status=428
        )
    if request.method in UNSAFE_METHODS:
        supplied = request.headers.get("X-CSRF-Token", "")
        if not supplied or supplied != session["csrf_token"]:
            raise ApiProblem("csrf_failed", "CSRF 校验失败。", status=403)
    return await handler(request)


def _selected_bot_id(request: web.Request) -> str:
    bot_id = str(request.headers.get("X-Bot-Id") or "").strip()
    if bot_id:
        return bot_id
    supervisor = getattr(wbb, "BOT_SUPERVISOR", None)
    if supervisor is not None:
        running = [
            key
            for key, value in supervisor.runtimes().items()
            if value.get("state") == "running"
        ]
        if len(running) == 1:
            return running[0]
    raise ApiProblem(
        "bot_selection_required",
        "请先选择要管理的机器人。",
        status=409,
    )


@web.middleware
async def bot_role_middleware(request: web.Request, handler):
    required = api_permission(request.method, request.path)
    if not required:
        return await handler(request)
    if bool(getattr(wbb, "SUPERVISOR_MODE", False)):
        profile = get_bot_profile_secrets(
            getattr(wbb, "BOT_PROFILES_PATH", "runtime/bot_profiles.json"),
            _selected_bot_id(request),
        )
        permissions = profile.get("permissions", [])
    else:
        permissions = getattr(wbb, "BOT_PERMISSIONS", {"*"})
    if not has_permission(required, permissions):
        raise ApiProblem(
            "bot_role_permission_denied",
            "所选机器人的职责角色不允许执行该操作。",
            status=403,
            details={"required_permission": required},
        )
    return await handler(request)


@web.middleware
async def bot_proxy_middleware(request: web.Request, handler):
    if not bool(getattr(wbb, "SUPERVISOR_MODE", False)) or not request.path.startswith(
        BOT_SCOPED_PREFIXES
    ):
        return await handler(request)
    supervisor = getattr(wbb, "BOT_SUPERVISOR", None)
    if supervisor is None:
        raise ApiProblem(
            "bot_supervisor_unavailable",
            "机器人监管服务尚未就绪。",
            status=503,
        )
    bot_id = _selected_bot_id(request)
    endpoint = supervisor.endpoint_for(bot_id)
    if endpoint is None:
        raise ApiProblem(
            "bot_not_running",
            "所选机器人当前未连接 Telegram。",
            status=409,
        )

    target = f"{endpoint}{request.rel_url}"
    headers = {
        key: value
        for key, value in request.headers.items()
        if key.lower() not in {"host", "content-length", "connection"}
    }
    body = await request.read()
    try:
        async with ClientSession(timeout=ClientTimeout(total=90)) as session:
            async with session.request(
                request.method,
                target,
                data=body or None,
                headers=headers,
                allow_redirects=False,
            ) as response:
                response_body = await response.read()
                response_headers = {
                    key: value
                    for key, value in response.headers.items()
                    if key.lower() in {"content-type", "content-disposition"}
                }
                return web.Response(
                    body=response_body,
                    status=response.status,
                    headers=response_headers,
                )
    except (ClientError, TimeoutError) as exc:
        raise ApiProblem(
            "bot_worker_unavailable",
            "机器人工作进程暂时不可用。",
            status=503,
        ) from exc


class AdminApi:
    def __init__(self) -> None:
        self.username = str(getattr(wbb, "ADMIN_WEB_USERNAME", "admin"))
        self.initial_password = str(
            getattr(wbb, "ADMIN_WEB_INITIAL_PASSWORD", "qwe0.123456")
        )
        self.session_hours = int(getattr(wbb, "ADMIN_WEB_SESSION_HOURS", 12))
        self.cookie_secure = bool(getattr(wbb, "ADMIN_WEB_COOKIE_SECURE", False))
        self.upload_limit = int(getattr(wbb, "ADMIN_WEB_UPLOAD_MAX_MB", 20)) * 1024 * 1024
        self.bot_config_path = Path(
            str(getattr(wbb, "BOT_PROFILES_PATH", "runtime/bot_profiles.json"))
        )

    async def initialize(self) -> None:
        await ensure_default_admin(
            username=self.username,
            password_hash=hash_password(self.initial_password),
        )

    def register(self, application: web.Application) -> None:
        router = application.router
        router.add_get(f"{API_PREFIX}/health", self.health)
        router.add_post(f"{API_PREFIX}/auth/login", self.login)
        router.add_get(f"{API_PREFIX}/auth/me", self.me)
        router.add_post(f"{API_PREFIX}/auth/logout", self.logout)
        router.add_put(f"{API_PREFIX}/auth/password", self.change_password)
        router.add_get(f"{API_PREFIX}/dashboard", self.dashboard)
        router.add_get(f"{API_PREFIX}/settings", self.system_settings)
        router.add_get(f"{API_PREFIX}/settings/telegram", self.telegram_settings)
        router.add_put(f"{API_PREFIX}/settings/telegram", self.telegram_settings_update)
        router.add_get(f"{API_PREFIX}/bots", self.bots)
        router.add_post(f"{API_PREFIX}/bots", self.bot_create)
        router.add_put(f"{API_PREFIX}/bots/{{bot_id}}", self.bot_update)
        router.add_delete(f"{API_PREFIX}/bots/{{bot_id}}", self.bot_delete)
        router.add_post(f"{API_PREFIX}/bots/{{bot_id}}/test", self.bot_test)
        router.add_post(f"{API_PREFIX}/bots/{{bot_id}}/restart", self.bot_restart)
        router.add_get(f"{API_PREFIX}/roles", self.roles)
        router.add_post(f"{API_PREFIX}/roles", self.role_create)
        router.add_put(f"{API_PREFIX}/roles/{{role_id}}", self.role_update)
        router.add_delete(f"{API_PREFIX}/roles/{{role_id}}", self.role_delete)
        router.add_get(f"{API_PREFIX}/audit-logs", self.audit_logs)
        router.add_post(f"{API_PREFIX}/media", self.upload_media)

        router.add_get(f"{API_PREFIX}/chats", self.chats)
        router.add_get(f"{API_PREFIX}/chats/{{chat_id}}", self.chat)
        router.add_patch(f"{API_PREFIX}/chats/{{chat_id}}/profile", self.chat_profile)
        router.add_put(f"{API_PREFIX}/chats/{{chat_id}}/permissions", self.chat_permissions)
        router.add_post(f"{API_PREFIX}/chats/{{chat_id}}/announcements", self.announcement)
        router.add_get(f"{API_PREFIX}/chats/{{chat_id}}/admins", self.chat_admins)
        router.add_get(f"{API_PREFIX}/chats/{{chat_id}}/members/search", self.member_search)
        router.add_get(f"{API_PREFIX}/chats/{{chat_id}}/members/recent", self.recent_members)
        router.add_get(
            f"{API_PREFIX}/chats/{{chat_id}}/members/identity-changes",
            self.member_identity_changes,
        )
        router.add_post(
            f"{API_PREFIX}/chats/{{chat_id}}/members/{{user_id}}/actions",
            self.member_action,
        )
        router.add_get(f"{API_PREFIX}/chats/{{chat_id}}/invites", self.invites)
        router.add_post(f"{API_PREFIX}/chats/{{chat_id}}/invites", self.invite_create)
        router.add_delete(f"{API_PREFIX}/chats/{{chat_id}}/invites", self.invite_revoke)
        router.add_get(f"{API_PREFIX}/chats/{{chat_id}}/rules", self.rules)
        router.add_put(f"{API_PREFIX}/chats/{{chat_id}}/rules", self.rules_update)
        router.add_get(f"{API_PREFIX}/chats/{{chat_id}}/automation", self.automation)
        router.add_put(f"{API_PREFIX}/chats/{{chat_id}}/automation", self.automation_update)

        router.add_get(f"{API_PREFIX}/chats/{{chat_id}}/points/settings", self.points_settings)
        router.add_put(f"{API_PREFIX}/chats/{{chat_id}}/points/settings", self.points_settings_update)
        router.add_get(f"{API_PREFIX}/points/accounts", self.point_accounts)
        router.add_get(f"{API_PREFIX}/points/leaderboard", self.point_leaderboard)
        router.add_get(f"{API_PREFIX}/points/transactions", self.point_transactions)
        router.add_post(f"{API_PREFIX}/points/adjustments", self.point_adjustment)
        router.add_get(f"{API_PREFIX}/points/export", self.points_export)

        router.add_get(f"{API_PREFIX}/giveaways", self.giveaways)
        router.add_post(f"{API_PREFIX}/giveaways", self.giveaway_create)
        router.add_get(f"{API_PREFIX}/giveaways/{{giveaway_id}}", self.giveaway_detail)
        router.add_post(f"{API_PREFIX}/giveaways/{{giveaway_id}}/finish", self.giveaway_finish)
        router.add_post(f"{API_PREFIX}/giveaways/{{giveaway_id}}/cancel", self.giveaway_cancel)
        router.add_post(f"{API_PREFIX}/giveaways/{{giveaway_id}}/reroll", self.giveaway_reroll)
        router.add_get(
            f"{API_PREFIX}/giveaways/{{giveaway_id}}/participants",
            self.giveaway_participants,
        )
        router.add_delete(
            f"{API_PREFIX}/giveaways/{{giveaway_id}}/participants/{{user_id}}",
            self.giveaway_participant_remove,
        )
        router.add_get(f"{API_PREFIX}/chats/{{chat_id}}/giveaway-bans", self.giveaway_bans)
        router.add_post(f"{API_PREFIX}/chats/{{chat_id}}/giveaway-bans", self.giveaway_ban_add)
        router.add_delete(
            f"{API_PREFIX}/chats/{{chat_id}}/giveaway-bans/{{user_id}}",
            self.giveaway_ban_remove,
        )
        router.add_get(f"{API_PREFIX}/giveaways/{{giveaway_id}}/export", self.giveaway_export)

    async def health(self, _: web.Request) -> web.Response:
        return success({"status": "ok"})

    def _set_session_cookie(self, response: web.Response, token: str) -> None:
        response.set_cookie(
            SESSION_COOKIE,
            token,
            httponly=True,
            secure=self.cookie_secure,
            samesite="Strict",
            max_age=self.session_hours * 3600,
            path="/",
        )

    async def _new_session(self, request: web.Request, username: str) -> tuple[str, str]:
        token = generate_session_token()
        csrf = generate_csrf_token()
        await create_admin_session(
            username=username,
            token_hash=hash_token(token),
            csrf_token=csrf,
            expires_at=datetime.now(UTC) + timedelta(hours=self.session_hours),
            remote_address=request.remote or "",
            user_agent=request.headers.get("User-Agent", ""),
        )
        return token, csrf

    async def login(self, request: web.Request) -> web.Response:
        body = await json_body(request)
        username = str(body.get("username") or "").strip()
        password = str(body.get("password") or "")
        set_audit(request, "auth.login", summary=f"Login as {username}")
        user = await get_admin_user(username)
        if user and int(user.get("failed_login_count", 0)) >= 5:
            last_failed = user.get("last_failed_login_at")
            if last_failed and datetime.now(UTC) - as_utc(last_failed) < timedelta(minutes=15):
                raise ApiProblem(
                    "login_rate_limited", "登录失败次数过多，请 15 分钟后再试。", status=429
                )
        if not user or not verify_password(password, user.get("password_hash", "")):
            await record_login_failure(username)
            raise ApiProblem("invalid_credentials", "用户名或密码错误。", status=401)
        await record_login_success(username)
        token, csrf = await self._new_session(request, username)
        response = success(
            {
                "username": username,
                "must_change_password": bool(user.get("must_change_password")),
                "csrf_token": csrf,
            }
        )
        self._set_session_cookie(response, token)
        request["admin"] = {"username": username}
        return response

    async def me(self, request: web.Request) -> web.Response:
        return success(
            {
                "username": request["admin"]["username"],
                "must_change_password": request["admin"]["must_change_password"],
                "csrf_token": request["admin"]["csrf_token"],
            }
        )

    async def logout(self, request: web.Request) -> web.Response:
        set_audit(request, "auth.logout", summary="Logout")
        await revoke_admin_session(request["admin"]["session_token_hash"])
        response = success({"logged_out": True})
        response.del_cookie(SESSION_COOKIE, path="/")
        return response

    async def change_password(self, request: web.Request) -> web.Response:
        body = await json_body(request)
        current = str(body.get("current_password") or "")
        new_password = str(body.get("new_password") or "")
        username = request["admin"]["username"]
        user = await get_admin_user(username)
        if not user or not verify_password(current, user["password_hash"]):
            raise ApiProblem("invalid_current_password", "当前密码错误。", status=409)
        if verify_password(new_password, user["password_hash"]):
            raise ApiProblem("password_unchanged", "新密码不能与当前密码相同。", status=409)
        errors = validate_new_password(new_password, username)
        if errors:
            raise ApiProblem("weak_password", "新密码不符合要求。", details=errors)
        set_audit(request, "auth.password.change", summary="Change administrator password")
        await update_admin_password(username, hash_password(new_password))
        token, csrf = await self._new_session(request, username)
        response = success(
            {"username": username, "must_change_password": False, "csrf_token": csrf}
        )
        self._set_session_cookie(response, token)
        return response

    async def dashboard(self, _: web.Request) -> web.Response:
        counts = await dashboard_counts()
        giveaways, giveaway_total = await list_giveaways_page(
            page=1,
            page_size=5,
            all_bots=bool(getattr(wbb, "SUPERVISOR_MODE", False)),
        )
        accounts, account_total = await list_point_accounts(page=1, page_size=5)
        transactions, transaction_total = await list_point_transactions(page=1, page_size=8)
        return success(
            {
                "counts": {
                    **counts,
                    "giveaways": giveaway_total,
                    "point_accounts": account_total,
                    "point_transactions": transaction_total,
                },
                "recent_giveaways": giveaways,
                "top_accounts": accounts,
                "recent_transactions": transactions,
            }
        )

    async def system_settings(self, _: web.Request) -> web.Response:
        bot_connected = bool(getattr(wbb, "TELEGRAM_CONNECTED", False))
        return success(
            {
                "web_address": f"http://127.0.0.1:{getattr(wbb, 'ADMIN_WEB_PORT', 8088)}/admin",
                "bind_host": str(getattr(wbb, "ADMIN_WEB_HOST", "0.0.0.0")),
                "upload_limit_mb": int(getattr(wbb, "ADMIN_WEB_UPLOAD_MAX_MB", 20)),
                "supervisor_mode": bool(getattr(wbb, "SUPERVISOR_MODE", False)),
                "bot": {
                    "connected": bot_connected,
                    "id": str(wbb.BOT_ID) if bot_connected else "",
                    "username": wbb.BOT_USERNAME if bot_connected else "",
                    "name": wbb.BOT_NAME if bot_connected else "",
                },
            }
        )

    def _supervisor(self):
        return getattr(wbb, "BOT_SUPERVISOR", None)

    def _telegram_status(self) -> dict[str, Any]:
        supervisor = self._supervisor()
        runtimes = supervisor.runtimes() if supervisor is not None else {}
        return telegram_config_status(self.bot_config_path, runtimes=runtimes)

    async def telegram_settings(self, _: web.Request) -> web.Response:
        return success(self._telegram_status())

    async def telegram_settings_update(self, request: web.Request) -> web.Response:
        body = await json_body(request)
        require_confirmation(body)
        set_audit(request, "telegram.settings.update", summary="Update Telegram API credentials")
        body.pop("confirm", None)
        update_telegram_config(self.bot_config_path, body)
        supervisor = self._supervisor()
        if supervisor is not None:
            status = telegram_config_status(self.bot_config_path)
            for profile in status["bots"]:
                if profile["enabled"] and profile["ready_to_connect"]:
                    await supervisor.reconcile(profile["bot_id"])
        return success(self._telegram_status())

    async def bots(self, _: web.Request) -> web.Response:
        return success({"items": self._telegram_status()["bots"]})

    async def bot_create(self, request: web.Request) -> web.Response:
        body = await json_body(request)
        require_confirmation(body)
        set_audit(request, "bot.create", summary=str(body.get("label") or ""))
        body.pop("confirm", None)
        profile = create_bot_profile(self.bot_config_path, body)
        supervisor = self._supervisor()
        if supervisor is not None and profile["enabled"] and profile["ready_to_connect"]:
            await supervisor.start(profile["bot_id"])
        current = next(
            item
            for item in self._telegram_status()["bots"]
            if item["bot_id"] == profile["bot_id"]
        )
        return success(current, status=201)

    async def bot_update(self, request: web.Request) -> web.Response:
        body = await json_body(request)
        require_confirmation(body)
        bot_id = request.match_info["bot_id"]
        set_audit(request, "bot.update", target_id=bot_id, summary="Update Bot profile")
        body.pop("confirm", None)
        update_bot_profile(self.bot_config_path, bot_id, body)
        supervisor = self._supervisor()
        if supervisor is not None:
            await supervisor.reconcile(bot_id)
        current = next(
            item for item in self._telegram_status()["bots"] if item["bot_id"] == bot_id
        )
        return success(current)

    async def bot_delete(self, request: web.Request) -> web.Response:
        body = await json_body(request)
        require_confirmation(body)
        bot_id = request.match_info["bot_id"]
        profile = get_bot_profile_secrets(self.bot_config_path, bot_id)
        set_audit(request, "bot.delete", target_id=bot_id, summary=str(profile.get("label") or ""))
        supervisor = self._supervisor()
        if supervisor is not None:
            await supervisor.stop(bot_id)
        delete_bot_profile(self.bot_config_path, bot_id)
        return success({"deleted": True})

    async def bot_test(self, request: web.Request) -> web.Response:
        body = await json_body(request)
        require_confirmation(body)
        bot_id = request.match_info["bot_id"]
        profile = get_bot_profile_secrets(self.bot_config_path, bot_id)
        set_audit(request, "bot.test", target_id=bot_id, summary="Validate Bot Token")
        identity = await test_bot_token(str(profile.get("bot_token") or ""))
        store_bot_identity(self.bot_config_path, bot_id, identity)
        return success(identity)

    async def bot_restart(self, request: web.Request) -> web.Response:
        body = await json_body(request)
        require_confirmation(body)
        bot_id = request.match_info["bot_id"]
        set_audit(request, "bot.restart", target_id=bot_id, summary="Restart Bot worker")
        supervisor = self._supervisor()
        if supervisor is None:
            raise ApiProblem(
                "bot_supervisor_unavailable",
                "机器人监管服务尚未就绪。",
                status=503,
            )
        return success(await supervisor.restart(bot_id))

    async def roles(self, _: web.Request) -> web.Response:
        status = self._telegram_status()
        return success(
            {
                "items": status["roles"],
                "permission_catalog": status["permission_catalog"],
            }
        )

    async def role_create(self, request: web.Request) -> web.Response:
        body = await json_body(request)
        require_confirmation(body)
        set_audit(
            request,
            "bot.role.create",
            summary=str(body.get("name") or ""),
        )
        body.pop("confirm", None)
        return success(create_bot_role(self.bot_config_path, body), status=201)

    async def role_update(self, request: web.Request) -> web.Response:
        body = await json_body(request)
        require_confirmation(body)
        role_id = request.match_info["role_id"]
        set_audit(
            request,
            "bot.role.update",
            target_id=role_id,
            summary=str(body.get("name") or ""),
        )
        body.pop("confirm", None)
        role = update_bot_role(self.bot_config_path, role_id, body)
        supervisor = self._supervisor()
        if supervisor is not None:
            for profile in self._telegram_status()["bots"]:
                if role_id in profile["role_ids"]:
                    await supervisor.reconcile(profile["bot_id"])
        return success(role)

    async def role_delete(self, request: web.Request) -> web.Response:
        body = await json_body(request)
        require_confirmation(body)
        role_id = request.match_info["role_id"]
        set_audit(
            request,
            "bot.role.delete",
            target_id=role_id,
            summary="Delete Bot role",
        )
        delete_bot_role(self.bot_config_path, role_id)
        return success({"deleted": True})

    async def audit_logs(self, request: web.Request) -> web.Response:
        page, page_size = page_params(request)
        chat_raw = request.query.get("chat_id")
        items, total = await list_audit_logs(
            chat_id=parse_int(chat_raw, "chat_id") if chat_raw else None,
            action=request.query.get("action") or None,
            page=page,
            page_size=page_size,
        )
        return success({"items": items, "total": total, "page": page, "page_size": page_size})

    async def upload_media(self, request: web.Request) -> web.Response:
        if not MESSAGE_DUMP_CHAT:
            raise ApiProblem(
                "message_dump_chat_missing",
                "请先配置媒体中转群。",
                status=409,
            )
        reader = await request.multipart()
        part = await reader.next()
        if not part or part.name != "file":
            raise ApiProblem("file_required", "请选择要上传的文件。")
        filename = Path(part.filename or "upload.bin").name
        content_type = part.headers.get("Content-Type", "application/octet-stream")
        size = 0
        temporary_path: Path | None = None
        try:
            with tempfile.NamedTemporaryFile(delete=False, suffix=Path(filename).suffix) as handle:
                temporary_path = Path(handle.name)
                while True:
                    chunk = await part.read_chunk(1024 * 1024)
                    if not chunk:
                        break
                    size += len(chunk)
                    if size > self.upload_limit:
                        raise ApiProblem("file_too_large", "上传文件超过 20 MB 限制。", status=413)
                    handle.write(chunk)
            if content_type == "image/gif":
                media_type = "animation"
                sent = await telegram_app.send_animation(MESSAGE_DUMP_CHAT, str(temporary_path))
                file_id = sent.animation.file_id
            elif content_type.startswith("image/"):
                media_type = "photo"
                sent = await telegram_app.send_photo(MESSAGE_DUMP_CHAT, str(temporary_path))
                file_id = sent.photo.file_id
            elif content_type.startswith("video/"):
                media_type = "video"
                sent = await telegram_app.send_video(MESSAGE_DUMP_CHAT, str(temporary_path))
                file_id = sent.video.file_id
            else:
                media_type = "document"
                sent = await telegram_app.send_document(
                    MESSAGE_DUMP_CHAT, str(temporary_path), file_name=filename
                )
                file_id = sent.document.file_id
        finally:
            if temporary_path:
                temporary_path.unlink(missing_ok=True)
        set_audit(request, "media.upload", summary=filename, target_id=file_id)
        return success(
            {
                "file_id": file_id,
                "type": media_type,
                "filename": filename,
                "size": size,
            },
            status=201,
        )

    async def chats(self, request: web.Request) -> web.Response:
        page, page_size = page_params(request)
        items, total = await list_accessible_chats(
            query=request.query.get("query", ""), page=page, page_size=page_size
        )
        return success({"items": items, "total": total, "page": page, "page_size": page_size})

    async def chat(self, request: web.Request) -> web.Response:
        return success(await get_chat_overview(chat_id_param(request)))

    async def chat_profile(self, request: web.Request) -> web.Response:
        chat_id = chat_id_param(request)
        body = await json_body(request)
        require_confirmation(body)
        set_audit(request, "chat.profile.update", chat_id=chat_id, summary="Update group profile")
        return success(
            await update_chat_profile(
                chat_id, title=body.get("title"), description=body.get("description")
            )
        )

    async def chat_permissions(self, request: web.Request) -> web.Response:
        chat_id = chat_id_param(request)
        body = await json_body(request)
        require_confirmation(body)
        set_audit(request, "chat.permissions.update", chat_id=chat_id, summary="Update default member permissions")
        return success(await update_chat_permissions(chat_id, body.get("permissions") or {}))

    async def announcement(self, request: web.Request) -> web.Response:
        chat_id = chat_id_param(request)
        body = await json_body(request)
        require_confirmation(body)
        set_audit(request, "chat.announcement.send", chat_id=chat_id, summary=str(body.get("text") or "")[:200])
        return success(
            await send_announcement(
                chat_id,
                text=str(body.get("text") or ""),
                media_type=body.get("media_type"),
                file_id=body.get("file_id"),
                pin=bool(body.get("pin")),
            ),
            status=201,
        )

    async def chat_admins(self, request: web.Request) -> web.Response:
        return success({"items": await list_chat_admins(chat_id_param(request))})

    async def member_search(self, request: web.Request) -> web.Response:
        return success(
            {
                "items": await search_chat_members(
                    chat_id_param(request),
                    request.query.get("query", ""),
                    limit=parse_int(request.query.get("limit", "20"), "limit"),
                )
            }
        )

    async def recent_members(self, request: web.Request) -> web.Response:
        return success(
            {
                "items": await list_recent_members(
                    chat_id_param(request),
                    limit=parse_int(request.query.get("limit", "30"), "limit"),
                )
            }
        )

    async def member_identity_changes(self, request: web.Request) -> web.Response:
        page, page_size = page_params(request)
        items, total = await list_member_identity_changes(
            chat_id_param(request), page=page, page_size=page_size
        )
        return success(
            {"items": items, "total": total, "page": page, "page_size": page_size}
        )

    async def member_action(self, request: web.Request) -> web.Response:
        chat_id = chat_id_param(request)
        user_id = parse_int(request.match_info["user_id"], "userId")
        body = await json_body(request)
        require_confirmation(body)
        action = str(body.get("action") or "")
        set_audit(
            request,
            f"chat.member.{action}",
            chat_id=chat_id,
            target_id=user_id,
            summary=str(body.get("reason") or ""),
        )
        duration = body.get("duration_seconds")
        return success(
            await execute_member_action(
                chat_id,
                user_id=user_id,
                action=action,
                reason=str(body.get("reason") or ""),
                duration_seconds=parse_int(duration, "duration_seconds", minimum=60) if duration else None,
                privileges=body.get("privileges") or {},
            )
        )

    async def invites(self, request: web.Request) -> web.Response:
        return success({"items": await get_invite_links(chat_id_param(request))})

    async def invite_create(self, request: web.Request) -> web.Response:
        chat_id = chat_id_param(request)
        body = await json_body(request)
        require_confirmation(body)
        expires_at = parse_datetime(body["expires_at"], "expires_at") if body.get("expires_at") else None
        member_limit = parse_int(body["member_limit"], "member_limit", minimum=1) if body.get("member_limit") else None
        set_audit(request, "chat.invite.create", chat_id=chat_id, summary=str(body.get("name") or ""))
        return success(
            await create_invite_link(
                chat_id,
                name=str(body.get("name") or "Admin panel"),
                expires_at=expires_at,
                member_limit=member_limit,
            ),
            status=201,
        )

    async def invite_revoke(self, request: web.Request) -> web.Response:
        chat_id = chat_id_param(request)
        body = await json_body(request)
        require_confirmation(body)
        invite_link = str(body.get("invite_link") or "")
        if not invite_link:
            raise ApiProblem("invite_link_required", "缺少邀请链接。")
        set_audit(request, "chat.invite.revoke", chat_id=chat_id, summary=invite_link)
        await revoke_invite_link(chat_id, invite_link)
        return success({"revoked": True})

    async def rules(self, request: web.Request) -> web.Response:
        return success({"rules": await get_rules(chat_id_param(request))})

    async def rules_update(self, request: web.Request) -> web.Response:
        chat_id = chat_id_param(request)
        body = await json_body(request)
        require_confirmation(body)
        await ensure_permission(chat_id, "can_change_info")
        rules = str(body.get("rules") or "")
        if len(rules) > 4000:
            raise ApiProblem("rules_too_long", "群规不能超过 4000 个字符。")
        set_audit(request, "chat.rules.update", chat_id=chat_id, summary="Update group rules")
        await set_chat_rules(chat_id, rules)
        return success({"rules": rules})

    async def automation(self, request: web.Request) -> web.Response:
        return success(await get_automation_settings(chat_id_param(request)))

    async def automation_update(self, request: web.Request) -> web.Response:
        chat_id = chat_id_param(request)
        body = await json_body(request)
        require_confirmation(body)
        set_audit(request, "chat.automation.update", chat_id=chat_id, summary="Update automation rules")
        body.pop("confirm", None)
        return success(await apply_automation_settings(chat_id, body))

    async def points_settings(self, request: web.Request) -> web.Response:
        return success(await get_point_rules(chat_id_param(request)))

    async def points_settings_update(self, request: web.Request) -> web.Response:
        chat_id = chat_id_param(request)
        body = await json_body(request)
        require_confirmation(body)
        await ensure_permission(chat_id, "can_change_info")
        set_audit(request, "points.settings.update", chat_id=chat_id, summary="Update point rules")
        body.pop("confirm", None)
        return success(await apply_point_rules(chat_id, body))

    async def point_accounts(self, request: web.Request) -> web.Response:
        page, page_size = page_params(request)
        chat_raw = request.query.get("chat_id")
        items, total = await list_point_accounts(
            chat_id=parse_int(chat_raw, "chat_id") if chat_raw else None,
            query=request.query.get("query", ""),
            page=page,
            page_size=page_size,
        )
        return success({"items": items, "total": total, "page": page, "page_size": page_size})

    async def point_leaderboard(self, request: web.Request) -> web.Response:
        if not request.query.get("chat_id"):
            raise ApiProblem("chat_id_required", "排行榜必须选择群组。")
        return await self.point_accounts(request)

    async def point_transactions(self, request: web.Request) -> web.Response:
        page, page_size = page_params(request)
        chat_raw = request.query.get("chat_id")
        user_raw = request.query.get("user_id")
        created_from = parse_datetime(request.query["created_from"], "created_from") if request.query.get("created_from") else None
        created_to = parse_datetime(request.query["created_to"], "created_to") if request.query.get("created_to") else None
        items, total = await list_point_transactions(
            chat_id=parse_int(chat_raw, "chat_id") if chat_raw else None,
            user_id=parse_int(user_raw, "user_id") if user_raw else None,
            source=request.query.get("source") or None,
            created_from=created_from,
            created_to=created_to,
            page=page,
            page_size=page_size,
        )
        return success({"items": items, "total": total, "page": page, "page_size": page_size})

    async def point_adjustment(self, request: web.Request) -> web.Response:
        body = await json_body(request)
        require_confirmation(body)
        chat_id = parse_int(body.get("chat_id"), "chat_id")
        user_id = parse_int(body.get("user_id"), "user_id")
        operation = str(body.get("operation") or "")
        amount = parse_int(body.get("amount"), "amount", minimum=0)
        reason = str(body.get("reason") or "").strip()
        if not reason:
            raise ApiProblem("reason_required", "积分调整必须填写原因。")
        if operation not in {"add", "deduct", "set"}:
            raise ApiProblem(
                "invalid_operation",
                "积分操作必须是增加、扣减或设置余额。",
            )
        try:
            user = await telegram_app.get_users(user_id)
            username, first_name = user.username, user.first_name
            display_name = " ".join(
                value for value in (user.first_name, user.last_name) if value
            )
        except Exception:
            username, first_name = body.get("username"), body.get("first_name")
            display_name = body.get("display_name") or first_name
        request_key = str(body.get("request_id") or generate_session_token())
        set_audit(
            request,
            f"points.adjust.{operation}",
            chat_id=chat_id,
            target_id=user_id,
            summary=reason,
            metadata={"amount": amount},
        )
        if operation == "set":
            account, created = await set_points(
                chat_id=chat_id,
                user_id=user_id,
                balance=amount,
                actor_id=request["admin"]["username"],
                reason=reason,
                idempotency_key=f"web-points:{request_key}",
                username=username,
                first_name=first_name,
                display_name=display_name,
            )
        else:
            account, created = await adjust_points(
                chat_id=chat_id,
                user_id=user_id,
                delta=amount if operation == "add" else -amount,
                source=SOURCE_ADMIN,
                idempotency_key=f"web-points:{request_key}",
                actor_id=request["admin"]["username"],
                reason=reason,
                username=username,
                first_name=first_name,
                display_name=display_name,
            )
        return success({"account": account, "created": created}, status=201 if created else 200)

    async def points_export(self, request: web.Request) -> web.Response:
        chat_raw = request.query.get("chat_id")
        if not chat_raw:
            raise ApiProblem("chat_id_required", "导出积分数据必须选择群组。")
        chat_id = parse_int(chat_raw, "chat_id")
        kind = request.query.get("kind", "accounts")
        output = io.StringIO()
        if kind == "accounts":
            items: list[dict[str, Any]] = []
            page = 1
            while True:
                batch, total = await list_point_accounts(
                    chat_id=chat_id, page=page, page_size=100
                )
                items.extend(batch)
                if not batch or len(items) >= total:
                    break
                page += 1
            fieldnames = ["chat_id", "user_id", "display_name", "username", "first_name", "balance", "lifetime_earned", "lifetime_spent", "updated_at"]
        elif kind == "transactions":
            items = []
            page = 1
            while True:
                batch, total = await list_point_transactions(
                    chat_id=chat_id, page=page, page_size=100
                )
                items.extend(batch)
                if not batch or len(items) >= total:
                    break
                page += 1
            fieldnames = ["transaction_id", "chat_id", "user_id", "display_name", "username", "first_name", "delta", "balance_after", "source", "actor_id", "reason", "reference_id", "created_at"]
        else:
            raise ApiProblem(
                "invalid_export_kind",
                "导出类型必须是积分账户或积分流水。",
            )
        writer = csv.DictWriter(output, fieldnames=fieldnames, extrasaction="ignore")
        writer.writeheader()
        for item in items:
            writer.writerow(jsonable(item))
        return web.Response(
            text=output.getvalue(),
            content_type="text/csv",
            headers={"Content-Disposition": f'attachment; filename="points-{chat_id}-{kind}.csv"'},
        )

    async def giveaways(self, request: web.Request) -> web.Response:
        page, page_size = page_params(request)
        chat_raw = request.query.get("chat_id")
        items, total = await list_giveaways_page(
            chat_id=parse_int(chat_raw, "chat_id") if chat_raw else None,
            status=request.query.get("status") or None,
            query=request.query.get("query", ""),
            page=page,
            page_size=page_size,
        )
        return success({"items": items, "total": total, "page": page, "page_size": page_size})

    async def giveaway_create(self, request: web.Request) -> web.Response:
        body = await json_body(request)
        require_confirmation(body)
        chat_id = parse_int(body.get("chat_id"), "chat_id")
        await ensure_permission(chat_id, "can_change_info")
        prizes = body.get("prizes")
        if not isinstance(prizes, list):
            raise ApiProblem("invalid_prizes", "奖项必须是数组。")
        set_audit(request, "giveaway.create", chat_id=chat_id, summary=str(body.get("title") or ""))
        giveaway = await create_and_publish_giveaway(
            chat_id=chat_id,
            creator_id=0,
            creator_name=request["admin"]["username"],
            title=str(body.get("title") or ""),
            description=str(body.get("description") or ""),
            prizes=prizes,
            ends_at=parse_datetime(body.get("ends_at"), "ends_at"),
            minimum_points=parse_int(body.get("minimum_points", 0), "minimum_points", minimum=0),
            entry_cost=parse_int(body.get("entry_cost", 0), "entry_cost", minimum=0),
            participation_reward=parse_int(body.get("participation_reward", 0), "participation_reward", minimum=0),
        )
        return success(giveaway, status=201)

    async def giveaway_detail(self, request: web.Request) -> web.Response:
        giveaway = await get_giveaway(request.match_info["giveaway_id"])
        if not giveaway:
            raise ApiProblem("giveaway_not_found", "未找到该抽奖。", status=404)
        return success(giveaway)

    async def giveaway_finish(self, request: web.Request) -> web.Response:
        body = await json_body(request)
        require_confirmation(body)
        giveaway_id = request.match_info["giveaway_id"]
        giveaway = await get_giveaway(giveaway_id)
        if not giveaway:
            raise ApiProblem("giveaway_not_found", "未找到该抽奖。", status=404)
        await ensure_permission(int(giveaway["chat_id"]), "can_change_info")
        set_audit(request, "giveaway.finish", chat_id=int(giveaway["chat_id"]), target_id=giveaway_id, summary="Manual draw")
        ok, message, current = await finish_and_publish_giveaway(giveaway_id)
        if not ok:
            raise ApiProblem("giveaway_not_running", message, status=409)
        return success(current)

    async def giveaway_cancel(self, request: web.Request) -> web.Response:
        body = await json_body(request)
        require_confirmation(body)
        giveaway_id = request.match_info["giveaway_id"]
        giveaway = await get_giveaway(giveaway_id)
        if not giveaway:
            raise ApiProblem("giveaway_not_found", "未找到该抽奖。", status=404)
        await ensure_permission(int(giveaway["chat_id"]), "can_change_info")
        set_audit(request, "giveaway.cancel", chat_id=int(giveaway["chat_id"]), target_id=giveaway_id, summary="Cancel and refund")
        ok, message, current = await cancel_and_refund_giveaway(
            giveaway_id, chat_id=int(giveaway["chat_id"])
        )
        if not ok:
            raise ApiProblem("giveaway_not_running", message, status=409)
        return success(current)

    async def giveaway_reroll(self, request: web.Request) -> web.Response:
        body = await json_body(request)
        require_confirmation(body)
        giveaway_id = request.match_info["giveaway_id"]
        giveaway = await get_giveaway(giveaway_id)
        if not giveaway:
            raise ApiProblem("giveaway_not_found", "未找到该抽奖。", status=404)
        await ensure_permission(int(giveaway["chat_id"]), "can_change_info")
        set_audit(request, "giveaway.reroll", chat_id=int(giveaway["chat_id"]), target_id=giveaway_id, summary=str(body.get("tier_name") or "All tiers"))
        winners, reroll_id = await reroll_giveaway(
            giveaway_id=giveaway_id,
            moderator_id=0,
            tier_name=body.get("tier_name") or None,
            reroll_id=body.get("request_id") or None,
        )
        return success({"winners": winners, "reroll_id": reroll_id})

    async def giveaway_participants(self, request: web.Request) -> web.Response:
        page, page_size = page_params(request)
        items, total = await list_participants_page(
            giveaway_id=request.match_info["giveaway_id"],
            active_only=request.query.get("active_only", "false").lower() == "true",
            page=page,
            page_size=page_size,
        )
        return success({"items": items, "total": total, "page": page, "page_size": page_size})

    async def giveaway_participant_remove(self, request: web.Request) -> web.Response:
        body = await json_body(request)
        require_confirmation(body)
        giveaway_id = request.match_info["giveaway_id"]
        user_id = parse_int(request.match_info["user_id"], "user_id")
        giveaway = await get_giveaway(giveaway_id)
        if not giveaway:
            raise ApiProblem("giveaway_not_found", "未找到该抽奖。", status=404)
        await ensure_permission(int(giveaway["chat_id"]), "can_change_info")
        set_audit(request, "giveaway.participant.remove", chat_id=int(giveaway["chat_id"]), target_id=user_id, summary=str(body.get("reason") or ""))
        removed = await remove_and_optionally_refund_participant(
            giveaway_id=giveaway_id,
            user_id=user_id,
            moderator_id=0,
            reason=str(body.get("reason") or "Removed in admin panel"),
            refund=bool(body.get("refund", True)),
        )
        if not removed:
            raise ApiProblem("participant_not_found", "参与者不存在或已被移除。", status=404)
        return success({"removed": True, "refunded": bool(body.get("refund", True))})

    async def giveaway_bans(self, request: web.Request) -> web.Response:
        page, page_size = page_params(request)
        items, total = await list_giveaway_bans(
            chat_id=chat_id_param(request), page=page, page_size=page_size
        )
        return success({"items": items, "total": total, "page": page, "page_size": page_size})

    async def giveaway_ban_add(self, request: web.Request) -> web.Response:
        chat_id = chat_id_param(request)
        body = await json_body(request)
        require_confirmation(body)
        user_id = parse_int(body.get("user_id"), "user_id")
        await ensure_permission(chat_id, "can_change_info")
        set_audit(request, "giveaway.ban.add", chat_id=chat_id, target_id=user_id, summary=str(body.get("reason") or ""))
        await add_giveaway_ban(chat_id=chat_id, user_id=user_id, moderator_id=0, reason=str(body.get("reason") or ""))
        return success({"created": True}, status=201)

    async def giveaway_ban_remove(self, request: web.Request) -> web.Response:
        chat_id = chat_id_param(request)
        user_id = parse_int(request.match_info["user_id"], "user_id")
        body = await json_body(request)
        require_confirmation(body)
        await ensure_permission(chat_id, "can_change_info")
        set_audit(request, "giveaway.ban.remove", chat_id=chat_id, target_id=user_id)
        return success({"removed": await remove_giveaway_ban(chat_id, user_id)})

    async def giveaway_export(self, request: web.Request) -> web.Response:
        giveaway_id = request.match_info["giveaway_id"]
        giveaway = await get_giveaway(giveaway_id)
        if not giveaway:
            raise ApiProblem("giveaway_not_found", "未找到该抽奖。", status=404)
        participants: list[dict[str, Any]] = []
        page = 1
        while True:
            batch, total = await list_participants_page(
                giveaway_id=giveaway_id, page=page, page_size=100
            )
            participants.extend(batch)
            if not batch or len(participants) >= total:
                break
            page += 1
        output = io.StringIO()
        fields = ["giveaway_id", "chat_id", "user_id", "display_name", "username", "first_name", "active", "entry_cost", "joined_at", "removed_at", "refunded_at"]
        writer = csv.DictWriter(output, fieldnames=fields, extrasaction="ignore")
        writer.writeheader()
        for participant in participants:
            writer.writerow(jsonable(participant))
        return web.Response(
            text=output.getvalue(),
            content_type="text/csv",
            headers={"Content-Disposition": f'attachment; filename="giveaway-{giveaway_id}.csv"'},
        )


def build_admin_application() -> web.Application:
    max_upload_mb = int(getattr(wbb, "ADMIN_WEB_UPLOAD_MAX_MB", 20))
    application = web.Application(
        middlewares=[
            api_error_middleware,
            authentication_middleware,
            bot_role_middleware,
            bot_proxy_middleware,
        ],
        client_max_size=(max_upload_mb + 1) * 1024 * 1024,
    )
    admin_api = AdminApi()
    application["admin_api"] = admin_api
    admin_api.register(application)
    return application