TelegramBotAdmin/wbb/modules/teacher_directory.py

from __future__ import annotations

from datetime import UTC, datetime, timedelta
from html import escape
from secrets import token_urlsafe
from typing import Any

from pyrogram import StopPropagation, filters
from pyrogram.enums import ChatMembersFilter, ChatMemberStatus, ParseMode
from pyrogram.types import (
    CallbackQuery,
    ChatMemberUpdated,
    InlineKeyboardButton,
    InlineKeyboardMarkup,
    KeyboardButton,
    Message,
    ReplyKeyboardMarkup,
)

from wbb import BOT_PROFILE_ID, BOT_USERNAME, app, log
from wbb.services.directory import (
    DirectoryServiceError,
    admin_change_teacher_state,
    admin_decide_teacher,
    admin_source_chat_ids,
    apply_as_teacher,
    change_own_teacher_state,
    clear_user_location,
    directory_for_user,
    format_distance,
    observe_directory_user,
    observe_group_member,
    profile_summary,
    public_teacher,
    require_platform_membership,
    require_teacher_admin,
    start_presence_sweeper,
    status_value,
    update_user_location,
    verify_platform_membership,
)
from wbb.services.message_cleanup import schedule_group_cleanup
from wbb.utils.dbdirectory import (
    APPLICATION_APPROVED,
    APPLICATION_PENDING,
    APPLICATION_REJECTED,
    APPLICATION_REVOKED,
    get_directory_location,
    get_directory_profile,
    list_directory_profiles,
    list_teacher_applications,
)

__MODULE__ = "技师与附近"
__HELP__ = """/directory - 打开附近技师菜单。
/nearby - 选择范围并查找附近技师。
/teacher_rank - 按距离查看全部上榜技师。
/teacher_apply - 申请成为技师。
/teacher_online - 技师上线。
/teacher_offline - 技师下线。
/teacher_list - 技师上榜。
/teacher_unlist - 技师下榜。"""

APPLICATION_LABELS = {
    "none": "未申请",
    APPLICATION_PENDING: "技师申请待审批",
    APPLICATION_APPROVED: "技师",
    APPLICATION_REJECTED: "技师申请已拒绝",
    APPLICATION_REVOKED: "技师资格已撤销",
}
TEACHER_ACTION_LABELS = {
    "list": "上榜",
    "unlist": "下榜",
    "online": "上线",
    "offline": "下线",
}
BUTTON_NEARBY = "附近技师"
BUTTON_RANK = "技师榜单"
BUTTON_APPLY = "申请成为技师"
BUTTON_APPROVAL = "技师审批"
BUTTON_STATUS = "我的状态"
BUTTON_LIST = "上榜"
BUTTON_UNLIST = "下榜"
BUTTON_ONLINE = "🟢 上线"
BUTTON_OFFLINE = "🔴 下线"
BUTTON_TELEGRAM_LOCATION = "Telegram 定位"
BUTTON_MANUAL_LOCATION = "手动选择位置"
BUTTON_CLEAR_LOCATION = "清除位置"

_admin_flows: dict[int, dict[str, Any]] = {}
_admin_confirmations: dict[str, dict[str, Any]] = {}
_location_flows: dict[int, datetime] = {}
_source_chat_contexts: dict[int, dict[str, Any]] = {}


def _display_name(profile: dict[str, Any]) -> str:
    return str(profile.get("display_name") or f"用户 {profile['user_id']}")


def _directory_keyboard(
    profile: dict[str, Any] | None,
    *,
    is_admin: bool,
) -> ReplyKeyboardMarkup:
    status = str((profile or {}).get("application_status") or "none")
    rows: list[list[KeyboardButton]] = [
        [KeyboardButton(BUTTON_TELEGRAM_LOCATION, request_location=True)],
        [KeyboardButton(BUTTON_MANUAL_LOCATION)],
        [KeyboardButton(BUTTON_NEARBY), KeyboardButton(BUTTON_RANK)],
    ]
    if status == APPLICATION_APPROVED:
        rows.extend(
            [
                [
                    KeyboardButton(
                        BUTTON_UNLIST if profile and profile.get("listed") else BUTTON_LIST
                    ),
                    KeyboardButton(
                        BUTTON_OFFLINE
                        if profile and profile.get("online")
                        else BUTTON_ONLINE
                    ),
                ],
                [KeyboardButton(BUTTON_STATUS)],
            ]
        )
    else:
        rows.append(
            [
                KeyboardButton(
                    "技师申请审核中"
                    if status == APPLICATION_PENDING
                    else BUTTON_APPLY
                )
            ]
        )
    rows.append([KeyboardButton(BUTTON_CLEAR_LOCATION)])
    if is_admin:
        rows.append([KeyboardButton(BUTTON_APPROVAL)])
    return ReplyKeyboardMarkup(
        rows,
        resize_keyboard=True,
        is_persistent=True,
        selective=True,
        placeholder="选择附近技师服务或输入消息",
    )


