TelegramBotAdmin/wbb/modules/admin_panel.py

from __future__ import annotations

import secrets
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from html import escape
from typing import Any

from pyrogram import filters
from pyrogram.enums import ParseMode
from pyrogram.types import CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup, Message

from wbb import BOT_PERMISSIONS, app, log
from wbb.services.blacklist_enforcement import add_risk_keyword, remove_risk_keyword
from wbb.services.bot_permissions import has_permission
from wbb.services.chat_management import (
    ChatManagementError,
    apply_automation_settings,
    ensure_permission,
    execute_member_action,
    get_automation_settings,
    get_chat_overview,
    list_accessible_chats,
    list_recent_members,
    send_announcement,
    update_chat_permissions,
)
from wbb.services.giveaways import (
    cancel_and_refund_giveaway,
    create_and_publish_giveaway,
    finish_and_publish_giveaway,
    parse_giveaway_time,
)
from wbb.services.member_identity import display_name, non_mention_account_name
from wbb.services.point_settings import apply_point_rules
from wbb.utils.dbadmin import record_audit
from wbb.utils.dbgiveaway import get_giveaway, list_running_giveaways
from wbb.utils.dbpoints import (
    SOURCE_ADMIN,
    adjust_points,
    get_point_account,
    get_point_rules,
    list_point_accounts,
    list_point_transactions,
    set_points,
)
from wbb.utils.i18n import (
    member_action_label,
    point_source_label,
    telegram_permission_label,
)

__MODULE__ = "私聊管理"
__HELP__ = """私聊管理员菜单：
/manage [群组 ID] - 打开可管理的群列表，或直接打开指定群。
/cancel - 取消当前输入流程。

普通管理员只能看到自己拥有管理权限的群，超级管理员可管理全部群。"""

FLOW_TTL = timedelta(minutes=10)
PAGE_SIZE = 6
SECTION_ROLE_PERMISSIONS = {
    "announce": "chat.announcements",
    "members": "chat.members",
    "recent_members": "chat.members",
    "permissions": "chat.permissions",
    "autoreply": "automation.manage",
    "risk": "automation.manage",
    "identity": "automation.manage",
    "points": "points.manage",
    "point_history": "points.manage",
    "giveaways": "giveaways.manage",
}
MEMBER_ACTIONS = {
    "warn",
    "ban",
    "unban",
    "kick",
    "mute",
    "unmute",
    "promote",
    "demote",
}


@dataclass
class PendingFlow:
    action: str
    chat_id: int
    step: str
    data: dict[str, Any] = field(default_factory=dict)
    expires_at: datetime = field(
        default_factory=lambda: datetime.now(UTC) + FLOW_TTL
    )


_flows: dict[int, PendingFlow] = {}
_confirmations: dict[str, dict[str, Any]] = {}


def _button(text: str, data: str) -> InlineKeyboardButton:
    return InlineKeyboardButton(text, callback_data=data)


def _back(chat_id: int) -> list[InlineKeyboardButton]:
    return [_button("返回群菜单", f"mg:c:{chat_id}")]


def _require_bot_role(permission: str | None) -> None:
    if permission and not has_permission(permission, BOT_PERMISSIONS):
        raise ChatManagementError(
            "bot_role_permission_denied",
            "当前机器人未分配执行该操作的职责角色。",
            status=403,
        )


def _flow_role_permission(action: str) -> str | None:
    if action in MEMBER_ACTIONS:
        return "chat.members"
    if action.startswith("points_"):
        return "points.manage"
    if action == "announcement":
        return "chat.announcements"
    if action == "giveaway_create":
        return "giveaways.manage"
    if action == "autoreply" or action.startswith("risk_"):
        return "automation.manage"
    return None


def _direct_role_permission(action: str) -> str | None:
    if action == "permissions":
        return "chat.permissions"
    if action in {
        "toggle",
        "identity_monitor_toggle",
        "risk_rule_toggle",
        "risk_rule_action",
        "risk_rule_delete",
        "autoreply_clear",
    }:
        return "automation.manage"
    if action == "point_rule_toggle":
        return "points.manage"
    if action in {"gfinish", "gcancel"}:
        return "giveaways.manage"
    return None


async def _edit(query: CallbackQuery, text: str, keyboard: InlineKeyboardMarkup) -> None:
    await query.message.edit_text(
        text,
        parse_mode=ParseMode.HTML,
        reply_markup=keyboard,
        disable_web_page_preview=True,
    )


def _media_from_message(message: Message) -> tuple[str | None, str | None]:
    for media_type in ("photo", "animation", "video", "document"):
        media = getattr(message, media_type, None)
        if media:
            if media_type == "photo":
                media = media[-1] if isinstance(media, list) else media
            return media_type, media.file_id
    return None, None