def _group_directory_keyboard(
    profile: dict[str, Any] | None,
    *,
    chat_id: int,
    is_admin: bool,
) -> InlineKeyboardMarkup:
    status = str((profile or {}).get("application_status") or "none")
    rows: list[list[InlineKeyboardButton]] = [
        [
            InlineKeyboardButton(
                "更新位置",
                url=f"https://t.me/{BOT_USERNAME}?start=directory_{chat_id}",
            ),
            InlineKeyboardButton(BUTTON_NEARBY, callback_data="dir:radius"),
            InlineKeyboardButton(BUTTON_RANK, callback_data="dir:list:all:1"),
        ]
    ]
    if status == APPLICATION_APPROVED:
        rows.append(
            [
                InlineKeyboardButton(
                    BUTTON_UNLIST if profile and profile.get("listed") else BUTTON_LIST,
                    callback_data=(
                        "dir:self:unlist"
                        if profile and profile.get("listed")
                        else "dir:self:list"
                    ),
                ),
                InlineKeyboardButton(
                    BUTTON_OFFLINE if profile and profile.get("online") else BUTTON_ONLINE,
                    callback_data=(
                        "dir:self:offline"
                        if profile and profile.get("online")
                        else "dir:self:online"
                    ),
                ),
                InlineKeyboardButton(BUTTON_STATUS, callback_data="dir:status"),
            ]
        )
    elif status == APPLICATION_PENDING:
        rows.append(
            [InlineKeyboardButton("技师申请审核中", callback_data="dir:status")]
        )
    else:
        rows.append(
            [InlineKeyboardButton(BUTTON_APPLY, callback_data="dir:apply")]
        )
    if is_admin:
        rows.append(
            [InlineKeyboardButton(BUTTON_APPROVAL, callback_data="dir:admin:pending:1")]
        )
    return InlineKeyboardMarkup(rows)


def _presence_label(online: bool) -> str:
    return "🟢 在线" if online else "🔴 离线"


def _menu_text(summary: dict[str, Any]) -> str:
    lines = [
        "<b>附近技师</b>",
        f"位置：{'已更新' if summary['has_location'] else '尚未更新'}",
    ]
    status = summary["application_status"]
    if status == APPLICATION_APPROVED:
        lines.extend(
            [
                "身份：技师",
                f"榜单：{'已上榜' if summary['listed'] else '未上榜'}",
                f"状态：{_presence_label(summary['online'])}",
            ]
        )
    elif status == APPLICATION_PENDING:
        lines.append("技师申请：待审批")
    elif status == APPLICATION_REJECTED:
        lines.append("技师申请：未通过")
    elif status == APPLICATION_REVOKED:
        lines.append("技师资格：已撤销")
    return "\n".join(lines)


async def send_directory_menu(
    message: Message,
    *,
    source_chat_id: int | None = None,
) -> None:
    if not message.from_user:
        return
    if source_chat_id is not None:
        try:
            member = await app.get_chat_member(source_chat_id, message.from_user.id)
            if getattr(member, "status", None) not in {
                ChatMemberStatus.OWNER,
                ChatMemberStatus.ADMINISTRATOR,
                ChatMemberStatus.MEMBER,
                ChatMemberStatus.RESTRICTED,
            }:
                return await message.reply_text("你已不是该群成员，无法打开菜单。")
            source_chat_title = ""
            try:
                source_chat = await app.get_chat(source_chat_id)
                source_chat_title = str(getattr(source_chat, "title", "") or "")
            except Exception:
                pass
            await observe_group_member(
                chat_id=source_chat_id,
                chat_title=source_chat_title,
                user=message.from_user,
                status=status_value(member.status),
                active=True,
                verified=True,
            )
            _source_chat_contexts[int(message.from_user.id)] = {
                "chat_id": int(source_chat_id),
                "chat_title": source_chat_title,
                "expires_at": datetime.now(UTC) + timedelta(minutes=30),
            }
        except Exception:
            return await message.reply_text("暂时无法核验来源群身份，请稍后重试。")
    try:
        await require_platform_membership(int(message.from_user.id))
    except DirectoryServiceError as exc:
        await message.reply_text(str(exc))
        return
    profile = await observe_directory_user(message.from_user)
    location = await get_directory_location(int(message.from_user.id))
    admin_result = await verify_platform_membership(
        user_id=int(message.from_user.id),
        require_admin=True,
    )
    summary = profile_summary(profile, location)
    await message.reply_text(
        _menu_text(summary),
        parse_mode=ParseMode.HTML,
        reply_markup=_directory_keyboard(
            {**profile, "online": summary["online"]},
            is_admin=bool(admin_result.get("allowed")),
        ),
    )


async def _verify_group_member(message: Message) -> bool:
    if not message.from_user:
        return False
    try:
        member = await app.get_chat_member(message.chat.id, message.from_user.id)
    except Exception:
        await message.reply_text("暂时无法验证你的群成员身份，请稍后重试。")
        return False
    active = member.status in {
        ChatMemberStatus.OWNER,
        ChatMemberStatus.ADMINISTRATOR,
        ChatMemberStatus.MEMBER,
        ChatMemberStatus.RESTRICTED,
    }
    await observe_group_member(
        chat_id=message.chat.id,
        chat_title=message.chat.title or "",
        user=message.from_user,
        status=status_value(member.status),
        active=active,
        verified=True,
    )
    if not active:
        await message.reply_text("该功能仅对当前群成员开放。")
    return active


@app.on_message(filters.command("directory"))
async def directory_command(_, message: Message):
    if message.chat.type.value == "private":
        return await send_directory_menu(message)
    if not await _verify_group_member(message):
        return
    profile = await observe_directory_user(message.from_user)
    summary = profile_summary(
        profile,
        await get_directory_location(message.from_user.id),
    )
    member = await app.get_chat_member(message.chat.id, message.from_user.id)
    response = await message.reply_text(
        _menu_text(summary),
        parse_mode=ParseMode.HTML,
        reply_markup=_group_directory_keyboard(
            {**profile, "online": summary["online"]},
            chat_id=message.chat.id,
            is_admin=member.status in {
                ChatMemberStatus.OWNER,
                ChatMemberStatus.ADMINISTRATOR,
            },
        ),
    )
    schedule_group_cleanup(message, response)


@app.on_message(filters.command("nearby"))
async def nearby_command(_, message: Message):
    if message.chat.type.value != "private" and not await _verify_group_member(message):
        return
    await _show_radius_picker(message)


@app.on_message(filters.command("teacher_rank"))
async def teacher_rank_command(_, message: Message):
    if message.chat.type.value != "private" and not await _verify_group_member(message):
        return
    await _show_teacher_results(message, radius_km=None, page=1)


@app.on_message(filters.command("teacher_apply"))
async def teacher_apply_command(_, message: Message):
    if message.chat.type.value != "private" and not await _verify_group_member(message):
        return
    await _apply_teacher(message)


async def _self_state_command(
    target: Message | CallbackQuery,
    action: str,
) -> None:
    if not target.from_user:
        return
    message = target.message if isinstance(target, CallbackQuery) else target
    is_group = int(message.chat.id) < 0
    if is_group:
        try:
            member = await app.get_chat_member(message.chat.id, target.from_user.id)
        except Exception:
            if isinstance(target, CallbackQuery):
                return await target.answer("暂时无法核验群成员身份。", show_alert=True)
            return await target.reply_text("暂时无法核验群成员身份。")
        if member.status not in {
            ChatMemberStatus.OWNER,
            ChatMemberStatus.ADMINISTRATOR,
            ChatMemberStatus.MEMBER,
            ChatMemberStatus.RESTRICTED,
        }:
            if isinstance(target, CallbackQuery):
                return await target.answer("该功能仅对当前群成员开放。", show_alert=True)
            return await target.reply_text("该功能仅对当前群成员开放。")
        await observe_group_member(
            chat_id=message.chat.id,
            chat_title=message.chat.title or "",
            user=target.from_user,
            status=status_value(member.status),
            active=True,
            verified=True,
        )
    try:
        profile, _ = await change_own_teacher_state(
            user=target.from_user,
            action=action,
            source="telegram_group" if is_group else "telegram_private",
            idempotency_key=(
                f"teacher-self:{message.chat.id}:{message.id}:"
                f"{target.from_user.id}:{action}"
                if is_group
                else None
            ),
            chat_id=message.chat.id if is_group else None,
        )
        response = await message.reply_text(
            f"已{TEACHER_ACTION_LABELS[action]}。"
            + (
                "本次上线状态将在 24 小时后自动失效。"
                if action == "online"
                else ""
            )
        )
        if isinstance(target, CallbackQuery):
            await target.answer("状态已更新。")
        if not is_group and isinstance(target, Message):
            await _refresh_keyboard(target, profile)
        else:
            schedule_group_cleanup(message, response)
    except DirectoryServiceError as exc:
        if isinstance(target, CallbackQuery):
            await target.answer(str(exc), show_alert=True)
        else:
            response = await target.reply_text(str(exc))
            schedule_group_cleanup(target, response)


@app.on_message(filters.command("teacher_online"))
async def teacher_online_command(_, message: Message):
    if message.chat.type.value != "private" and not await _verify_group_member(message):
        return
    await _self_state_command(message, "online")


@app.on_message(filters.command("teacher_offline"))
async def teacher_offline_command(_, message: Message):
    if message.chat.type.value != "private" and not await _verify_group_member(message):
        return
    await _self_state_command(message, "offline")


@app.on_message(filters.command("teacher_list"))
async def teacher_list_command(_, message: Message):
    if message.chat.type.value != "private" and not await _verify_group_member(message):
        return
    await _self_state_command(message, "list")


@app.on_message(filters.command("teacher_unlist"))
async def teacher_unlist_command(_, message: Message):
    if message.chat.type.value != "private" and not await _verify_group_member(message):
        return
    await _self_state_command(message, "unlist")


@app.on_message(filters.private & filters.location, group=12)
async def directory_location_message(_, message: Message):
    if not message.from_user or not message.location:
        return
    try:
        await update_user_location(
            user=message.from_user,
            longitude=message.location.longitude,
            latitude=message.location.latitude,
            source="telegram_location",
        )
    except DirectoryServiceError as exc:
        return await message.reply_text(str(exc))
    _location_flows.pop(message.from_user.id, None)
    await message.reply_text("Telegram 位置已更新，可立即查找附近技师。")
    await _refresh_keyboard(message, await get_directory_profile(message.from_user.id))


async def _start_manual_location(message: Message) -> None:
    if not message.from_user:
        return
    _location_flows[message.from_user.id] = datetime.now(UTC) + timedelta(minutes=10)
    await message.reply_text(
        "请直接发送 Telegram 位置并在地图上选择地点；如果 Telegram 无法定位，"
        "也可以输入“纬度,经度”，例如：39.9042,116.4074。\n"
        "发送 /cancel 可取消。"
    )


async def _refresh_keyboard(
    message: Message,
    profile: dict[str, Any] | None,
) -> None:
    if not message.from_user:
        return
    admin_result = await verify_platform_membership(
        user_id=int(message.from_user.id),
        require_admin=True,
    )
    summary = profile_summary(
        profile or {"user_id": message.from_user.id},
        await get_directory_location(message.from_user.id),
    )
    await message.reply_text(
        "菜单已更新。",
        reply_markup=_directory_keyboard(
            {**(profile or {}), "online": summary["online"]},
            is_admin=bool(admin_result.get("allowed")),
        ),
    )