async def _show_group_list(
    target: Message | CallbackQuery, actor_id: int, page: int = 1
) -> None:
    items, total = await list_accessible_chats(
        page=page, page_size=PAGE_SIZE, actor_id=actor_id
    )
    if not items:
        text = "你当前没有可管理的群。只有真实群管理员或 SUDOERS 能使用此菜单。"
        if isinstance(target, CallbackQuery):
            await target.message.edit_text(text)
        else:
            await target.reply_text(text)
        return
    rows = [
        [_button(str(item.get("title") or item["chat_id"])[:45], f"mg:c:{item['chat_id']}")]
        for item in items
    ]
    pages = max(1, (total + PAGE_SIZE - 1) // PAGE_SIZE)
    navigation = []
    if page > 1:
        navigation.append(_button("上一页", f"mg:p:{page - 1}"))
    if page < pages:
        navigation.append(_button("下一页", f"mg:p:{page + 1}"))
    if navigation:
        rows.append(navigation)
    text = f"<b>Telegram 群管理</b>\n选择群组（第 {page}/{pages} 页）："
    keyboard = InlineKeyboardMarkup(rows)
    if isinstance(target, CallbackQuery):
        await _edit(target, text, keyboard)
    else:
        await target.reply_text(text, parse_mode=ParseMode.HTML, reply_markup=keyboard)


async def _show_chat_menu(query: CallbackQuery, chat_id: int) -> None:
    overview = await get_chat_overview(chat_id, actor_id=query.from_user.id)
    privileges = (
        "、".join(
            telegram_permission_label(item)
            for item in overview.get("bot_privileges", [])
        )
        or "无管理权限"
    )
    text = (
        f"<b>{escape(overview.get('title') or str(chat_id))}</b>\n"
        f"群 ID：<code>{chat_id}</code>\n"
        f"成员数：{overview.get('member_count') or '-'}\n"
        f"机器人权限：{escape(privileges)}"
    )
    rows = [[_button("概览", f"mg:s:{chat_id}:overview")]]
    options = (
        ("chat.announcements", "公告", "announce"),
        ("chat.members", "成员管理", "members"),
        ("chat.permissions", "权限", "permissions"),
        ("automation.manage", "自动回复", "autoreply"),
        ("automation.manage", "风控", "risk"),
        ("automation.manage", "资料监控", "identity"),
        ("points.manage", "积分", "points"),
        ("giveaways.manage", "抽奖", "giveaways"),
    )
    available = [
        _button(label, f"mg:s:{chat_id}:{section}")
        for permission, label, section in options
        if has_permission(permission, BOT_PERMISSIONS)
    ]
    rows.extend(available[index : index + 2] for index in range(0, len(available), 2))
    rows.append([_button("返回群列表", "mg:p:1")])
    keyboard = InlineKeyboardMarkup(rows)
    await _edit(query, text, keyboard)


def _risk_member_action(rule: dict[str, Any]) -> str:
    actions = set(rule.get("actions", []))
    return next(
        (item for item in ("ban", "kick", "mute", "warn") if item in actions),
        "none",
    )


def _duration_label(seconds: int) -> str:
    if seconds % 86400 == 0:
        return f"{seconds // 86400} 天"
    if seconds % 3600 == 0:
        return f"{seconds // 3600} 小时"
    return f"{max(1, seconds // 60)} 分钟"


async def _show_risk_rule(
    query: CallbackQuery, chat_id: int, rule_id: str
) -> None:
    _require_bot_role("automation.manage")
    await get_chat_overview(chat_id, actor_id=query.from_user.id)
    settings = await get_automation_settings(chat_id)
    rule = next(
        (item for item in settings["risk_rules"] if item["rule_id"] == rule_id),
        None,
    )
    if not rule:
        await query.answer("风控规则不存在或已删除。", show_alert=True)
        return await _show_section(query, chat_id, "risk")

    action = _risk_member_action(rule)
    action_labels = {
        "none": "不处罚",
        "warn": "警告",
        "mute": "禁言",
        "kick": "踢出",
        "ban": "封禁",
    }
    triggers = []
    if rule["keywords"]:
        triggers.append(f"关键词 {len(rule['keywords'])} 个")
    if rule["match_images"]:
        triggers.append("图片")
    if rule["match_links"]:
        triggers.append("链接")
    rows = [
        [
            _button(
                f"规则状态：{'开' if rule['enabled'] else '关'}",
                f"mg:k:{chat_id}:{rule_id}:e",
            )
        ],
        [
            _button(
                f"图片：{'开' if rule['match_images'] else '关'}",
                f"mg:k:{chat_id}:{rule_id}:i",
            ),
            _button(
                f"链接：{'开' if rule['match_links'] else '关'}",
                f"mg:k:{chat_id}:{rule_id}:l",
            ),
        ],
        [
            _button(
                f"删除消息：{'开' if 'delete' in rule['actions'] else '关'}",
                f"mg:k:{chat_id}:{rule_id}:d",
            )
        ],
        [
            _button(
                f"{'✓ ' if action == 'none' else ''}不处罚",
                f"mg:k:{chat_id}:{rule_id}:n",
            ),
            _button(
                f"{'✓ ' if action == 'warn' else ''}警告",
                f"mg:k:{chat_id}:{rule_id}:w",
            ),
        ],
        [
            _button(
                f"{'✓ ' if action == 'mute' else ''}禁言",
                f"mg:k:{chat_id}:{rule_id}:m",
            ),
            _button(
                f"{'✓ ' if action == 'kick' else ''}踢出",
                f"mg:k:{chat_id}:{rule_id}:k",
            ),
        ],
        [
            _button(
                f"{'✓ ' if action == 'ban' else ''}封禁",
                f"mg:k:{chat_id}:{rule_id}:b",
            )
        ],
    ]
    if action in {"warn", "mute", "ban"}:
        duration_name = "禁言时长" if action == "mute" else "封禁时长"
        rows.append(
            [
                _button(
                    f"{duration_name}：{_duration_label(int(rule['duration_seconds']))}",
                    f"mg:q:{chat_id}:{rule_id}",
                )
            ]
        )
    rows.extend(
        [
            [
                _button("编辑规则", f"mg:e:{chat_id}:{rule_id}"),
                _button("删除规则", f"mg:k:{chat_id}:{rule_id}:x"),
            ],
            [_button("返回风控规则", f"mg:s:{chat_id}:risk")],
        ]
    )
    duration = (
        f"\n{'禁言' if action == 'mute' else '封禁'}时长："
        f"{_duration_label(int(rule['duration_seconds']))}"
        if action in {"warn", "mute", "ban"}
        else ""
    )
    text = (
        f"<b>{escape(rule['name'])}</b>\n"
        f"状态：{'已开启' if rule['enabled'] else '已关闭'}\n"
        f"触发条件：{escape('、'.join(triggers) or '未配置')}\n"
        f"消息处置：{'删除' if 'delete' in rule['actions'] else '保留'}\n"
        f"成员处置：{action_labels[action]}{duration}"
    )
    await _edit(query, text, InlineKeyboardMarkup(rows))


async def _show_section(query: CallbackQuery, chat_id: int, section: str) -> None:
    _require_bot_role(SECTION_ROLE_PERMISSIONS.get(section))
    await get_chat_overview(chat_id, actor_id=query.from_user.id)
    if section == "overview":
        return await _show_chat_menu(query, chat_id)
    if section == "announce":
        _flows[query.from_user.id] = PendingFlow("announcement", chat_id, "content")
        return await _edit(
            query,
            "<b>发送公告</b>\n请发送公告文字、图片、GIF、视频或文件。发送 /cancel 取消。",
            InlineKeyboardMarkup([_back(chat_id)]),
        )
    if section == "members":
        rows = [
            [_button("最近活跃成员", f"mg:s:{chat_id}:recent_members")],
            [_button("警告", f"mg:a:{chat_id}:warn"), _button("封禁", f"mg:a:{chat_id}:ban")],
            [_button("解封", f"mg:a:{chat_id}:unban"), _button("踢出", f"mg:a:{chat_id}:kick")],
            [_button("禁言", f"mg:a:{chat_id}:mute"), _button("解除禁言", f"mg:a:{chat_id}:unmute")],
            [_button("提升管理员", f"mg:a:{chat_id}:promote"), _button("降级管理员", f"mg:a:{chat_id}:demote")],
            [_back(chat_id)[0]],
        ]
        return await _edit(
            query,
            "<b>成员管理</b>\n可从最近活跃成员中选择，也可输入用户 ID 或 @username。",
            InlineKeyboardMarkup(rows),
        )
    if section == "recent_members":
        members = await list_recent_members(chat_id, limit=12)
        rows = []
        for member in members:
            user = member["user"]
            name = " ".join(
                item for item in (user.get("first_name"), user.get("last_name")) if item
            )
            label = name or (f"@{user['username']}" if user.get("username") else user["id"])
            rows.append([_button(label[:45], f"mg:u:{chat_id}:{user['id']}")])
        rows.append([_button("返回成员管理", f"mg:s:{chat_id}:members")])
        return await _edit(
            query,
            "<b>最近活跃成员</b>\n列表来自 Telegram 当前结果和机器人近期看到的消息。"
            if members
            else "<b>最近活跃成员</b>\n暂无记录。",
            InlineKeyboardMarkup(rows),
        )
    if section == "permissions":
        return await _edit(
            query,
            "<b>群权限</b>\n选择预设；执行前仍会检查你和机器人的限制成员权限。",
            InlineKeyboardMarkup(
                [
                    [_button("正常发言", f"mg:d:{chat_id}:permissions:normal")],
                    [_button("全员只读", f"mg:d:{chat_id}:permissions:readonly")],
                    [_back(chat_id)[0]],
                ]
            ),
        )
    if section == "autoreply":
        settings = await get_automation_settings(chat_id)
        return await _edit(
            query,
            f"<b>自动回复</b>\n面板规则数：{len(settings.get('auto_replies', []))}",
            InlineKeyboardMarkup(
                [
                    [_button("新增规则", f"mg:a:{chat_id}:autoreply")],
                    [_button("清空面板规则", f"mg:d:{chat_id}:autoreply:clear")],
                    [_back(chat_id)[0]],
                ]
            ),
        )
    if section == "risk":
        settings = await get_automation_settings(chat_id)
        rules = settings["risk_rules"]
        rows = [
            [
                _button(
                    f"{'开' if rule['enabled'] else '关'} · {rule['name']}"[:45],
                    f"mg:r:{chat_id}:{rule['rule_id']}",
                )
            ]
            for rule in rules
        ]
        rows.append([_button("新增风控规则", f"mg:n:{chat_id}")])
        for key, label in (
            ("captcha_enabled", "入群验证"),
            ("antiflood_enabled", "防刷屏"),
            ("chatbot_enabled", "聊天机器人"),
        ):
            state = bool(settings.get(key))
            rows.append(
                [_button(f"{label}：{'开' if state else '关'}", f"mg:d:{chat_id}:toggle:{key}")]
            )
        rows.append(_back(chat_id))
        text = (
            f"<b>内容风控规则</b>\n共 {len(rules)} 条，"
            f"已启用 {sum(1 for rule in rules if rule['enabled'])} 条。"
        )
        return await _edit(query, text, InlineKeyboardMarkup(rows))
    if section == "identity":
        settings = await get_automation_settings(chat_id)
        monitor = settings["identity_monitor"]
        return await _edit(
            query,
            (
                "<b>成员资料监控</b>\n"
                f"监控状态：{'已开启' if monitor['enabled'] else '已关闭'}\n"
                f"群内提醒：{'已开启' if monitor['notify_in_chat'] else '已关闭'}\n"
                "机器人会在成员发言、入群或产生其他可见事件时比对昵称和用户名。"
            ),
            InlineKeyboardMarkup(
                [
                    [
                        _button(
                            f"资料监控：{'开' if monitor['enabled'] else '关'}",
                            f"mg:d:{chat_id}:identity:enabled",
                        )
                    ],
                    [
                        _button(
                            f"群内提醒：{'开' if monitor['notify_in_chat'] else '关'}",
                            f"mg:d:{chat_id}:identity:notify_in_chat",
                        )
                    ],
                    [_back(chat_id)[0]],
                ]
            ),
        )
    if section == "points":
        rules = await get_point_rules(chat_id)
        accounts, _ = await list_point_accounts(chat_id=chat_id, page=1, page_size=5)
        lines = [
            f"<b>积分管理</b>\n状态：{'已开启' if rules['enabled'] else '已关闭'}",
            "自动规则：签到 {checkin} · 活跃 {activity} · 点赞 {upvote}".format(
                checkin="开" if rules["checkin_enabled"] else "关",
                activity="开" if rules["activity_enabled"] else "关",
                upvote="开" if rules["upvote_enabled"] else "关",
            ),
            f"底部签到按钮：{'开' if rules['checkin_button_enabled'] else '关'}",
        ]
        for index, account in enumerate(accounts, 1):
            label = non_mention_account_name(account)
            lines.append(f"{index}. {escape(str(label))}：{account['balance']}")
        return await _edit(
            query,
            "\n".join(lines),
            InlineKeyboardMarkup(
                [
                    [_button("查询成员", f"mg:a:{chat_id}:points_query"), _button("最近流水", f"mg:s:{chat_id}:point_history")],
                    [_button("增加", f"mg:a:{chat_id}:points_add"), _button("扣减", f"mg:a:{chat_id}:points_deduct")],
                    [_button("设置余额", f"mg:a:{chat_id}:points_set")],
                    [_button("开关积分系统", f"mg:d:{chat_id}:pointtoggle:enabled")],
                    [
                        _button("签到开关", f"mg:d:{chat_id}:pointtoggle:checkin_enabled"),
                        _button("活跃开关", f"mg:d:{chat_id}:pointtoggle:activity_enabled"),
                    ],
                    [
                        _button("点赞开关", f"mg:d:{chat_id}:pointtoggle:upvote_enabled"),
                        _button("底部签到按钮", f"mg:d:{chat_id}:pointtoggle:checkin_button_enabled"),
                    ],
                    [_back(chat_id)[0]],
                ]
            ),
        )
    if section == "point_history":
        transactions, _ = await list_point_transactions(
            chat_id=chat_id, page=1, page_size=10
        )
        lines = ["<b>最近积分流水</b>"]
        if not transactions:
            lines.append("暂无流水。")
        for item in transactions:
            delta = int(item["delta"])
            lines.append(
                f"<code>{item['user_id']}</code> · {'+' if delta > 0 else ''}{delta} · "
                f"{escape(point_source_label(item['source']))} · 余额 {item['balance_after']}"
            )
        return await _edit(
            query,
            "\n".join(lines),
            InlineKeyboardMarkup(
                [[_button("返回积分", f"mg:s:{chat_id}:points")], _back(chat_id)]
            ),
        )
    if section == "giveaways":
        giveaways = await list_running_giveaways(chat_id=chat_id, limit=5)
        lines = ["<b>抽奖管理</b>"]
        rows = [[_button("创建抽奖", f"mg:a:{chat_id}:giveaway_create")]]
        for giveaway in giveaways:
            lines.append(f"#{giveaway['giveaway_id']} · {escape(giveaway['title'])}")
            rows.append(
                [
                    _button("立即开奖", f"mg:d:{chat_id}:gfinish:{giveaway['giveaway_id']}"),
                    _button("取消", f"mg:d:{chat_id}:gcancel:{giveaway['giveaway_id']}"),
                ]
            )
        rows.append(_back(chat_id))
        return await _edit(query, "\n".join(lines), InlineKeyboardMarkup(rows))


async def _start_flow(query: CallbackQuery, chat_id: int, action: str) -> None:
    _require_bot_role(_flow_role_permission(action))
    if action == "points_query":
        await get_chat_overview(chat_id, actor_id=query.from_user.id)
    else:
        permission = (
            "can_promote_members"
            if action in {"promote", "demote"}
            else "can_restrict_members"
        )
        if action.startswith("points_") or action in {
            "autoreply",
            "giveaway_create",
            "risk_keyword_add",
            "risk_keyword_remove",
        }:
            permission = "can_change_info"
        await ensure_permission(chat_id, permission, actor_id=query.from_user.id)
    if action in {"warn", "ban", "unban", "kick", "mute", "unmute", "promote", "demote"}:
        _flows[query.from_user.id] = PendingFlow(action, chat_id, "target")
        text = f"<b>{member_action_label(action)}</b>\n请输入用户 ID 或 @username。"
    elif action.startswith("points_"):
        _flows[query.from_user.id] = PendingFlow(action, chat_id, "target")
        text = "请输入成员 ID 或 @username。"
    elif action == "autoreply":
        _flows[query.from_user.id] = PendingFlow(action, chat_id, "keyword")
        text = "请输入自动回复关键词。"
    elif action in {"risk_keyword_add", "risk_keyword_remove"}:
        _flows[query.from_user.id] = PendingFlow(action, chat_id, "keyword")
        text = "请输入要新增的风控关键词。" if action.endswith("add") else "请输入要移除的风控关键词。"
    elif action == "giveaway_create":
        _flows[query.from_user.id] = PendingFlow(action, chat_id, "title")
        text = "请输入抽奖标题。"
    else:
        return
    await _edit(
        query,
        text + "\n流程 10 分钟无操作自动失效；发送 /cancel 取消。",
        InlineKeyboardMarkup([_back(chat_id)]),
    )


async def _start_risk_rule_flow(
    query: CallbackQuery,
    chat_id: int,
    action: str,
    rule_id: str | None = None,
) -> None:
    _require_bot_role("automation.manage")
    await ensure_permission(chat_id, "can_change_info", actor_id=query.from_user.id)
    data: dict[str, Any] = {}
    if rule_id:
        settings = await get_automation_settings(chat_id)
        rule = next(
            (item for item in settings["risk_rules"] if item["rule_id"] == rule_id),
            None,
        )
        if not rule:
            return await query.answer("风控规则不存在或已删除。", show_alert=True)
        data.update(rule)
    else:
        data.update(
            {
                "rule_id": secrets.token_hex(8),
                "enabled": True,
                "keywords": [],
                "match_images": False,
                "match_links": False,
                "actions": ["delete", "warn"],
                "duration_seconds": 3600,
            }
        )

    if action == "risk_rule_duration":
        flow = PendingFlow(action, chat_id, "ban_duration", data)
        prompt = "请输入封禁分钟数，范围 1 到 525600。"
    else:
        flow = PendingFlow(action, chat_id, "risk_name", data)
        prompt = "请输入规则名称。"
    _flows[query.from_user.id] = flow
    await _edit(
        query,
        prompt + "\n流程 10 分钟无操作自动失效；发送 /cancel 取消。",
        InlineKeyboardMarkup(
            [[_button("返回规则", f"mg:r:{chat_id}:{data['rule_id']}")]]
            if rule_id
            else [[_button("返回风控规则", f"mg:s:{chat_id}:risk")]]
        ),
    )


async def _show_member_target(
    query: CallbackQuery,
    chat_id: int,
    user_id: int,
) -> None:
    await get_chat_overview(chat_id, actor_id=query.from_user.id)
    try:
        member = await app.get_chat_member(chat_id, user_id)
    except Exception:
        return await query.answer("该成员当前已不在群内。", show_alert=True)
    user = member.user
    name = " ".join(
        item for item in (user.first_name, user.last_name) if item
    ) or (f"@{user.username}" if user.username else str(user.id))
    rows = [
        [_button("警告", f"mg:v:{chat_id}:{user_id}:warn"), _button("封禁", f"mg:v:{chat_id}:{user_id}:ban")],
        [_button("踢出", f"mg:v:{chat_id}:{user_id}:kick"), _button("禁言", f"mg:v:{chat_id}:{user_id}:mute")],
        [_button("解除禁言", f"mg:v:{chat_id}:{user_id}:unmute"), _button("提升管理员", f"mg:v:{chat_id}:{user_id}:promote")],
        [_button("返回最近活跃", f"mg:s:{chat_id}:recent_members")],
    ]
    await _edit(
        query,
        f"<b>{escape(name)}</b>\n用户 ID：<code>{user.id}</code>",
        InlineKeyboardMarkup(rows),
    )


async def _start_flow_for_member(
    query: CallbackQuery,
    chat_id: int,
    user_id: int,
    action: str,
) -> None:
    _require_bot_role("chat.members")
    permission = "can_promote_members" if action in {"promote", "demote"} else "can_restrict_members"
    await ensure_permission(chat_id, permission, actor_id=query.from_user.id)
    flow = PendingFlow(action, chat_id, "reason", {"user_id": user_id})
    if action == "mute":
        flow.step = "duration"
        text = "请输入禁言分钟数；输入 0 表示永久禁言。"
    else:
        text = "请输入操作原因。"
    _flows[query.from_user.id] = flow
    await _edit(
        query,
        text + "\n流程 10 分钟无操作自动失效；发送 /cancel 取消。",
        InlineKeyboardMarkup([[_button("返回成员", f"mg:u:{chat_id}:{user_id}")]]),
    )


async def _resolve_user(raw: str) -> Any | None:
    try:
        user = await app.get_users(int(raw) if raw.lstrip("-").isdigit() else raw)
    except Exception:
        return None
    return user


async def _queue_confirmation(message: Message, flow: PendingFlow) -> None:
    token = secrets.token_urlsafe(8)
    _confirmations[token] = {
        "actor_id": message.from_user.id,
        "flow": flow,
        "expires_at": datetime.now(UTC) + FLOW_TTL,
    }
    _flows.pop(message.from_user.id, None)
    await message.reply_text(
        "请确认执行该操作。确认时会再次检查 Telegram 权限。",
        reply_markup=InlineKeyboardMarkup(
            [[_button("确认执行", f"mg:x:{token}"), _button("取消", f"mg:z:{token}")]]
        ),
    )


async def _handle_flow_input(message: Message, flow: PendingFlow) -> None:
    text = (message.text or message.caption or "").strip()
    flow.expires_at = datetime.now(UTC) + FLOW_TTL
    if flow.action in {"risk_rule_add", "risk_rule_edit", "risk_rule_duration"}:
        if flow.step == "risk_name":
            if not text:
                return await message.reply_text("规则名称不能为空。")
            flow.data["name"] = text[:60]
            flow.step = "risk_keywords"
            return await message.reply_text("请输入关键词，用逗号分隔；输入“无”表示不配置关键词。")
        if flow.step == "risk_keywords":
            pieces = [] if text in {"无", "none", "0"} else text.replace("，", ",").replace("\n", ",").split(",")
            flow.data["keywords"] = [item.strip()[:100] for item in pieces if item.strip()][:200]
            flow.step = "risk_triggers"
            return await message.reply_text("请输入其他触发条件：图片、链接、图片+链接或无。")
        if flow.step == "risk_triggers":
            normalized = text.lower().replace(" ", "").replace("、", "+").replace("，", "+")
            if normalized in {"无", "none", "0"}:
                match_images = match_links = False
            else:
                match_images = "图片" in normalized or "image" in normalized
                match_links = "链接" in normalized or "link" in normalized
                if not match_images and not match_links:
                    return await message.reply_text("请输入图片、链接、图片+链接或无。")
            flow.data["match_images"] = match_images
            flow.data["match_links"] = match_links
            flow.step = "risk_delete"
            return await message.reply_text("命中后是否删除消息？请输入是或否。")
        if flow.step == "risk_delete":
            normalized = text.lower()
            if normalized not in {"是", "否", "yes", "no", "1", "0"}:
                return await message.reply_text("请输入是或否。")
            flow.data["delete_message"] = normalized in {"是", "yes", "1"}
            flow.step = "risk_member_action"
            return await message.reply_text("请输入成员处置：不处罚、警告、禁言、踢出或封禁。")
        if flow.step == "risk_member_action":
            action_map = {
                "不处罚": "none",
                "无": "none",
                "none": "none",
                "警告": "warn",
                "warn": "warn",
                "禁言": "mute",
                "mute": "mute",
                "踢出": "kick",
                "kick": "kick",
                "封禁": "ban",
                "ban": "ban",
            }
            member_action = action_map.get(text.lower())
            if not member_action:
                return await message.reply_text("请输入不处罚、警告、禁言、踢出或封禁。")
            actions = ["delete"] if flow.data.pop("delete_message", False) else []
            if member_action != "none":
                actions.append(member_action)
            flow.data["actions"] = actions
            if member_action in {"warn", "mute", "ban"}:
                flow.step = "ban_duration"
                duration_name = "禁言" if member_action == "mute" else "封禁"
                return await message.reply_text(
                    f"请输入{duration_name}分钟数，范围 1 到 525600。"
                )
            return await _queue_confirmation(message, flow)
        if flow.step == "ban_duration":
            if not text.isdigit() or not 1 <= int(text) <= 525600:
                return await message.reply_text("请输入 1 到 525600 之间的整数分钟数。")
            flow.data["duration_seconds"] = int(text) * 60
            return await _queue_confirmation(message, flow)
    if flow.step == "target":
        target = await _resolve_user(text)
        if not target:
            return await message.reply_text("未找到用户，请重新输入用户 ID 或 @username。")
        user_id = target.id
        flow.data["user_id"] = user_id
        flow.data["username"] = target.username
        flow.data["first_name"] = target.first_name
        flow.data["display_name"] = display_name(target.first_name, target.last_name)
        if flow.action == "mute":
            flow.step = "duration"
            return await message.reply_text("请输入禁言分钟数；输入 0 表示永久禁言。")
        if flow.action == "points_query":
            await get_chat_overview(flow.chat_id, actor_id=message.from_user.id)
            account = await get_point_account(flow.chat_id, user_id)
            transactions, _ = await list_point_transactions(
                chat_id=flow.chat_id, user_id=user_id, page=1, page_size=5
            )
            _flows.pop(message.from_user.id, None)
            label = (
                flow.data.get("display_name")
                or (f"@{flow.data['username']}" if flow.data.get("username") else None)
                or str(user_id)
            )
            lines = [
                f"成员 {escape(str(label))}（<code>{user_id}</code>）当前积分：<b>{account['balance']}</b>",
                "最近流水：",
            ]
            if not transactions:
                lines.append("暂无流水。")
            for item in transactions:
                delta = int(item["delta"])
                lines.append(
                    f"{'+' if delta > 0 else ''}{delta} · "
                    f"{escape(point_source_label(item['source']))} · "
                    f"余额 {item['balance_after']}"
                )
            return await message.reply_text(
                "\n".join(lines),
                parse_mode=ParseMode.HTML,
                reply_markup=InlineKeyboardMarkup(
                    [[_button("返回积分菜单", f"mg:s:{flow.chat_id}:points")]]
                ),
            )
        if flow.action.startswith("points_"):
            flow.step = "amount"
            return await message.reply_text("请输入非负积分数值。")
        flow.step = "reason"
        return await message.reply_text("请输入操作原因。")
    if flow.step == "duration":
        if not text.isdigit():
            return await message.reply_text("请输入整数分钟数。")
        flow.data["duration_seconds"] = int(text) * 60 or None
        flow.step = "reason"
        return await message.reply_text("请输入禁言原因。")
    if flow.step == "amount":
        if not text.isdigit():
            return await message.reply_text("请输入非负整数积分。")
        flow.data["amount"] = int(text)
        flow.step = "reason"
        return await message.reply_text("请输入积分调整原因（必填）。")
    if flow.step == "reason":
        if not text:
            return await message.reply_text("原因不能为空。")
        flow.data["reason"] = text[:500]
        return await _queue_confirmation(message, flow)
    if flow.action == "announcement" and flow.step == "content":
        media_type, file_id = _media_from_message(message)
        flow.data.update({"text": text, "media_type": media_type, "file_id": file_id})
        return await _queue_confirmation(message, flow)
    if flow.action in {"risk_keyword_add", "risk_keyword_remove"}:
        if not text:
            return await message.reply_text("关键词不能为空。")
        flow.data["keyword"] = text[:100]
        return await _queue_confirmation(message, flow)
    if flow.action == "autoreply":
        if flow.step == "keyword":
            if not text:
                return await message.reply_text("关键词不能为空。")
            flow.data["keyword"] = text[:100]
            flow.step = "reply"
            return await message.reply_text("请发送回复文字或媒体。")
        media_type, file_id = _media_from_message(message)
        flow.data.update(
            {"type": media_type or "text", "file_id": file_id, "text": text[:4000]}
        )
        return await _queue_confirmation(message, flow)
    if flow.action == "giveaway_create":
        prompts = {
            "title": ("starts_at", "请输入报名开始时间，格式 YYYY-MM-DD HH:mm（北京时间）。"),
            "starts_at": ("ends_at", "请输入开奖时间，格式 YYYY-MM-DD HH:mm（北京时间）。"),
            "ends_at": ("prizes", "请输入奖项，如 一等奖:1:50, 二等奖:3:10。"),
            "prizes": ("minimum_points", "请输入最低积分，0 表示不限制。"),
            "minimum_points": ("entry_cost", "请输入报名消耗积分，0 表示免费。"),
            "entry_cost": ("participation_reward", "请输入参与奖励积分，0 表示无奖励。"),
        }
        if flow.step == "title":
            flow.data["title"] = text[:200]
        elif flow.step in {"starts_at", "ends_at"}:
            parsed = parse_giveaway_time(text)
            if not parsed:
                return await message.reply_text(
                    "时间格式不正确，请按 YYYY-MM-DD HH:mm 输入。"
                )
            if flow.step == "ends_at":
                if parsed <= datetime.now(UTC):
                    return await message.reply_text("开奖时间必须晚于当前时间。")
                if parsed <= flow.data["starts_at"]:
                    return await message.reply_text("开奖时间必须晚于报名开始时间。")
            flow.data[flow.step] = parsed
        elif flow.step == "prizes":
            prizes = []
            for piece in text.split(","):
                parts = [part.strip() for part in piece.split(":")]
                if len(parts) not in {2, 3} or not parts[1].isdigit() or (len(parts) == 3 and not parts[2].isdigit()):
                    return await message.reply_text("奖项格式不正确，请重新输入。")
                prizes.append(
                    {
                        "name": parts[0],
                        "count": int(parts[1]),
                        "points_reward": int(parts[2]) if len(parts) == 3 else 0,
                    }
                )
            flow.data["prizes"] = prizes
        elif flow.step in {"minimum_points", "entry_cost", "participation_reward"}:
            if not text.isdigit():
                return await message.reply_text("请输入非负整数。")
            flow.data[flow.step] = int(text)
        if flow.step == "participation_reward":
            return await _queue_confirmation(message, flow)
        next_step, prompt = prompts[flow.step]
        flow.step = next_step
        return await message.reply_text(prompt)


async def _execute_confirmation(query: CallbackQuery, token: str) -> None:
    confirmation = _confirmations.pop(token, None)
    if not confirmation or confirmation["actor_id"] != query.from_user.id:
        return await query.answer("确认已失效。", show_alert=True)
    if confirmation["expires_at"] < datetime.now(UTC):
        return await query.answer("确认已过期。", show_alert=True)
    flow: PendingFlow = confirmation["flow"]
    actor_id = query.from_user.id
    action = flow.action
    _require_bot_role(_flow_role_permission(action))
    data = flow.data
    if action.startswith("points_") or action.startswith("risk_rule_") or action in {
        "announcement",
        "giveaway_create",
        "risk_keyword_add",
        "risk_keyword_remove",
    }:
        await ensure_permission(flow.chat_id, "can_change_info", actor_id=actor_id)
    if action in {"warn", "ban", "unban", "kick", "mute", "unmute", "promote", "demote"}:
        result = await execute_member_action(
            flow.chat_id,
            user_id=int(data["user_id"]),
            action=action,
            reason=data["reason"],
            duration_seconds=data.get("duration_seconds"),
            privileges={
                "can_delete_messages": True,
                "can_restrict_members": True,
                "can_invite_users": True,
            }
            if action == "promote"
            else None,
            actor_id=actor_id,
        )
        summary = str(result)
    elif action == "announcement":
        result = await send_announcement(
            flow.chat_id,
            text=data["text"],
            media_type=data.get("media_type"),
            file_id=data.get("file_id"),
            actor_id=actor_id,
        )
        summary = f"消息编号={result['message_id']}"
    elif action.startswith("points_"):
        operation = action.removeprefix("points_")
        amount = int(data["amount"])
        key = f"private-points:{flow.chat_id}:{actor_id}:{token}"
        if operation == "set":
            account, _ = await set_points(
                chat_id=flow.chat_id,
                user_id=int(data["user_id"]),
                balance=amount,
                actor_id=actor_id,
                reason=data["reason"],
                idempotency_key=key,
                username=data.get("username"),
                first_name=data.get("first_name"),
                display_name=data.get("display_name"),
            )
        else:
            account, _ = await adjust_points(
                chat_id=flow.chat_id,
                user_id=int(data["user_id"]),
                delta=amount if operation == "add" else -amount,
                source=SOURCE_ADMIN,
                idempotency_key=key,
                actor_id=actor_id,
                reason=data["reason"],
                username=data.get("username"),
                first_name=data.get("first_name"),
                display_name=data.get("display_name"),
            )
        summary = f"余额={account['balance']}"
    elif action == "autoreply":
        settings = await get_automation_settings(flow.chat_id)
        rules = list(settings.get("auto_replies", []))
        rules = [item for item in rules if item.get("keyword") != data["keyword"]]
        rules.append(
            {
                "keyword": data["keyword"],
                "type": data["type"],
                "text": data["text"],
                "file_id": data["file_id"],
            }
        )
        await apply_automation_settings(
            flow.chat_id, {"auto_replies": rules}, actor_id=actor_id
        )
        summary = f"关键词={data['keyword']}"
    elif action in {"risk_keyword_add", "risk_keyword_remove"}:
        keyword = data["keyword"].strip()
        if action == "risk_keyword_add":
            await add_risk_keyword(flow.chat_id, keyword)
        else:
            await remove_risk_keyword(flow.chat_id, keyword)
        summary = f"风控关键词={keyword}"
    elif action in {"risk_rule_add", "risk_rule_edit", "risk_rule_duration"}:
        settings = await get_automation_settings(flow.chat_id)
        rules = [dict(item) for item in settings["risk_rules"]]
        rule_id = data["rule_id"]
        index = next(
            (i for i, item in enumerate(rules) if item["rule_id"] == rule_id),
            None,
        )
        if action == "risk_rule_add":
            if index is not None:
                return await query.answer("规则编号冲突，请重新新增。", show_alert=True)
            rules.append(dict(data))
        elif index is None:
            return await query.answer("风控规则不存在或已删除。", show_alert=True)
        elif action == "risk_rule_duration":
            rules[index]["duration_seconds"] = data["duration_seconds"]
        else:
            rules[index] = dict(data)
        await apply_automation_settings(
            flow.chat_id,
            {"risk_rules": rules},
            actor_id=actor_id,
        )
        summary = f"风控规则={rule_id}"
    elif action == "giveaway_create":
        giveaway = await create_and_publish_giveaway(
            chat_id=flow.chat_id,
            creator_id=actor_id,
            creator_name=query.from_user.first_name,
            title=data["title"],
            description="",
            prizes=data["prizes"],
            starts_at=data["starts_at"],
            ends_at=data["ends_at"],
            minimum_points=data["minimum_points"],
            entry_cost=data["entry_cost"],
            participation_reward=data["participation_reward"],
        )
        summary = f"抽奖编号={giveaway['giveaway_id']}"
    else:
        return await query.answer("未知操作。", show_alert=True)
    await record_audit(
        source="telegram_private",
        actor_id=actor_id,
        actor_name=query.from_user.first_name,
        action=action,
        chat_id=flow.chat_id,
        target_id=data.get("user_id"),
        summary=summary,
    )
    await query.answer("操作成功。", show_alert=True)
    if action.startswith("risk_rule_"):
        await _show_section(query, flow.chat_id, "risk")
    else:
        await _show_chat_menu(query, flow.chat_id)


async def _danger_action(query: CallbackQuery, parts: list[str]) -> None:
    chat_id = int(parts[2])
    category = parts[3]
    value = parts[4] if len(parts) > 4 else ""
    await get_chat_overview(chat_id, actor_id=query.from_user.id)
    if category == "permissions":
        payload = {"action": category, "chat_id": chat_id, "mode": value}
    elif category == "toggle":
        payload = {"action": category, "chat_id": chat_id, "key": value}
    elif category == "pointtoggle":
        payload = {"action": "point_rule_toggle", "chat_id": chat_id, "key": value}
    elif category == "identity":
        payload = {"action": "identity_monitor_toggle", "chat_id": chat_id, "key": value}
    elif category in {"gfinish", "gcancel"}:
        payload = {"action": category, "chat_id": chat_id, "giveaway_id": value}
    elif category == "autoreply":
        payload = {"action": "autoreply_clear", "chat_id": chat_id}
    else:
        return
    _require_bot_role(_direct_role_permission(payload["action"]))
    token = secrets.token_urlsafe(8)
    _confirmations[token] = {
        "actor_id": query.from_user.id,
        "direct": payload,
        "expires_at": datetime.now(UTC) + FLOW_TTL,
    }
    await _edit(
        query,
        "该操作会立即影响群组，请确认。",
        InlineKeyboardMarkup(
            [[_button("确认执行", f"mg:y:{token}"), _button("取消", f"mg:z:{token}")]]
        ),
    )


async def _start_risk_rule_direct(
    query: CallbackQuery, chat_id: int, rule_id: str, operation: str
) -> None:
    _require_bot_role("automation.manage")
    await ensure_permission(chat_id, "can_change_info", actor_id=query.from_user.id)
    settings = await get_automation_settings(chat_id)
    rule = next(
        (item for item in settings["risk_rules"] if item["rule_id"] == rule_id),
        None,
    )
    if not rule:
        return await query.answer("风控规则不存在或已删除。", show_alert=True)

    if operation in {"e", "i", "l", "d"}:
        payload = {
            "action": "risk_rule_toggle",
            "chat_id": chat_id,
            "rule_id": rule_id,
            "key": {
                "e": "enabled",
                "i": "match_images",
                "l": "match_links",
                "d": "delete",
            }[operation],
        }
    elif operation in {"n", "w", "m", "k", "b"}:
        payload = {
            "action": "risk_rule_action",
            "chat_id": chat_id,
            "rule_id": rule_id,
            "key": {
                "n": "none",
                "w": "warn",
                "m": "mute",
                "k": "kick",
                "b": "ban",
            }[operation],
        }
    elif operation == "x":
        payload = {
            "action": "risk_rule_delete",
            "chat_id": chat_id,
            "rule_id": rule_id,
        }
    else:
        return await query.answer("不支持的风控规则操作。", show_alert=True)

    token = secrets.token_urlsafe(8)
    _confirmations[token] = {
        "actor_id": query.from_user.id,
        "direct": payload,
        "expires_at": datetime.now(UTC) + FLOW_TTL,
    }
    await _edit(
        query,
        f"即将修改风控规则 <b>{escape(rule['name'])}</b>，请确认。",
        InlineKeyboardMarkup(
            [[_button("确认执行", f"mg:y:{token}"), _button("取消", f"mg:z:{token}")]]
        ),
    )


async def _execute_direct_confirmation(query: CallbackQuery, token: str) -> None:
    confirmation = _confirmations.pop(token, None)
    if not confirmation or confirmation.get("actor_id") != query.from_user.id:
        return await query.answer("确认已失效。", show_alert=True)
    if confirmation["expires_at"] < datetime.now(UTC):
        return await query.answer("确认已过期。", show_alert=True)
    payload = confirmation["direct"]
    chat_id = int(payload["chat_id"])
    action = payload["action"]
    _require_bot_role(_direct_role_permission(action))
    if action == "permissions":
        readonly = payload["mode"] == "readonly"
        await update_chat_permissions(
            chat_id,
            {
                "can_send_messages": not readonly,
                "can_send_media_messages": not readonly,
                "can_send_other_messages": not readonly,
                "can_send_polls": not readonly,
                "can_add_web_page_previews": not readonly,
                "can_invite_users": not readonly,
            },
            actor_id=query.from_user.id,
        )
    elif action == "toggle":
        key = payload["key"]
        if key == "points":
            rules = await get_point_rules(chat_id)
            await ensure_permission(chat_id, "can_change_info", actor_id=query.from_user.id)
            await apply_point_rules(chat_id, {"enabled": not rules["enabled"]})
        else:
            settings = await get_automation_settings(chat_id)
            await apply_automation_settings(
                chat_id, {key: not bool(settings.get(key))}, actor_id=query.from_user.id
            )
    elif action == "point_rule_toggle":
        key = payload["key"]
        allowed = {
            "enabled",
            "checkin_enabled",
            "checkin_button_enabled",
            "activity_enabled",
            "upvote_enabled",
        }
        if key not in allowed:
            return await query.answer("不支持的积分规则。", show_alert=True)
        await ensure_permission(chat_id, "can_change_info", actor_id=query.from_user.id)
        rules = await get_point_rules(chat_id)
        await apply_point_rules(chat_id, {key: not bool(rules[key])})
    elif action == "identity_monitor_toggle":
        key = payload["key"]
        if key not in {"enabled", "notify_in_chat"}:
            return await query.answer("不支持的资料监控开关。", show_alert=True)
        await ensure_permission(chat_id, "can_change_info", actor_id=query.from_user.id)
        settings = await get_automation_settings(chat_id)
        monitor = dict(settings["identity_monitor"])
        monitor[key] = not bool(monitor[key])
        await apply_automation_settings(
            chat_id,
            {"identity_monitor": monitor},
            actor_id=query.from_user.id,
        )
    elif action in {"risk_rule_toggle", "risk_rule_action", "risk_rule_delete"}:
        await ensure_permission(chat_id, "can_change_info", actor_id=query.from_user.id)
        settings = await get_automation_settings(chat_id)
        rules = [dict(item) for item in settings["risk_rules"]]
        rule_id = payload["rule_id"]
        index = next(
            (i for i, item in enumerate(rules) if item["rule_id"] == rule_id),
            None,
        )
        if index is None:
            return await query.answer("风控规则不存在或已删除。", show_alert=True)
        if action == "risk_rule_delete":
            rules.pop(index)
        elif action == "risk_rule_toggle":
            key = payload["key"]
            if key == "delete":
                actions = set(rules[index]["actions"])
                if "delete" in actions:
                    actions.remove("delete")
                else:
                    actions.add("delete")
                rules[index]["actions"] = sorted(actions)
            elif key in {"enabled", "match_images", "match_links"}:
                rules[index][key] = not bool(rules[index][key])
            else:
                return await query.answer("不支持的风控开关。", show_alert=True)
        else:
            key = payload["key"]
            if key not in {"none", "warn", "mute", "kick", "ban"}:
                return await query.answer("不支持的成员处置。", show_alert=True)
            actions = set(rules[index]["actions"])
            actions.difference_update({"warn", "mute", "kick", "ban"})
            if key != "none":
                actions.add(key)
            rules[index]["actions"] = sorted(actions)
        await apply_automation_settings(
            chat_id,
            {"risk_rules": rules},
            actor_id=query.from_user.id,
        )
    elif action == "autoreply_clear":
        await apply_automation_settings(
            chat_id, {"auto_replies": []}, actor_id=query.from_user.id
        )
    elif action == "gfinish":
        giveaway = await get_giveaway(payload["giveaway_id"])
        if not giveaway or int(giveaway["chat_id"]) != chat_id:
            return await query.answer("抽奖不存在。", show_alert=True)
        await ensure_permission(chat_id, "can_change_info", actor_id=query.from_user.id)
        await finish_and_publish_giveaway(payload["giveaway_id"])
    elif action == "gcancel":
        giveaway = await get_giveaway(payload["giveaway_id"])
        if not giveaway or int(giveaway["chat_id"]) != chat_id:
            return await query.answer("抽奖不存在。", show_alert=True)
        await ensure_permission(chat_id, "can_change_info", actor_id=query.from_user.id)
        await cancel_and_refund_giveaway(payload["giveaway_id"], chat_id=chat_id)
    await record_audit(
        source="telegram_private",
        actor_id=query.from_user.id,
        actor_name=query.from_user.first_name,
        action=action,
        chat_id=chat_id,
        summary=str(payload),
    )
    await query.answer("操作成功。", show_alert=True)
    if action == "identity_monitor_toggle":
        await _show_section(query, chat_id, "identity")
    elif action == "risk_rule_delete":
        await _show_section(query, chat_id, "risk")
    elif action.startswith("risk_rule_"):
        await _show_risk_rule(query, chat_id, payload["rule_id"])
    else:
        await _show_chat_menu(query, chat_id)


@app.on_message(filters.command("manage") & filters.private)
async def manage_command(_, message: Message):
    _flows.pop(message.from_user.id, None)
    if len(message.command) > 1:
        if not message.command[1].lstrip("-").isdigit():
            return await message.reply_text("用法：/manage [群组 ID]")
        chat_id = int(message.command[1])
        try:
            overview = await get_chat_overview(chat_id, actor_id=message.from_user.id)
        except ChatManagementError:
            return await message.reply_text("你无权管理该群，或机器人已无法访问该群。")
        return await message.reply_text(
            f"已选择 <b>{escape(overview['title'])}</b>。",
            parse_mode=ParseMode.HTML,
            reply_markup=InlineKeyboardMarkup([[_button("打开群菜单", f"mg:c:{chat_id}")]]),
        )
    await _show_group_list(message, message.from_user.id)


@app.on_message(filters.command("cancel") & filters.private)
async def cancel_manage_flow(_, message: Message):
    removed = _flows.pop(message.from_user.id, None)
    confirmation_tokens = [
        token
        for token, confirmation in _confirmations.items()
        if confirmation.get("actor_id") == message.from_user.id
    ]
    for token in confirmation_tokens:
        _confirmations.pop(token, None)
    canceled = bool(removed or confirmation_tokens)
    return await message.reply_text(
        "已取消当前操作。" if canceled else "当前没有进行中的操作。"
    )


@app.on_callback_query(filters.regex(r"^mg:"))
async def manage_callback(_, query: CallbackQuery):
    parts = query.data.split(":")
    try:
        if parts[1] == "p":
            return await _show_group_list(query, query.from_user.id, int(parts[2]))
        if parts[1] == "c":
            return await _show_chat_menu(query, int(parts[2]))
        if parts[1] == "s":
            return await _show_section(query, int(parts[2]), parts[3])
        if parts[1] == "r":
            return await _show_risk_rule(query, int(parts[2]), parts[3])
        if parts[1] == "k":
            return await _start_risk_rule_direct(
                query, int(parts[2]), parts[3], parts[4]
            )
        if parts[1] == "n":
            return await _start_risk_rule_flow(
                query, int(parts[2]), "risk_rule_add"
            )
        if parts[1] == "e":
            return await _start_risk_rule_flow(
                query, int(parts[2]), "risk_rule_edit", parts[3]
            )
        if parts[1] == "q":
            return await _start_risk_rule_flow(
                query, int(parts[2]), "risk_rule_duration", parts[3]
            )
        if parts[1] == "u":
            return await _show_member_target(query, int(parts[2]), int(parts[3]))
        if parts[1] == "v":
            return await _start_flow_for_member(
                query,
                int(parts[2]),
                int(parts[3]),
                parts[4],
            )
        if parts[1] == "a":
            return await _start_flow(query, int(parts[2]), parts[3])
        if parts[1] == "d":
            return await _danger_action(query, parts)
        if parts[1] == "x":
            return await _execute_confirmation(query, parts[2])
        if parts[1] == "y":
            return await _execute_direct_confirmation(query, parts[2])
        if parts[1] == "z":
            _confirmations.pop(parts[2], None)
            await query.answer("已取消。")
            return await _show_group_list(query, query.from_user.id)
    except ChatManagementError as exc:
        return await query.answer(str(exc), show_alert=True)
    except Exception as exc:
        log.error(f"私聊管理操作失败：{exc}")
        return await query.answer(
            "操作失败，请检查输入内容和机器人权限后重试。",
            show_alert=True,
        )


@app.on_message(filters.private & ~filters.command(["manage", "cancel"]), group=14)
async def manage_flow_input(_, message: Message):
    if not message.from_user:
        return
    flow = _flows.get(message.from_user.id)
    if not flow:
        return
    if flow.expires_at < datetime.now(UTC):
        _flows.pop(message.from_user.id, None)
        return await message.reply_text("管理流程已超时，请重新使用 /manage。")
    try:
        await _handle_flow_input(message, flow)
    except ChatManagementError as exc:
        _flows.pop(message.from_user.id, None)
        await message.reply_text(str(exc))