def _admin_mention(user: Any) -> str:
    name = " ".join(
        value
        for value in (
            str(getattr(user, "first_name", "") or "").strip(),
            str(getattr(user, "last_name", "") or "").strip(),
        )
        if value
    ) or str(getattr(user, "username", "") or "管理员")
    return f'<a href="tg://user?id={int(user.id)}">{escape(name)}</a>'


async def _notify_application_admins(profile: dict[str, Any]) -> None:
    chat_id = profile.get("application_source_chat_id")
    if not chat_id:
        return
    admins: list[Any] = []
    try:
        async for member in app.get_chat_members(
            int(chat_id),
            filter=ChatMembersFilter.ADMINISTRATORS,
        ):
            user = getattr(member, "user", None)
            if user and not getattr(user, "is_bot", False):
                admins.append(user)
    except Exception as exc:
        log.error(f"获取技师申请来源群管理员失败：{exc}")
    mentions = " ".join(_admin_mention(user) for user in admins[:20])
    applicant = escape(_display_name(profile))
    text = (
        f"{mentions}\n" if mentions else ""
    ) + (
        f"有新的技师申请待审批：<b>{applicant}</b>\n"
        f"用户 ID：<code>{profile['user_id']}</code>"
    )
    markup = InlineKeyboardMarkup(
        [
            [
                InlineKeyboardButton(
                    "查看并审批",
                    callback_data=f"dir:admin:detail:{profile['user_id']}",
                )
            ]
        ]
    )
    try:
        await app.send_message(
            int(chat_id),
            text,
            parse_mode=ParseMode.HTML,
            reply_markup=markup,
        )
    except Exception as exc:
        log.error(f"发送技师申请群提醒失败：{exc}")
    private_text = (
        f"群“{escape(profile.get('application_source_chat_title') or str(chat_id))}”"
        f"有新的技师申请：<b>{applicant}</b>。"
    )
    for admin in admins:
        if int(admin.id) == int(profile["user_id"]):
            continue
        try:
            await app.send_message(
                int(admin.id),
                private_text,
                parse_mode=ParseMode.HTML,
                reply_markup=markup,
            )
        except Exception:
            pass


async def _apply_teacher(target: Message | CallbackQuery) -> None:
    if not target.from_user:
        return
    message = target.message if isinstance(target, CallbackQuery) else target
    is_group = int(message.chat.id) < 0
    source_context = None
    if not is_group:
        source_context = _source_chat_contexts.get(int(target.from_user.id))
        if (
            source_context
            and source_context["expires_at"] < datetime.now(UTC)
        ):
            _source_chat_contexts.pop(int(target.from_user.id), None)
            source_context = None
    if is_group:
        try:
            member = await app.get_chat_member(message.chat.id, target.from_user.id)
        except Exception:
            if isinstance(target, CallbackQuery):
                return await target.answer("暂时无法核验群成员身份。", show_alert=True)
            return await target.reply_text("暂时无法核验群成员身份。")
        if member.status not in {
            ChatMemberStatus.OWNER,
            ChatMemberStatus.ADMINISTRATOR,
            ChatMemberStatus.MEMBER,
            ChatMemberStatus.RESTRICTED,
        }:
            if isinstance(target, CallbackQuery):
                return await target.answer("该功能仅对当前群成员开放。", show_alert=True)
            return await target.reply_text("该功能仅对当前群成员开放。")
        await observe_group_member(
            chat_id=message.chat.id,
            chat_title=message.chat.title or "",
            user=target.from_user,
            status=status_value(member.status),
            active=True,
            verified=True,
        )
    try:
        profile, created = await apply_as_teacher(
            user=target.from_user,
            chat_id=(
                int(message.chat.id)
                if is_group
                else int(source_context["chat_id"])
                if source_context
                else None
            ),
            chat_title=(
                str(message.chat.title or "")
                if is_group
                else str(source_context.get("chat_title") or "")
                if source_context
                else ""
            ),
            source="telegram_group" if is_group else "telegram_private",
        )
    except DirectoryServiceError as exc:
        if isinstance(target, CallbackQuery):
            return await target.answer(str(exc), show_alert=True)
        return await target.reply_text(str(exc))
    response_text = (
        "技师申请已提交，来源群管理员已收到审批提醒。"
        if created
        else "技师申请正在等待管理员审批。"
    )
    response = await message.reply_text(response_text)
    if isinstance(target, CallbackQuery):
        await target.answer("申请已提交。" if created else "申请正在审批中。")
    if created:
        _source_chat_contexts.pop(int(target.from_user.id), None)
        await _notify_application_admins(profile)
    if is_group:
        schedule_group_cleanup(message, response)
    elif isinstance(target, Message):
        await _refresh_keyboard(target, profile)


async def _show_status(target: Message | CallbackQuery) -> None:
    if not target.from_user:
        return
    message = target.message if isinstance(target, CallbackQuery) else target
    profile = await observe_directory_user(target.from_user)
    summary = profile_summary(
        profile,
        await get_directory_location(target.from_user.id),
    )
    if summary["application_status"] != APPLICATION_APPROVED:
        status_text = {
            APPLICATION_PENDING: "技师申请正在等待管理员审批。",
            APPLICATION_REJECTED: "技师申请未通过。",
            APPLICATION_REVOKED: "技师资格已被撤销。",
        }.get(summary["application_status"], "你尚未申请成为技师。")
        response = await message.reply_text(status_text)
        if isinstance(target, CallbackQuery):
            await target.answer()
        schedule_group_cleanup(message, response)
        return
    lines = [
        "<b>我的技师状态</b>",
        f"位置：{'已更新' if summary['has_location'] else '尚未更新'}",
        f"榜单：{'已上榜' if summary['listed'] else '未上榜'}",
        f"上下线：{_presence_label(summary['online'])}",
        (
            "自动下线："
            + escape(
                summary["online_until"].astimezone(UTC).strftime(
                    "%Y-%m-%d %H:%M UTC"
                )
            )
            if summary["online"] and summary["online_until"]
            else ""
        ),
    ]
    response = await message.reply_text(
        "\n".join(line for line in lines if line),
        parse_mode=ParseMode.HTML,
    )
    if isinstance(target, CallbackQuery):
        await target.answer()
    schedule_group_cleanup(message, response)


async def _show_radius_picker(target: Message | CallbackQuery) -> None:
    markup = InlineKeyboardMarkup(
        [
            [
                InlineKeyboardButton(
                    f"{radius} 公里",
                    callback_data=f"dir:list:{radius}:1",
                )
                for radius in (5, 10)
            ],
            [
                InlineKeyboardButton(
                    f"{radius} 公里",
                    callback_data=f"dir:list:{radius}:1",
                )
                for radius in (20, 50)
            ],
        ]
    )
    if isinstance(target, CallbackQuery):
        await target.message.edit_text("请选择附近技师的搜索范围：", reply_markup=markup)
    else:
        response = await target.reply_text(
            "请选择附近技师的搜索范围：",
            reply_markup=markup,
        )
        schedule_group_cleanup(target, response)


def _teacher_result_markup(
    teachers: list[dict[str, Any]],
    *,
    radius_key: str,
    page: int,
    total: int,
) -> InlineKeyboardMarkup:
    rows: list[list[InlineKeyboardButton]] = []
    teacher_buttons: list[InlineKeyboardButton] = []
    for item in teachers:
        username = str(item.get("username") or "")
        status = "🟢" if item["online"] else "🔴"
        teacher_buttons.append(
            InlineKeyboardButton(
                f"{status} {item['display_name']}"[:18],
                url=f"https://t.me/{username}",
            )
        )
    rows.extend(
        teacher_buttons[index : index + 3]
        for index in range(0, len(teacher_buttons), 3)
    )
    navigation: list[InlineKeyboardButton] = []
    if page > 1:
        navigation.append(
            InlineKeyboardButton(
                "上一页",
                callback_data=f"dir:list:{radius_key}:{page - 1}",
            )
        )
    if page * 10 < total:
        navigation.append(
            InlineKeyboardButton(
                "下一页",
                callback_data=f"dir:list:{radius_key}:{page + 1}",
            )
        )
    if navigation:
        rows.append(navigation)
    rows.append([InlineKeyboardButton("切换附近范围", callback_data="dir:radius")])
    return InlineKeyboardMarkup(rows)


async def _show_teacher_results(
    target: Message | CallbackQuery,
    *,
    radius_km: int | None,
    page: int,
) -> None:
    user = target.from_user
    try:
        raw_items, total = await directory_for_user(
            user=user,
            radius_km=radius_km,
            page=page,
            page_size=10,
        )
    except DirectoryServiceError as exc:
        if isinstance(target, CallbackQuery):
            return await target.answer(str(exc), show_alert=True)
        return await target.reply_text(str(exc))
    teachers = [public_teacher(item) for item in raw_items]
    title = (
        f"<b>{radius_km} 公里内的技师</b>"
        if radius_km is not None
        else "<b>技师榜单</b>"
    )
    lines = [title, f"共 {total} 位已上榜技师，按距离排序。"]
    for index, item in enumerate(teachers, start=(page - 1) * 10 + 1):
        lines.append(
            f"\n{index}. <b>{escape(item['display_name'])}</b> "
            f"(@{escape(item['username'])})\n"
            f"状态：{_presence_label(item['online'])} | "
            f"距离：{format_distance(item['distance_meters'])}"
        )
    if not teachers:
        lines.append("\n当前范围内没有已上榜技师。")
    markup = _teacher_result_markup(
        teachers,
        radius_key=str(radius_km) if radius_km is not None else "all",
        page=page,
        total=total,
    )
    if isinstance(target, CallbackQuery):
        await target.message.edit_text(
            "\n".join(lines),
            parse_mode=ParseMode.HTML,
            disable_web_page_preview=True,
            reply_markup=markup,
        )
        await target.answer()
    else:
        response = await target.reply_text(
            "\n".join(lines),
            parse_mode=ParseMode.HTML,
            disable_web_page_preview=True,
            reply_markup=markup,
        )
        schedule_group_cleanup(target, response)


async def _show_admin_applications(
    target: Message | CallbackQuery,
    *,
    page: int = 1,
) -> None:
    try:
        source_chat_ids = await admin_source_chat_ids(target.from_user)
    except DirectoryServiceError as exc:
        if isinstance(target, CallbackQuery):
            return await target.answer(str(exc), show_alert=True)
        return await target.reply_text(str(exc))
    items, total = await list_teacher_applications(
        status=APPLICATION_PENDING,
        source_bot_id=str(BOT_PROFILE_ID)
        if source_chat_ids is not None
        else "",
        source_chat_ids=source_chat_ids,
        page=page,
        page_size=5,
    )
    lines = ["<b>技师申请审批</b>", f"待审批：{total}"]
    rows: list[list[InlineKeyboardButton]] = []
    for item in items:
        lines.append(
            f"\n{escape(_display_name(item))} (@{escape(item.get('username') or '')})"
            f"\n用户 ID：<code>{item['user_id']}</code>"
        )
        rows.append(
            [
                InlineKeyboardButton(
                    f"查看 {_display_name(item)[:18]}",
                    callback_data=f"dir:admin:detail:{item['user_id']}",
                )
            ]
        )
    if page > 1:
        rows.append(
            [
                InlineKeyboardButton(
                    "上一页",
                    callback_data=f"dir:admin:pending:{page - 1}",
                )
            ]
        )
    if page * 5 < total:
        rows.append(
            [
                InlineKeyboardButton(
                    "下一页",
                    callback_data=f"dir:admin:pending:{page + 1}",
                )
            ]
        )
    rows.append(
        [InlineKeyboardButton("已批准技师", callback_data="dir:admin:teachers:1")]
    )
    markup = InlineKeyboardMarkup(rows)
    if isinstance(target, CallbackQuery):
        await target.message.edit_text(
            "\n".join(lines),
            parse_mode=ParseMode.HTML,
            reply_markup=markup,
        )
        await target.answer()
    else:
        await target.reply_text(
            "\n".join(lines),
            parse_mode=ParseMode.HTML,
            reply_markup=markup,
        )


async def _show_admin_teachers(target: CallbackQuery, page: int) -> None:
    try:
        source_chat_ids = await admin_source_chat_ids(target.from_user)
    except DirectoryServiceError as exc:
        return await target.answer(str(exc), show_alert=True)
    items, total = await list_directory_profiles(
        application_status=APPLICATION_APPROVED,
        source_bot_id=str(BOT_PROFILE_ID)
        if source_chat_ids is not None
        else "",
        source_chat_ids=source_chat_ids,
        page=page,
        page_size=8,
    )
    rows = [
        [
            InlineKeyboardButton(
                _display_name(item)[:28],
                callback_data=f"dir:admin:detail:{item['user_id']}",
            )
        ]
        for item in items
    ]
    navigation: list[InlineKeyboardButton] = []
    if page > 1:
        navigation.append(
            InlineKeyboardButton(
                "上一页",
                callback_data=f"dir:admin:teachers:{page - 1}",
            )
        )
    if page * 8 < total:
        navigation.append(
            InlineKeyboardButton(
                "下一页",
                callback_data=f"dir:admin:teachers:{page + 1}",
            )
        )
    if navigation:
        rows.append(navigation)
    rows.append([InlineKeyboardButton("待审批", callback_data="dir:admin:pending:1")])
    await target.message.edit_text(
        f"<b>已批准技师</b>\n共 {total} 人",
        parse_mode=ParseMode.HTML,
        reply_markup=InlineKeyboardMarkup(rows),
    )
    await target.answer()


async def _show_admin_teacher_detail(target: CallbackQuery, user_id: int) -> None:
    try:
        await require_teacher_admin(target.from_user, user_id)
    except DirectoryServiceError as exc:
        return await target.answer(str(exc), show_alert=True)
    profile = await get_directory_profile(user_id)
    if not profile:
        return await target.answer("未找到该成员资料。", show_alert=True)
    summary = profile_summary(profile, await get_directory_location(user_id))
    lines = [
        f"<b>{escape(summary['display_name'])}</b>",
        f"用户名：@{escape(summary.get('username') or '未设置')}",
        f"用户 ID：<code>{user_id}</code>",
        f"申请：{escape(APPLICATION_LABELS.get(summary['application_status'], '未申请'))}",
        f"位置：{'已更新' if summary['has_location'] else '尚未更新'}",
        f"榜单：{'已上榜' if summary['listed'] else '未上榜'}",
        f"状态：{_presence_label(summary['online'])}",
    ]
    rows: list[list[InlineKeyboardButton]] = []
    if summary["application_status"] == APPLICATION_PENDING:
        rows.append(
            [
                InlineKeyboardButton(
                    "批准",
                    callback_data=f"dir:admin:approveask:{user_id}",
                ),
                InlineKeyboardButton(
                    "拒绝",
                    callback_data=f"dir:admin:flow:reject:{user_id}",
                ),
            ]
        )
    if summary["application_status"] == APPLICATION_APPROVED:
        rows.extend(
            [
                [
                    InlineKeyboardButton(
                        "强制下榜" if summary["listed"] else "强制上榜",
                        callback_data=(
                            f"dir:admin:flow:"
                            f"{'unlist' if summary['listed'] else 'list'}:{user_id}"
                        ),
                    ),
                    InlineKeyboardButton(
                        "强制下线" if summary["online"] else "强制上线",
                        callback_data=(
                            f"dir:admin:flow:"
                            f"{'offline' if summary['online'] else 'online'}:{user_id}"
                        ),
                    ),
                ],
                [
                    InlineKeyboardButton(
                        "撤销技师资格",
                        callback_data=f"dir:admin:flow:revoke:{user_id}",
                    )
                ],
            ]
        )
    rows.append([InlineKeyboardButton("返回审批", callback_data="dir:admin:pending:1")])
    await target.message.edit_text(
        "\n".join(lines),
        parse_mode=ParseMode.HTML,
        reply_markup=InlineKeyboardMarkup(rows),
    )
    try:
        await target.answer()
    except Exception:
        pass


async def _start_admin_reason_flow(
    target: CallbackQuery,
    *,
    action: str,
    user_id: int,
) -> None:
    try:
        await require_teacher_admin(target.from_user, user_id)
    except DirectoryServiceError as exc:
        return await target.answer(str(exc), show_alert=True)
    _admin_flows[int(target.from_user.id)] = {
        "action": action,
        "user_id": int(user_id),
        "expires_at": datetime.now(UTC) + timedelta(minutes=10),
    }
    await target.message.reply_text(
        f"请输入“{TEACHER_ACTION_LABELS.get(action, action)}”的原因，"
        "发送 /cancel 可取消。"
    )
    await target.answer()


async def _confirm_admin_action(target: CallbackQuery, token: str) -> None:
    confirmation = _admin_confirmations.get(token)
    if not confirmation or confirmation["actor_id"] != target.from_user.id:
        return await target.answer("确认已失效，请重新操作。", show_alert=True)
    if confirmation["expires_at"] < datetime.now(UTC):
        _admin_confirmations.pop(token, None)
        return await target.answer("确认已超时，请重新操作。", show_alert=True)
    action = confirmation["action"]
    user_id = int(confirmation["user_id"])
    reason = str(confirmation.get("reason") or "")
    try:
        if action in {"reject", "revoke"}:
            profile = await admin_decide_teacher(
                actor=target.from_user,
                user_id=user_id,
                action=action,
                reason=reason,
            )
        else:
            profile = await admin_change_teacher_state(
                actor=target.from_user,
                user_id=user_id,
                action=action,
                reason=reason,
            )
        with_reason = f"\n原因：{reason}" if reason else ""
        with_suppressed_notification = (
            "技师申请已被拒绝" if action == "reject" else
            "技师资格已被撤销" if action == "revoke" else
            f"技师状态已被管理员修改为“{TEACHER_ACTION_LABELS[action]}”"
        )
        try:
            await app.send_message(
                user_id,
                f"{with_suppressed_notification}。{with_reason}",
            )
        except Exception:
            pass
        _admin_confirmations.pop(token, None)
        await target.answer("操作成功。")
        await _show_admin_teacher_detail(target, int(profile["user_id"]))
    except DirectoryServiceError as exc:
        await target.answer(str(exc), show_alert=True)


@app.on_callback_query(filters.regex(r"^dir:"))
async def directory_callback(_, query: CallbackQuery):
    parts = str(query.data).split(":")
    try:
        if parts[1] == "apply":
            return await _apply_teacher(query)
        if parts[1] == "status":
            return await _show_status(query)
        if parts[1] == "self":
            return await _self_state_command(query, parts[2])
        if parts[1] == "radius":
            return await _show_radius_picker(query)
        if parts[1] == "list":
            radius = None if parts[2] == "all" else int(parts[2])
            return await _show_teacher_results(
                query,
                radius_km=radius,
                page=int(parts[3]),
            )
        if parts[1] == "clearask":
            return await query.message.edit_text(
                "清除位置后会自动下榜并下线，确认继续吗？",
                reply_markup=InlineKeyboardMarkup(
                    [
                        [
                            InlineKeyboardButton(
                                "确认清除",
                                callback_data="dir:clear",
                            ),
                            InlineKeyboardButton("取消", callback_data="dir:cancel"),
                        ]
                    ]
                ),
            )
        if parts[1] == "clear":
            await clear_user_location(user=query.from_user)
            await query.message.edit_text("位置已清除，技师状态已同步下线并下榜。")
            return await query.answer("已清除。")
        if parts[1] == "cancel":
            await query.message.edit_text("已取消。")
            return await query.answer()
        if parts[1] == "admin":
            if parts[2] == "pending":
                return await _show_admin_applications(query, page=int(parts[3]))
            if parts[2] == "teachers":
                return await _show_admin_teachers(query, int(parts[3]))
            if parts[2] == "detail":
                return await _show_admin_teacher_detail(query, int(parts[3]))
            if parts[2] == "approveask":
                user_id = int(parts[3])
                return await query.message.edit_text(
                    "批准后技师身份将在全部群和 Bot 中生效，确认继续吗？",
                    reply_markup=InlineKeyboardMarkup(
                        [
                            [
                                InlineKeyboardButton(
                                    "确认批准",
                                    callback_data=f"dir:admin:approve:{user_id}",
                                ),
                                InlineKeyboardButton(
                                    "取消",
                                    callback_data=f"dir:admin:detail:{user_id}",
                                ),
                            ]
                        ]
                    ),
                )
            if parts[2] == "approve":
                profile = await admin_decide_teacher(
                    actor=query.from_user,
                    user_id=int(parts[3]),
                    action="approve",
                )
                try:
                    await app.send_message(
                        int(profile["user_id"]),
                        "你的技师申请已通过，可以在菜单中自助上榜。",
                    )
                except Exception:
                    pass
                await query.answer("已批准。")
                return await _show_admin_teacher_detail(
                    query,
                    int(profile["user_id"]),
                )
            if parts[2] == "flow":
                return await _start_admin_reason_flow(
                    query,
                    action=parts[3],
                    user_id=int(parts[4]),
                )
            if parts[2] == "confirm":
                return await _confirm_admin_action(query, parts[3])
    except (DirectoryServiceError, ValueError) 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.command("cancel") & filters.private, group=-20)
async def cancel_directory_flow(_, message: Message):
    if not message.from_user:
        return
    admin_flow = _admin_flows.pop(message.from_user.id, None)
    location_flow = _location_flows.pop(message.from_user.id, None)
    if not admin_flow and not location_flow:
        return
    await message.reply_text("已取消当前操作。")
    raise StopPropagation


@app.on_message(filters.private & filters.text, group=13)
async def directory_private_text(_, message: Message):
    if not message.from_user:
        return
    flow = _admin_flows.get(message.from_user.id)
    if flow:
        if flow["expires_at"] < datetime.now(UTC):
            _admin_flows.pop(message.from_user.id, None)
            return await message.reply_text("操作已超时，请重新打开技师审批。")
        reason = str(message.text or "").strip()
        if not reason:
            return await message.reply_text("原因不能为空，请重新输入。")
        token = token_urlsafe(8)
        _admin_confirmations[token] = {
            **flow,
            "actor_id": message.from_user.id,
            "reason": reason,
            "expires_at": datetime.now(UTC) + timedelta(minutes=10),
        }
        _admin_flows.pop(message.from_user.id, None)
        return await message.reply_text(
            f"确认执行“{TEACHER_ACTION_LABELS.get(flow['action'], flow['action'])}”吗？"
            f"\n用户 ID：{flow['user_id']}\n原因：{escape(reason)}",
            reply_markup=InlineKeyboardMarkup(
                [
                    [
                        InlineKeyboardButton(
                            "确认执行",
                            callback_data=f"dir:admin:confirm:{token}",
                        ),
                        InlineKeyboardButton("取消", callback_data="dir:cancel"),
                    ]
                ]
            ),
        )
    location_expires_at = _location_flows.get(message.from_user.id)
    if location_expires_at:
        if location_expires_at < datetime.now(UTC):
            _location_flows.pop(message.from_user.id, None)
            return await message.reply_text("位置输入已超时，请重新选择手动位置。")
        raw = str(message.text or "").strip().replace("，", ",")
        values = [value.strip() for value in raw.split(",")]
        if len(values) != 2:
            return await message.reply_text(
                "格式不正确，请输入“纬度,经度”，例如：39.9042,116.4074。"
            )
        try:
            latitude, longitude = (float(value) for value in values)
            await update_user_location(
                user=message.from_user,
                longitude=longitude,
                latitude=latitude,
                source="manual_coordinates",
            )
        except (DirectoryServiceError, ValueError) as exc:
            return await message.reply_text(f"位置保存失败：{exc}")
        _location_flows.pop(message.from_user.id, None)
        await message.reply_text("手动位置已更新，可立即查找附近技师。")
        return await _refresh_keyboard(
            message,
            await get_directory_profile(message.from_user.id),
        )
    text = str(message.text or "").strip()
    if text == BUTTON_MANUAL_LOCATION:
        return await _start_manual_location(message)
    if text == BUTTON_NEARBY:
        return await _show_radius_picker(message)
    if text == BUTTON_RANK:
        return await _show_teacher_results(message, radius_km=None, page=1)
    if text == BUTTON_APPLY:
        return await _apply_teacher(message)
    if text == BUTTON_STATUS:
        return await _show_status(message)
    if text == "技师申请审核中":
        return await _show_status(message)
    if text == BUTTON_CLEAR_LOCATION:
        return await message.reply_text(
            "清除位置后会自动下榜并下线，确认继续吗？",
            reply_markup=InlineKeyboardMarkup(
                [
                    [
                        InlineKeyboardButton("确认清除", callback_data="dir:clear"),
                        InlineKeyboardButton("取消", callback_data="dir:cancel"),
                    ]
                ]
            ),
        )
    if text == BUTTON_APPROVAL:
        return await _show_admin_applications(message)
    action = {
        BUTTON_LIST: "list",
        BUTTON_UNLIST: "unlist",
        BUTTON_ONLINE: "online",
        BUTTON_OFFLINE: "offline",
        "上线": "online",
        "下线": "offline",
    }.get(text)
    if action:
        return await _self_state_command(message, action)


@app.on_message(filters.group & filters.text, group=13)
async def teacher_presence_group_text(_, message: Message):
    if not message.from_user:
        return
    text = str(message.text or "").strip()
    action = {
        "上线": "online",
        "下线": "offline",
        BUTTON_ONLINE: "online",
        BUTTON_OFFLINE: "offline",
    }.get(text)
    if not action:
        return
    profile = await get_directory_profile(message.from_user.id)
    if not profile or profile.get("application_status") != APPLICATION_APPROVED:
        return
    if not await _verify_group_member(message):
        return
    try:
        _, applied = await change_own_teacher_state(
            user=message.from_user,
            action=action,
            source="telegram_group",
            idempotency_key=(
                f"teacher-presence:{message.chat.id}:{message.id}:"
                f"{message.from_user.id}"
            ),
            chat_id=message.chat.id,
        )
        if applied:
            response = await message.reply_text(
                f"全平台状态已切换为{_presence_label(action == 'online')}。"
                + ("24 小时后将自动下线。" if action == "online" else "")
            )
            schedule_group_cleanup(message, response)
    except DirectoryServiceError as exc:
        response = await message.reply_text(str(exc))
        schedule_group_cleanup(message, response)


@app.on_chat_member_updated(group=13)
async def directory_chat_member_updated(_, update: ChatMemberUpdated):
    member = update.new_chat_member or update.old_chat_member
    if not member or not member.user or member.user.is_bot:
        return
    status = update.new_chat_member.status if update.new_chat_member else ChatMemberStatus.LEFT
    active = status in {
        ChatMemberStatus.OWNER,
        ChatMemberStatus.ADMINISTRATOR,
        ChatMemberStatus.MEMBER,
        ChatMemberStatus.RESTRICTED,
    }
    await observe_group_member(
        chat_id=update.chat.id,
        chat_title=update.chat.title or "",
        user=member.user,
        status=status_value(status),
        active=active,
        verified=True,
    )


start_presence_sweeper()