TelegramBotAdmin/wbb/modules/technician_services.py

from __future__ import annotations

import asyncio
from contextlib import suppress
from datetime import UTC, datetime, timedelta
from html import escape
from io import BytesIO
from typing import Any
from urllib.parse import quote

import qrcode
from pyrogram import StopPropagation, filters
from pyrogram.enums import ParseMode
from pyrogram.types import (
    CallbackQuery,
    InlineKeyboardButton,
    InlineKeyboardMarkup,
    KeyboardButton,
    Message,
    ReplyKeyboardMarkup,
    WebAppInfo,
)

import wbb
from wbb import BOT_USERNAME, app, log
from wbb.services.technician_reviews import ensure_technician_review_topic
from wbb.utils.dbservice import (
    ServiceDataError,
    begin_review_draft,
    cancel_service_order,
    claim_review_qr,
    confirm_service_quote,
    create_service_order,
    create_service_report,
    delete_review_draft,
    dispute_service_order,
    get_active_review_template,
    get_order_for_actor,
    get_technician_review_summary,
    get_technician_review_topic,
    get_technician_service_profile,
    instantiate_package_template,
    issue_review_qr,
    list_actor_service_orders,
    list_available_technicians,
    list_leaderboard,
    list_package_templates,
    list_public_reviews,
    normalize_quote,
    observe_customer,
    preview_review_qr,
    publish_technician_profile,
    record_service_funnel_event,
    redact_expired_addresses,
    reject_service_order,
    require_customer_allowed,
    save_review_draft_progress,
    set_technician_customer_block,
    start_service_order,
    submit_review,
    submit_service_quote,
)

__MODULE__ = "技师服务"
__HELP__ = """/find_technician - 浏览技师服务并直接咨询。
/technician_profile - 管理技师服务资料和套餐。
/review_teacher - 主动评价技师。
/technician_top - 查看审核评价排行。
旧 /teacher_top、/teacher_profile 命令继续兼容。"""

ORDER_LABELS = {
    "requested": "待技师处理",
    "quoted": "待顾客确认报价",
    "confirmed": "已确认，待开始服务",
    "in_progress": "服务中",
    "completion_pending": "待顾客扫码确认",
    "completed": "已完成",
    "rejected": "技师已拒绝",
    "canceled_customer": "顾客已取消",
    "canceled_technician": "技师已取消",
    "disputed": "争议处理中",
    "voided": "管理员已作废",
}
MODE_LABELS = {"at_store": "到店", "onsite": "上门"}
PRICE_MODE_LABELS = {
    "fixed": "固定价",
    "starting_at": "起步价",
    "range": "价格区间",
    "negotiable": "面议",
}
PRICE_UNIT_LABELS = {
    "per_service": "次",
    "per_hour": "小时",
    "per_item": "件",
    "per_visit": "次上门",
}
TRAVEL_FEE_LABELS = {
    "included": "已包含",
    "quoted": "最终报价确认",
}
PRICE_MODE_INPUTS = {
    **{key: key for key in PRICE_MODE_LABELS},
    **{label: key for key, label in PRICE_MODE_LABELS.items()},
}
PRICE_UNIT_INPUTS = {
    **{key: key for key in PRICE_UNIT_LABELS},
    "每次": "per_service",
    "每小时": "per_hour",
    "每件": "per_item",
    "每次上门": "per_visit",
}
TRAVEL_FEE_INPUTS = {
    "included": "included",
    "fixed": "fixed",
    "per_km": "per_km",
    "quoted": "quoted",
    "已包含": "included",
    "固定交通费": "fixed",
    "按公里": "per_km",
    "报价时确认": "quoted",
}
_flows: dict[int, dict[str, Any]] = {}
_address_redaction_task: asyncio.Task[None] | None = None


def _technician_mini_app_url() -> str:
    base_url = str(
        getattr(wbb, "SERVICE_TECHNICIAN_MINI_APP_URL", "") or ""
    ).strip()
    if not base_url:
        return ""
    separator = "&" if "?" in base_url else "?"
    bot_id = quote(str(getattr(wbb, "BOT_PROFILE_ID", "primary")), safe="")
    return f"{base_url}{separator}bot_id={bot_id}"


async def _address_redaction_sweeper() -> None:
    while True:
        try:
            await redact_expired_addresses()
        except Exception as exc:
            log.error(f"服务地址脱敏任务失败：{type(exc).__name__}")
        await asyncio.sleep(3600)


def start_address_redaction_sweeper() -> None:
    global _address_redaction_task
    if _address_redaction_task and not _address_redaction_task.done():
        return
    try:
        loop = asyncio.get_running_loop()
        _address_redaction_task = loop.create_task(
            _address_redaction_sweeper(),
            name="service-address-redaction-sweeper",
        )
    except RuntimeError:
        _address_redaction_task = None


async def stop_address_redaction_sweeper() -> None:
    global _address_redaction_task
    if _address_redaction_task:
        _address_redaction_task.cancel()
        with suppress(asyncio.CancelledError):
            await _address_redaction_task
    _address_redaction_task = None


def _flow(user_id: int, stage: str, **values: Any) -> dict[str, Any]:
    user_id = int(user_id)
    value = {
        "stage": stage,
        "expires_at": datetime.now(UTC) + timedelta(minutes=20),
        **values,
    }
    _flows[user_id] = value
    try:
        loop = asyncio.get_running_loop()

        def expire() -> None:
            if _flows.get(user_id) is value:
                _flows.pop(user_id, None)

        loop.call_later(20 * 60, expire)
    except RuntimeError:
        pass
    return value


def _current_flow(user_id: int) -> dict[str, Any] | None:
    flow = _flows.get(int(user_id))
    if flow and flow["expires_at"] >= datetime.now(UTC):
        return flow
    _flows.pop(int(user_id), None)
    return None


def _location_keyboard(label: str) -> ReplyKeyboardMarkup:
    return ReplyKeyboardMarkup(
        [[KeyboardButton(label, request_location=True)]],
        resize_keyboard=True,
        one_time_keyboard=True,
        selective=True,
    )


def _short(value: Any, limit: int) -> str:
    text = str(value or "").strip()
    return text if len(text) <= limit else f"{text[: limit - 1]}…"


def _package_price_text(package: dict[str, Any]) -> str:
    if package.get("price_display"):
        return str(package["price_display"])
    mode = package.get("price_mode")
    currency = package.get("currency") or ""
    minimum = package.get("min_price") or "0"
    maximum = package.get("max_price") or minimum
    unit = PRICE_UNIT_LABELS.get(package.get("price_unit"), "次")
    if mode == "negotiable":
        return "面议"
    if mode == "starting_at":
        return f"{currency} {minimum} 起/{unit}"
    if mode == "range":
        return f"{currency} {minimum}–{maximum}/{unit}"
    return f"{currency} {minimum}/{unit}"


def _travel_fee_text(package: dict[str, Any]) -> str:
    travel = package.get("travel_fee") or {}
    mode = travel.get("mode")
    if mode == "fixed":
        return f"{package.get('currency', '')} {travel.get('amount') or '0'}"
    if mode == "per_km":
        return f"{package.get('currency', '')} {travel.get('per_km') or '0'}/公里"
    return TRAVEL_FEE_LABELS.get(mode, "最终报价确认")


async def _send_message(target: Message | CallbackQuery, text: str, **kwargs: Any):
    message = target.message if isinstance(target, CallbackQuery) else target
    return await message.reply_text(text, **kwargs)


async def _notify(user_id: int, text: str, **kwargs: Any) -> None:
    try:
        await app.send_message(int(user_id), text, **kwargs)
    except Exception:
        pass


async def _require_terms(target: Message | CallbackQuery) -> bool:
    if not target.from_user:
        return False
    await observe_customer(target.from_user)
    try:
        await require_customer_allowed(int(target.from_user.id))
        return True
    except ServiceDataError as exc:
        if exc.code != "terms_required":
            await _send_message(target, str(exc))
            return False
    await _send_message(
        target,
        "<b>使用技师咨询与评价前请确认</b>\n\n"
        "平台提供技师信息展示、咨询入口和公开评价，不处理预约、支付或退款。"
        "服务细节和具体位置由双方在 Telegram 私聊中沟通。"
        "\n\n点击即表示你接受服务规则和隐私说明。",
        parse_mode=ParseMode.HTML,
        reply_markup=InlineKeyboardMarkup(
            [[InlineKeyboardButton("接受并继续", callback_data="svc:terms")]]
        ),
    )
    return False


async def send_service_home(target: Message | CallbackQuery) -> None:
    if not target.from_user:
        return
    if not await _require_terms(target):
        return
    await _send_message(
        target,
        "<b>技师服务咨询</b>\n浏览技师的服务、价格和公开评价，需要时直接私聊咨询。",
        parse_mode=ParseMode.HTML,
        reply_markup=InlineKeyboardMarkup(
            [
                [InlineKeyboardButton("浏览技师", callback_data="svc:find")],
                [InlineKeyboardButton("主动评价", callback_data="svc:review_start")],
                [InlineKeyboardButton("审核评价排行", callback_data="svc:top:all:1")],
            ]
        ),
    )


async def _ask_search_location(target: Message | CallbackQuery) -> None:
    if not target.from_user or not await _require_terms(target):
        return
    await _show_nearby(
        target,
        longitude=None,
        latitude=None,
        page=1,
    )


async def _show_nearby(
    target: Message | CallbackQuery,
    *,
    longitude: float | None,
    latitude: float | None,
    page: int = 1,
) -> None:
    if page == 1:
        await record_service_funnel_event(target.from_user.id, "directory_viewed")
    items, total = await list_available_technicians(
        longitude=longitude,
        latitude=latitude,
        page=page,
        page_size=6,
    )
    if not items:
        return await _send_message(target, "当前暂时没有公开接单的技师。")
    rows = []
    lines = [f"<b>技师目录</b>（共 {total} 位）"]
    for item in items:
        lines.append(
            f"\n<b>{escape(item.get('display_name') or str(item['user_id']))}</b>\n"
            f"{escape(item.get('public_area_text') or '服务区域由技师填写')}\n"
            f"{escape(item.get('headline') or '')}"
        )
        rows.append(
            [
                InlineKeyboardButton(
                    f"查看 {item.get('display_name') or item['user_id']}",
                    callback_data=f"svc:tech:{item['user_id']}",
                )
            ]
        )
    if page > 1:
        rows.append([InlineKeyboardButton("上一页", callback_data=f"svc:near:{page - 1}")])
    if page * 6 < total:
        rows.append([InlineKeyboardButton("下一页", callback_data=f"svc:near:{page + 1}")])
    await _send_message(
        target,
        "\n".join(lines),
        parse_mode=ParseMode.HTML,
        reply_markup=InlineKeyboardMarkup(rows),
    )


async def _show_technician(target: Message | CallbackQuery, technician_id: int) -> None:
    if not await _require_terms(target):
        return
    await record_service_funnel_event(target.from_user.id, "technician_viewed")
    profile = await get_technician_service_profile(technician_id)
    service_profile = profile.get("service_profile") or {}
    if (
        profile.get("application_status") != "approved"
        or not profile.get("listed")
        or not service_profile.get("is_complete")
        or not service_profile.get("accepting_requests", True)
    ):
        raise ServiceDataError("technician_unavailable", "该技师当前未公开接单。")
    lines = [
        f"<b>{escape(profile.get('display_name') or str(technician_id))}</b>",
        escape(service_profile.get("headline") or ""),
        escape(_short(service_profile.get("bio"), 300)),
        f"服务区域：{escape(service_profile.get('public_area_text') or '由技师填写')}",
        "",
        "<b>服务套餐</b>",
    ]
    review_summary = await get_technician_review_summary(technician_id)
    lines.insert(
        4,
        (
            f"审核评价：{review_summary['average_score']:.2f} 分 · "
            f"{review_summary['review_count']} 条"
            if review_summary["review_count"]
            else "审核评价：暂无"
        ),
    )
    rows = [
        [
            InlineKeyboardButton(
                "直接咨询技师",
                url=f"https://t.me/{profile['username']}",
            ),
            InlineKeyboardButton(
                "评价该技师",
                callback_data=f"svc:review:{technician_id}",
            ),
        ]
    ]
    for package in service_profile.get("packages", []):
        extra_fee = str(
            (package.get("travel_fee") or {}).get("description") or ""
        ).strip()
        if not extra_fee:
            extra_fee = _travel_fee_text(package)
        package_lines = [
            f"\n<b>{escape(package['name'])}</b>\n"
            f"服务内容：{escape(_short(package['description'], 180))}\n"
            f"服务价格：{escape(_package_price_text(package))}\n"
            f"服务时长：约 {package.get('duration_minutes') or '未设'} 分钟\n"
            f"额外费用：{escape(_short(extra_fee, 120))}",
        ]
        lines.append("\n".join(package_lines))
    topic = await get_technician_review_topic(technician_id)
    if not topic:
        try:
            topic = await ensure_technician_review_topic(
                str(getattr(app, "bot_token", "") or ""),
                technician_id,
            )
        except Exception:
            topic = None
    if topic and topic.get("topic_url"):
        rows.append(
            [
                InlineKeyboardButton(
                    "查看全部公开评价",
                    url=str(topic["topic_url"]),
                )
            ]
        )
    public_reviews, _ = await list_public_reviews(technician_id, page_size=3)
    if public_reviews:
        lines.append("\n<b>最近审核评价</b>")
        for review in public_reviews:
            choice_values = []
            for value in (review.get("choice_answers") or {}).values():
                choice_values.extend(value if isinstance(value, list) else [value])
            comment = "；".join(
                [
                    *(str(value) for value in choice_values),
                    *(review.get("text_answers", {}).values()),
                ]
            ) or "未填写补充内容"
            lines.append(
                f"{escape(review['source_label'])} · {review['score']:.2f} 分 · "
                f"{escape(review.get('customer_name') or '匿名顾客')}\n"
                f"{escape(comment[:120])}"
            )
    await _send_message(
        target,
        "\n".join(lines),
        parse_mode=ParseMode.HTML,
        reply_markup=InlineKeyboardMarkup(rows),
    )


async def _choose_request_mode(
    query: CallbackQuery, technician_id: int, package_id: str
) -> None:
    profile = await get_technician_service_profile(technician_id)
    package = next(
        (
            item
            for item in (profile.get("service_profile") or {}).get("packages", [])
            if item.get("package_id") == package_id
        ),
        None,
    )
    if not package:
        raise ServiceDataError("package_not_found", "套餐不存在或已下架。")
    await record_service_funnel_event(query.from_user.id, "request_started")
    rows = [
        [
            InlineKeyboardButton(
                MODE_LABELS[mode],
                callback_data=f"svc:mode:{technician_id}:{package_id}:{mode}",
            )
        ]
        for mode in package["service_modes"]
    ]
    await query.message.reply_text(
        f"已选择「{package['name']}」，请选择服务方式。报价确认前不会展示精确地址。",
        reply_markup=InlineKeyboardMarkup(rows),
    )


async def _request_mode(
    query: CallbackQuery,
    technician_id: int,
    package_id: str,
    service_mode: str,
) -> None:
    current = _current_flow(query.from_user.id) or {}
    values = {
        "technician_id": technician_id,
        "package_id": package_id,
        "service_mode": service_mode,
    }
    if service_mode == "onsite":
        _flow(query.from_user.id, "request_location", **values)
        await query.message.reply_text(
            "请发送上门位置。精确坐标会单独加密，报价确认前技师只能看到距离区间。",
            reply_markup=_location_keyboard("发送上门位置"),
        )
    else:
        if "longitude" not in current or "latitude" not in current:
            raise ServiceDataError(
                "customer_location_required", "位置已过期，请重新查找附近技师。"
            )
        values.update(
            {
                "longitude": current["longitude"],
                "latitude": current["latitude"],
            }
        )
        _flow(query.from_user.id, "request_requirements", **values)
        await query.message.reply_text("请描述服务需求和补充说明。")


async def _preview_request(message: Message, flow: dict[str, Any]) -> None:
    profile = await get_technician_service_profile(flow["technician_id"])
    package = next(
        item
        for item in profile["service_profile"]["packages"]
        if item["package_id"] == flow["package_id"]
    )
    text = (
        "<b>确认服务请求</b>\n"
        f"技师：{escape(profile.get('display_name') or str(profile['user_id']))}\n"
        f"套餐：{escape(package['name'])}\n"
        f"方式：{MODE_LABELS[flow['service_mode']]}\n"
        f"需求：{escape(flow['requirements'])}\n\n"
        "提交后等待技师拒绝或给出最终报价。"
    )
    await message.reply_text(
        text,
        parse_mode=ParseMode.HTML,
        reply_markup=InlineKeyboardMarkup(
            [
                [InlineKeyboardButton("确认提交", callback_data="svc:request_submit")],
                [InlineKeyboardButton("取消", callback_data="svc:flow_cancel")],
            ]
        ),
    )


def _order_text(order: dict[str, Any]) -> str:
    quote = order.get("quote_snapshot") or {}
    lines = [
        f"<b>服务单 {escape(order['order_id'][:8])}</b>",
        f"状态：{ORDER_LABELS.get(order['status'], order['status'])}",
        f"技师：{escape(order.get('technician_name') or str(order['technician_id']))}",
        f"顾客：{escape(order.get('customer_name') or str(order['customer_id']))}",
        f"套餐：{escape((order.get('package_snapshot') or {}).get('name') or '')}",
        f"方式：{MODE_LABELS.get(order['service_mode'], order['service_mode'])}",
        f"距离：{escape(order.get('distance_band') or '')}",
        f"需求：{escape(order.get('requirements') or '')}",
    ]
    if quote:
        lines.extend(
            [
                "",
                "<b>最终报价</b>",
                f"服务金额：{quote['currency']} {quote['base_amount']}",
                f"交通费：{quote['travel_fee']}",
                f"优惠：{quote['discount']}",
                f"总价：<b>{quote['currency']} {quote['total_amount']}</b>",
                f"说明：{escape(quote.get('note') or '无')}",
            ]
        )
    return "\n".join(lines)


def _order_markup(order: dict[str, Any], actor_id: int) -> InlineKeyboardMarkup | None:
    is_customer = actor_id == int(order["customer_id"])
    is_technician = actor_id == int(order["technician_id"])
    status = order["status"]
    rows: list[list[InlineKeyboardButton]] = []
    if status in {"requested", "quoted"} and is_technician:
        rows.append(
            [
                InlineKeyboardButton("提交最终报价", callback_data=f"svc:quote:{order['order_id']}"),
                InlineKeyboardButton("拒绝", callback_data=f"svc:reject:{order['order_id']}"),
            ]
        )
    if status == "quoted" and is_customer:
        rows.append(
            [InlineKeyboardButton("确认最终报价", callback_data=f"svc:confirm:{order['order_id']}")]
        )
    if status == "confirmed" and is_technician:
        rows.append(
            [InlineKeyboardButton("开始服务", callback_data=f"svc:start:{order['order_id']}")]
        )
    if status == "in_progress" and is_technician:
        rows.append(
            [InlineKeyboardButton("完成并生成评价二维码", callback_data=f"svc:finish:{order['order_id']}")]
        )
    if status == "completion_pending" and is_technician:
        rows.append(
            [InlineKeyboardButton("二维码过期后重新生成", callback_data=f"svc:finish:{order['order_id']}")]
        )
    if status in {"confirmed", "in_progress", "completion_pending"}:
        other_id = order["technician_id"] if is_customer else order["customer_id"]
        rows.append([InlineKeyboardButton("联系对方协商时间", url=f"tg://user?id={other_id}")])
    if status in {"requested", "quoted", "confirmed", "in_progress", "completion_pending"} and (
        is_customer or is_technician
    ):
        rows.append(
            [
                InlineKeyboardButton("取消服务单", callback_data=f"svc:cancel:{order['order_id']}"),
                InlineKeyboardButton("发起争议", callback_data=f"svc:dispute:{order['order_id']}"),
            ]
        )
    return InlineKeyboardMarkup(rows) if rows else None


async def _show_order(target: Message | CallbackQuery, order_id: str) -> None:
    order = await get_order_for_actor(order_id, target.from_user.id, reveal_address=True)
    text = _order_text(order)
    exact = order.get("exact_address")
    if exact:
        text += f"\n\n<b>履约地址</b>\n{escape(exact.get('address_text') or '')}"
    await _send_message(
        target,
        text,
        parse_mode=ParseMode.HTML,
        reply_markup=_order_markup(order, target.from_user.id),
    )


async def _show_orders(target: Message | CallbackQuery, *, role: str, page: int) -> None:
    items, total = await list_actor_service_orders(
        target.from_user.id, role=role, page=page, page_size=8
    )
    if not items:
        return await _send_message(target, "暂时没有服务单。")
    rows = [
        [
            InlineKeyboardButton(
                f"{item['order_id'][:8]} · {ORDER_LABELS.get(item['status'], item['status'])}",
                callback_data=f"svc:order:{item['order_id']}",
            )
        ]
        for item in items
    ]
    navigation = []
    if page > 1:
        navigation.append(
            InlineKeyboardButton("上一页", callback_data=f"svc:orders:{role}:{page - 1}")
        )
    if page * 8 < total:
        navigation.append(
            InlineKeyboardButton("下一页", callback_data=f"svc:orders:{role}:{page + 1}")
        )
    if navigation:
        rows.append(navigation)
    await _send_message(
        target,
        f"<b>我的服务单</b>（共 {total} 单）",
        parse_mode=ParseMode.HTML,
        reply_markup=InlineKeyboardMarkup(rows),
    )


async def _start_quote(target: Message | CallbackQuery, order_id: str) -> None:
    actor_id = target.from_user.id
    order = await get_order_for_actor(order_id, actor_id)
    if actor_id != order["technician_id"] or order["status"] not in {
        "requested",
        "quoted",
    }:
        raise ServiceDataError("invalid_order_state", "当前服务单不能报价。")
    _flow(actor_id, "quote_base", order=order, quote={})
    reference = order["package_snapshot"]
    await _send_message(
        target,
        f"套餐参考价：{reference['currency']} {reference['min_price']} - {reference['max_price']}。\n"
        "第 1/5 步：请输入本次服务金额。"
    )


def _parse_addons(text: str) -> list[dict[str, Any]]:
    if text.strip() in {"跳过", "无", "0"}:
        return []
    addons = []
    for raw in text.replace("；", ";").split(";"):
        values = [item.strip() for item in raw.split("|")]
        if len(values) != 3:
            raise ServiceDataError("invalid_addons", "加项格式应为：名称|数量|单价。")
        addons.append({"name": values[0], "quantity": int(values[1]), "unit_price": values[2]})
    return addons


async def _preview_quote(message: Message, flow: dict[str, Any]) -> None:
    quote = normalize_quote(flow["quote"], package=flow["order"]["package_snapshot"])
    flow["normalized_quote"] = quote
    flow["stage"] = "quote_preview"
    await message.reply_text(
        "<b>确认最终报价</b>\n"
        f"服务金额：{quote['currency']} {quote['base_amount']}\n"
        f"交通费：{quote['travel_fee']}\n"
        f"加项数量：{len(quote['addons'])}\n"
        f"优惠：{quote['discount']}\n"
        f"总价：<b>{quote['currency']} {quote['total_amount']}</b>\n"
        f"说明：{escape(quote['note'] or '无')}\n\n"
        "顾客确认后不能原地改价；如需改价必须取消并重新发起服务单。",
        parse_mode=ParseMode.HTML,
        reply_markup=InlineKeyboardMarkup(
            [
                [InlineKeyboardButton("提交报价", callback_data="svc:quote_submit")],
                [InlineKeyboardButton("取消", callback_data="svc:flow_cancel")],
            ]
        ),
    )


async def _issue_qr(target: Message | CallbackQuery, order_id: str) -> None:
    qr_record, token = await issue_review_qr(order_id, target.from_user.id)
    deep_link = f"https://t.me/{BOT_USERNAME}?start=review_{token}"
    image = qrcode.make(deep_link)
    output = BytesIO()
    output.name = f"service-{order_id[:8]}-review.png"
    image.save(output, format="PNG")
    output.seek(0)
    await app.send_photo(
        target.from_user.id,
        photo=output,
        caption=(
            f"服务单 {order_id[:8]} 的完成评价二维码\n"
            f"仅顾客 {qr_record['customer_id']} 可领取，{qr_record['expires_at']:%Y-%m-%d %H:%M} UTC 前有效。\n\n"
            f"备用链接：{deep_link}"
        ),
    )
    await _notify(
        qr_record["customer_id"],
        f"技师已发起服务完成确认。请扫码或点击备用链接确认：{deep_link}",
    )


async def open_review_qr(message: Message, token: str) -> None:
    if not message.from_user or not await _require_terms(message):
        return
    try:
        record = await preview_review_qr(token, message.from_user.id)
    except ServiceDataError as exc:
        return await message.reply_text(str(exc))
    if record.get("status") == "claimed":
        await message.reply_text("服务单已确认完成，继续填写评价。")
        return await _begin_review(
            message,
            technician_id=record["technician_id"],
            source="qr_verified",
            qr_id=record["qr_id"],
        )
    quote = record.get("quote_snapshot") or {}
    await message.reply_text(
        "<b>确认服务完成</b>\n"
        f"服务单：{escape(record['order_id'][:8])}\n"
        f"套餐：{escape((record.get('package_snapshot') or {}).get('name') or '')}\n"
        f"最终报价：{quote.get('currency', '')} {quote.get('total_amount', '')}\n\n"
        "确认后服务单将进入已完成，并立即打开评价模板。",
        parse_mode=ParseMode.HTML,
        reply_markup=InlineKeyboardMarkup(
            [[InlineKeyboardButton("确认完成并评价", callback_data=f"svc:claim:{token}")]]
        ),
    )


async def _begin_review(
    target: Message | CallbackQuery,
    *,
    technician_id: int,
    source: str,
    package_id: str | None = None,
    qr_id: str | None = None,
) -> None:
    template = await get_active_review_template()
    draft = await begin_review_draft(
        customer_id=target.from_user.id,
        technician_id=technician_id,
        source=source,
        package_id=package_id,
        qr_id=qr_id,
        template=template,
    )
    _flow(
        target.from_user.id,
        "review_question",
        technician_id=technician_id,
        source=source,
        package_id=package_id,
        qr_id=qr_id,
        template=draft["template_snapshot"],
        question_index=int(
            draft.get("question_index")
            or draft.get("rating_index")
            or 0
        ),
        answers=dict(draft.get("answers") or {}),
    )
    await _ask_review_question(target)


def _review_options_markup(
    question: dict[str, Any],
    selected: list[str] | None = None,
) -> InlineKeyboardMarkup:
    selected_values = set(selected or [])
    action = "single" if question["type"] == "single_choice" else "multi"
    buttons = []
    for index, option in enumerate(question["options"]):
        prefix = "✓ " if str(option["value"]) in selected_values else ""
        buttons.append(
            InlineKeyboardButton(
                f"{prefix}{option['label']}",
                callback_data=f"svc:{action}:{index}",
            )
        )
    rows = [buttons[index : index + 2] for index in range(0, len(buttons), 2)]
    if question["type"] == "multiple_choice":
        rows.append(
            [
                InlineKeyboardButton(
                    "完成选择",
                    callback_data="svc:multi_done",
                )
            ]
        )
    return InlineKeyboardMarkup(rows)


async def _ask_review_question(target: Message | CallbackQuery) -> None:
    flow = _current_flow(target.from_user.id)
    if not flow:
        return
    questions = flow["template"]["questions"]
    index = flow["question_index"]
    if index >= len(questions):
        flow["stage"] = "review_anonymous"
        return await _send_message(
            target,
            "评价已填写完成。请选择公开显示方式，提交后由管理员审核。",
            reply_markup=InlineKeyboardMarkup(
                [
                    [InlineKeyboardButton("匿名发布（默认）", callback_data="svc:anonymous:1")],
                    [InlineKeyboardButton("显示昵称", callback_data="svc:anonymous:0")],
                ]
            ),
        )
    question = questions[index]
    if question["type"] == "text":
        flow["stage"] = "review_text"
        return await _send_message(
            target,
            f"评价 {index + 1}/{len(questions)}：<b>{escape(question['label'])}</b>\n"
            f"{escape(question.get('description') or '')}\n"
            f"最多 {question['max_length']} 字。"
            + ("" if question.get("required") else " 可发送“跳过”。"),
            parse_mode=ParseMode.HTML,
        )
    flow["stage"] = "review_question"
    selected = flow["answers"].get(question["question_id"])
    await _send_message(
        target,
        f"评价 {index + 1}/{len(questions)}：<b>{escape(question['label'])}</b>\n"
        f"{escape(question.get('description') or '')}",
        parse_mode=ParseMode.HTML,
        reply_markup=_review_options_markup(
            question,
            selected if isinstance(selected, list) else None,
        ),
    )


async def _submit_pending_review(query: CallbackQuery, anonymous: bool) -> None:
    flow = _current_flow(query.from_user.id)
    if not flow or flow["stage"] != "review_anonymous":
        raise ServiceDataError("review_flow_expired", "评价流程已过期，请重新发起。")
    review = await submit_review(
        customer=query.from_user,
        technician_id=flow["technician_id"],
        source=flow["source"],
        answers=flow["answers"],
        anonymous=anonymous,
        package_id=flow.get("package_id"),
        qr_id=flow.get("qr_id"),
        template_snapshot=flow["template"],
    )
    await delete_review_draft(query.from_user.id)
    _flows.pop(query.from_user.id, None)
    await query.message.reply_text(
        f"评价 {review['review_id'][:8]} 已提交审核。审核通过后会立即发布到该技师的公开评价话题。"
    )


async def _choose_review_package(target: Message | CallbackQuery, technician_id: int) -> None:
    profile = await get_technician_service_profile(technician_id)
    packages = (profile.get("service_profile") or {}).get("packages", [])
    if not packages:
        raise ServiceDataError("package_not_found", "该技师暂无可评价套餐。")
    await _begin_review(
        target,
        technician_id=technician_id,
        source="student_initiated",
    )


async def _show_top(target: Message | CallbackQuery, category: str, page: int) -> None:
    items, total = await list_leaderboard(
        category="" if category == "all" else category,
        page=page,
        page_size=10,
    )
    lines = [
        "<b>审核评价排行</b>",
        "每条管理员审核通过的评价都会公开并参与排行。",
        "公式：v/(v+5)×技师平均分 + 5/(v+5)×分类平均分。",
    ]
    for index, item in enumerate(items, start=(page - 1) * 10 + 1):
        lines.append(
            f"\n{index}. <b>{escape(item.get('display_name') or str(item['technician_id']))}</b>"
            f" · {escape(item['category'])}\n"
            f"排行分 {item['rank_score']:.2f} · 平均 {item['average_score']:.2f} · {item['review_count']} 条"
        )
    if not items:
        lines.append("\n暂时还没有审核通过的评价。")
    rows = []
    if page > 1:
        rows.append([InlineKeyboardButton("上一页", callback_data=f"svc:top:{category}:{page - 1}")])
    if page * 10 < total:
        rows.append([InlineKeyboardButton("下一页", callback_data=f"svc:top:{category}:{page + 1}")])
    await _send_message(
        target,
        "\n".join(lines),
        parse_mode=ParseMode.HTML,
        reply_markup=InlineKeyboardMarkup(rows) if rows else None,
    )


async def _show_profile(target: Message | CallbackQuery) -> None:
    try:
        profile = await get_technician_service_profile(target.from_user.id)
    except ServiceDataError as exc:
        return await _send_message(target, str(exc))
    if profile.get("application_status") != "approved":
        return await _send_message(target, "只有已通过管理员认证的技师可以发布服务资料。")
    mini_app_url = _technician_mini_app_url()
    service_profile = profile.get("service_profile") or {}
    if service_profile.get("is_complete"):
        package_items = service_profile.get("packages", [])
        packages = "、".join(item["name"] for item in package_items)
        rows = []
        if mini_app_url:
            rows.append(
                [
                    InlineKeyboardButton(
                        "快速管理套餐",
                        web_app=WebAppInfo(url=mini_app_url),
                    )
                ]
            )
        rows.append(
            [
                InlineKeyboardButton(
                    "暂停接单" if service_profile.get("accepting_requests") else "开始接单",
                    callback_data="svc:profile_toggle",
                )
            ]
        )
        if not mini_app_url and len(package_items) < 5:
            rows.append(
                [InlineKeyboardButton("新增套餐", callback_data="svc:profile_add")]
            )
        rows.append(
            [
                InlineKeyboardButton(
                    (
                        "使用聊天配置（备用）"
                        if mini_app_url
                        else "重建资料与套餐"
                    ),
                    callback_data="svc:profile_reconfigure",
                )
            ]
        )
        return await _send_message(
            target,
            "<b>我的技师服务资料</b>\n"
            f"简介：{escape(service_profile.get('headline') or '')}\n"
            f"区域：{escape(service_profile.get('public_area_text') or '')}\n"
            f"接单：{'开启' if service_profile.get('accepting_requests') else '暂停'}\n"
            f"套餐：{escape(packages)}",
            parse_mode=ParseMode.HTML,
            reply_markup=InlineKeyboardMarkup(rows),
        )
    if mini_app_url:
        return await _send_message(
            target,
            "使用快速版选择模板，只需确认价格、时长和服务区域即可发布。"
            "聊天分步配置继续保留为备用入口。",
            reply_markup=InlineKeyboardMarkup(
                [
                    [
                        InlineKeyboardButton(
                            "快速创建套餐",
                            web_app=WebAppInfo(url=mini_app_url),
                        )
                    ],
                    [
                        InlineKeyboardButton(
                            "使用聊天配置（备用）",
                            callback_data="svc:profile_reconfigure",
                        )
                    ],
                ]
            ),
        )
    await _show_profile_sources(target)


async def _show_profile_sources(target: Message | CallbackQuery) -> None:
    flow = _current_flow(target.from_user.id)
    if not flow or flow.get("stage") != "profile_source":
        flow = _flow(target.from_user.id, "profile_source", profile_mode="replace")
    profile_mode = flow.get("profile_mode", "replace")
    templates = await list_package_templates()
    rows = [
        [
            InlineKeyboardButton(
                item["name"], callback_data=f"svc:profile_template:{item['template_id']}"
            )
        ]
        for item in templates
        if item.get("status") == "enabled"
    ]
    rows.append(
        [InlineKeyboardButton("完全自建套餐", callback_data="svc:profile_custom")]
    )
    await _send_message(
        target,
        (
            "选择一个后台套餐模板新增套餐，或完全自建套餐。"
            if profile_mode == "append"
            else "选择一个后台套餐模板开始，或完全自建套餐。"
        )
        + "模板只用于预填，发布时会保存独立快照；后台后续修改不会覆盖你的套餐。",
        reply_markup=InlineKeyboardMarkup(rows),
    )


async def _start_profile_flow(query: CallbackQuery, template_id: str) -> None:
    current = _current_flow(query.from_user.id) or {}
    package = await instantiate_package_template(template_id)
    flow = _flow(
        query.from_user.id,
        "template_preview",
        package=package,
        profile={},
        profile_mode=current.get("profile_mode", "replace"),
    )
    await _show_template_package_preview(query, flow)


async def _show_template_package_preview(
    target: Message | CallbackQuery,
    flow: dict[str, Any],
) -> None:
    package = flow["package"]
    flow["stage"] = "template_preview"
    await _send_message(
        target,
        "<b>模板套餐预览</b>\n"
        f"名称：{escape(package['name'])}\n"
        f"分类：{escape(package['category'])}\n"
        f"方式：{'/'.join(MODE_LABELS[item] for item in package['service_modes'])}\n"
        f"价格：{package['currency']} {package['min_price']} - {package['max_price']}\n"
        f"时长：{package.get('duration_minutes') or '未设'} 分钟\n"
        f"说明：{escape(package['description'])}\n\n"
        "可只修改需要调整的部分；未修改字段保留模板快照。",
        parse_mode=ParseMode.HTML,
        reply_markup=InlineKeyboardMarkup(
            [
                [InlineKeyboardButton("直接使用并填写资料", callback_data="svc:profile_template_use")],
                [
                    InlineKeyboardButton("名称", callback_data="svc:profile_edit:name"),
                    InlineKeyboardButton("分类", callback_data="svc:profile_edit:category"),
                    InlineKeyboardButton("说明", callback_data="svc:profile_edit:description"),
                ],
                [
                    InlineKeyboardButton("标签", callback_data="svc:profile_edit:tags"),
                    InlineKeyboardButton("方式", callback_data="svc:profile_edit:modes"),
                ],
                [
                    InlineKeyboardButton("价格", callback_data="svc:profile_edit:price"),
                    InlineKeyboardButton("币种/计价", callback_data="svc:profile_edit:billing"),
                    InlineKeyboardButton("时长/半径", callback_data="svc:profile_edit:duration"),
                ],
                [
                    InlineKeyboardButton("包含与交通", callback_data="svc:profile_edit:details"),
                    InlineKeyboardButton("标准加项", callback_data="svc:profile_edit:addons"),
                ],
            ]
        ),
    )


async def _start_custom_package_flow(query: CallbackQuery) -> None:
    current = _current_flow(query.from_user.id) or {}
    _flow(
        query.from_user.id,
        "custom_name",
        package={},
        profile={},
        profile_mode=current.get("profile_mode", "replace"),
    )
    await query.message.reply_text("自建套餐 1/9：请输入套餐名称。")


async def _publish_profile(
    target: Message | CallbackQuery, flow: dict[str, Any]
) -> None:
    user_id = target.from_user.id
    if flow.get("profile_mode") == "append":
        current = await get_technician_service_profile(user_id)
        current_values = current.get("service_profile") or {}
        packages = list(current_values.get("packages", []))
        if len(packages) >= 5:
            raise ServiceDataError("invalid_package_count", "每位技师最多发布 5 个套餐。")
        packages.append(flow["package"])
        values = {
            "headline": current_values.get("headline"),
            "bio": current_values.get("bio"),
            "contact_hours": current_values.get("contact_hours"),
            "public_area_text": current_values.get("public_area_text"),
            "tags": current_values.get("tags", []),
            "venue": current_values.get("venue", {}),
            "onsite_policy": current_values.get("onsite_policy", {}),
            "packages": packages,
            "accepting_requests": current_values.get("accepting_requests", True),
        }
    else:
        profile_values = dict(flow["profile"])
        values = {
            "headline": profile_values["headline"],
            "bio": profile_values["bio"],
            "contact_hours": profile_values["contact_hours"],
            "public_area_text": profile_values["public_area_text"],
            "tags": flow["package"].get("tags", []),
            "venue": {
                "name": profile_values["venue_name"],
                "address_hint": profile_values["venue_address"],
            },
            "onsite_policy": {"description": profile_values["onsite_description"]},
            "packages": [flow["package"]],
            "accepting_requests": True,
        }
    profile = await publish_technician_profile(
        user_id, values, actor_id=user_id
    )
    with suppress(Exception):
        await ensure_technician_review_topic(
            str(getattr(app, "bot_token", "") or ""),
            user_id,
        )
    _flows.pop(user_id, None)
    await _send_message(
        target,
        f"服务资料已发布，共 {len(profile['service_profile']['packages'])} 个套餐。"
        "套餐已保存模板版本和完整快照。"
    )


@app.on_message(filters.private & filters.command("find_technician"))
async def find_technician_command(_, message: Message):
    await _ask_search_location(message)


@app.on_message(filters.private & filters.command(["my_services", "my_orders"]))
async def my_services_command(_, message: Message):
    if await _require_terms(message):
        await _show_orders(message, role="all", page=1)


@app.on_message(filters.private & filters.command(["technician_top", "teacher_top"]))
async def technician_top_command(_, message: Message):
    if not await _require_terms(message):
        return
    category = message.command[1] if len(message.command) > 1 else "all"
    await _show_top(message, category, 1)


@app.on_message(filters.private & filters.command(["technician_profile", "teacher_profile"]))
async def technician_profile_command(_, message: Message):
    await _show_profile(message)


@app.on_message(filters.private & filters.command("review_teacher"))
async def review_teacher_command(_, message: Message):
    if not await _require_terms(message):
        return
    if len(message.command) < 2 or not message.command[1].isdigit():
        return await message.reply_text("请使用 /review_teacher 技师ID，或从技师详情点击“主动评价”。")
    await _choose_review_package(message, int(message.command[1]))


@app.on_message(filters.private & filters.command("review_qr"))
async def review_qr_command(_, message: Message):
    if len(message.command) < 2:
        return await message.reply_text("请使用 /review_qr 服务单ID，或在服务单详情点击生成二维码。")
    try:
        await _issue_qr(message, message.command[1])
    except ServiceDataError as exc:
        await message.reply_text(str(exc))


@app.on_message(filters.private & filters.command("block_customer"))
async def block_customer_command(_, message: Message):
    if len(message.command) < 3 or not message.command[1].isdigit():
        return await message.reply_text("请使用 /block_customer 顾客ID 原因。")
    try:
        blocked = await set_technician_customer_block(
            technician_id=message.from_user.id,
            customer_id=int(message.command[1]),
            active=True,
            reason=" ".join(message.command[2:]),
        )
        await message.reply_text(f"已拉黑顾客 {blocked['customer_id']}，之后的新请求会被拒绝。")
    except ServiceDataError as exc:
        await message.reply_text(str(exc))


@app.on_message(filters.private & filters.command("report_service"))
async def report_service_command(_, message: Message):
    if len(message.command) < 3:
        return await message.reply_text("请使用 /report_service 服务单ID 举报原因。")
    order_id = message.command[1]
    try:
        await get_order_for_actor(order_id, message.from_user.id)
        report = await create_service_report(
            reporter_id=message.from_user.id,
            target_type="order",
            target_id=order_id,
            reason=" ".join(message.command[2:]),
        )
        await message.reply_text(f"举报 {report['report_id'][:8]} 已提交后台处理。")
    except ServiceDataError as exc:
        await message.reply_text(str(exc))


@app.on_message(filters.private & filters.location, group=9)
async def service_location_message(_, message: Message):
    if not message.from_user or not message.location:
        return
    flow = _current_flow(message.from_user.id)
    if not flow or flow["stage"] not in {"search_location", "request_location"}:
        return
    longitude = message.location.longitude
    latitude = message.location.latitude
    if flow["stage"] == "search_location":
        await _show_nearby(message, longitude=longitude, latitude=latitude)
    else:
        flow.update(
            {
                "longitude": longitude,
                "latitude": latitude,
                "stage": "request_address",
            }
        )
        await message.reply_text("请填写上门精确地址（门牌号/楼层/房间）。该文本会加密保存。")
    raise StopPropagation


@app.on_message(filters.private & filters.command("cancel"), group=-30)
async def cancel_service_flow(_, message: Message):
    if not message.from_user or not _flows.pop(message.from_user.id, None):
        return
    await message.reply_text("已取消当前技师服务操作。")
    raise StopPropagation


@app.on_message(filters.private & filters.text, group=9)
async def service_private_text(_, message: Message):
    if not message.from_user:
        return
    flow = _current_flow(message.from_user.id)
    if not flow:
        return
    text = str(message.text or "").strip()
    stage = flow["stage"]
    try:
        if stage == "request_address":
            if len(text) < 3:
                return await message.reply_text("请填写完整精确地址。")
            flow.update({"address_text": text, "stage": "request_requirements"})
            await message.reply_text("请描述服务需求和补充说明。")
        elif stage == "request_requirements":
            flow.update({"requirements": text, "stage": "request_preview"})
            await _preview_request(message, flow)
        elif stage == "quote_base":
            flow["quote"]["base_amount"] = text
            flow["stage"] = "quote_travel"
            await message.reply_text("第 2/5 步：请输入交通费；没有则输入 0。")
        elif stage == "quote_travel":
            flow["quote"]["travel_fee"] = text
            flow["stage"] = "quote_addons"
            await message.reply_text(
                "第 3/5 步：输入加项“名称|数量|单价”，多个用分号分隔；没有则输入“跳过”。"
            )
        elif stage == "quote_addons":
            flow["quote"]["addons"] = _parse_addons(text)
            flow["stage"] = "quote_discount"
            await message.reply_text("第 4/5 步：请输入优惠金额；没有则输入 0。")
        elif stage == "quote_discount":
            flow["quote"]["discount"] = text
            flow["stage"] = "quote_note"
            await message.reply_text("第 5/5 步：请输入报价说明；没有则输入“无”。")
        elif stage == "quote_note":
            flow["quote"]["note"] = "" if text == "无" else text
            await _preview_quote(message, flow)
        elif stage in {"cancel_reason", "dispute_reason", "reject_reason"}:
            order_id = flow["order_id"]
            if stage == "cancel_reason":
                order = await cancel_service_order(order_id, message.from_user.id, text)
            elif stage == "dispute_reason":
                order = await dispute_service_order(order_id, message.from_user.id, text)
            else:
                order = await reject_service_order(order_id, message.from_user.id, text)
            _flows.pop(message.from_user.id, None)
            other_id = (
                order["technician_id"]
                if message.from_user.id == order["customer_id"]
                else order["customer_id"]
            )
            await message.reply_text(f"服务单已更新为：{ORDER_LABELS[order['status']]}。")
            await _notify(other_id, f"服务单 {order_id[:8]} 状态已更新：{ORDER_LABELS[order['status']]}。原因：{text}")
        elif stage == "review_text":
            question = flow["template"]["questions"][flow["question_index"]]
            if text != "跳过" or question.get("required"):
                flow["answers"][question["question_id"]] = "" if text == "跳过" else text
            flow["question_index"] += 1
            await save_review_draft_progress(
                message.from_user.id,
                answers=flow["answers"],
                question_index=flow["question_index"],
            )
            await _ask_review_question(message)
        elif stage == "template_edit_name":
            flow["package"]["name"] = text
            flow["package"]["customized_from_template"] = True
            await _show_template_package_preview(message, flow)
        elif stage == "template_edit_category":
            flow["package"]["category"] = text
            flow["package"]["customized_from_template"] = True
            await _show_template_package_preview(message, flow)
        elif stage == "template_edit_description":
            flow["package"]["description"] = text
            flow["package"]["customized_from_template"] = True
            await _show_template_package_preview(message, flow)
        elif stage == "template_edit_tags":
            flow["package"]["tags"] = text
            flow["package"]["customized_from_template"] = True
            await _show_template_package_preview(message, flow)
        elif stage == "template_edit_modes":
            modes = {
                "到店": ["at_store"],
                "上门": ["onsite"],
                "两者": ["at_store", "onsite"],
            }.get(text)
            if not modes:
                return await message.reply_text("请输入“到店”、“上门”或“两者”。")
            flow["package"]["service_modes"] = modes
            if "onsite" in modes and not flow["package"].get("service_radius_km"):
                flow["package"]["service_radius_km"] = 20
            flow["package"]["customized_from_template"] = True
            await _show_template_package_preview(message, flow)
        elif stage == "template_edit_price":
            values = [item.strip() for item in text.split("|")]
            if len(values) != 3:
                return await message.reply_text("格式为：价格模式|最低价|最高价。")
            price_mode = PRICE_MODE_INPUTS.get(values[0])
            if not price_mode:
                return await message.reply_text(
                    "价格模式请选择：固定价、起步价、价格区间或面议。"
                )
            flow["package"].update(
                {
                    "price_mode": price_mode,
                    "min_price": values[1],
                    "max_price": values[2],
                    "customized_from_template": True,
                }
            )
            await _show_template_package_preview(message, flow)
        elif stage == "template_edit_billing":
            values = [item.strip() for item in text.split("|")]
            if len(values) != 2:
                return await message.reply_text("格式为：ISO币种|计价单位。")
            price_unit = PRICE_UNIT_INPUTS.get(values[1])
            if not price_unit:
                return await message.reply_text(
                    "计价单位请选择：每次、每小时、每件或每次上门。"
                )
            flow["package"].update(
                {
                    "currency": values[0],
                    "price_unit": price_unit,
                    "customized_from_template": True,
                }
            )
            await _show_template_package_preview(message, flow)
        elif stage == "template_edit_duration":
            values = [item.strip() for item in text.split("|")]
            if len(values) != 2:
                return await message.reply_text("格式为：时长分钟|上门半径公里。")
            flow["package"].update(
                {
                    "duration_minutes": int(values[0]),
                    "service_radius_km": float(values[1]),
                    "customized_from_template": True,
                }
            )
            await _show_template_package_preview(message, flow)
        elif stage == "template_edit_details":
            values = [item.strip() for item in text.split("|")]
            if len(values) != 7:
                return await message.reply_text("请按 7 个字段输入，字段内不要使用 |。")
            flow["package"].update(
                {
                    "included_items": values[0],
                    "excluded_items": values[1],
                    "preparation": values[2],
                    "out_of_range_policy": values[3],
                    "travel_fee": {
                        "mode": values[4],
                        "amount": values[5],
                        "per_km": values[6],
                    },
                    "customized_from_template": True,
                }
            )
            await _show_template_package_preview(message, flow)
        elif stage == "template_edit_addons":
            addons = []
            if text not in {"无", "清空", "0"}:
                for raw in text.replace("；", ";").split(";"):
                    values = [item.strip() for item in raw.split("|")]
                    if len(values) != 4:
                        return await message.reply_text("加项格式为：名称|单位|价格|额外分钟。")
                    addons.append(
                        {
                            "name": values[0],
                            "unit": values[1],
                            "price": values[2],
                            "extra_minutes": int(values[3]),
                        }
                    )
            flow["package"]["addons"] = addons
            flow["package"]["customized_from_template"] = True
            await _show_template_package_preview(message, flow)
        elif stage == "custom_name":
            flow["package"]["name"] = text
            flow["stage"] = "custom_category"
            await message.reply_text("自建套餐 2/9：请输入服务分类。")
        elif stage == "custom_category":
            flow["package"]["category"] = text
            flow["stage"] = "custom_description"
            await message.reply_text("自建套餐 3/9：请输入套餐详细说明。")
        elif stage == "custom_description":
            flow["package"]["description"] = text
            flow["stage"] = "custom_tags"
            await message.reply_text("自建套餐 4/9：请输入标签，多个用逗号分隔。")
        elif stage == "custom_tags":
            flow["package"]["tags"] = text
            flow["stage"] = "custom_modes"
            await message.reply_text("自建套餐 5/9：输入“到店”、“上门”或“两者”。")
        elif stage == "custom_modes":
            modes = {
                "到店": ["at_store"],
                "上门": ["onsite"],
                "两者": ["at_store", "onsite"],
            }.get(text)
            if not modes:
                return await message.reply_text("请输入“到店”、“上门”或“两者”。")
            flow["package"]["service_modes"] = modes
            flow["stage"] = "custom_price"
            await message.reply_text(
                "自建套餐 6/9：按“价格模式|币种|最低价|最高价|计价单位”输入。\n"
                "价格模式：固定价/起步价/价格区间/面议；"
                "计价单位：每次/每小时/每件/每次上门。\n"
                "示例：固定价|CNY|200|200|每次"
            )
        elif stage == "custom_price":
            values = [item.strip() for item in text.split("|")]
            if len(values) != 5:
                return await message.reply_text("格式不正确，请按 5 个字段重新输入。")
            raw_price_mode, currency, min_price, max_price, raw_price_unit = values
            price_mode = PRICE_MODE_INPUTS.get(raw_price_mode)
            price_unit = PRICE_UNIT_INPUTS.get(raw_price_unit)
            if not price_mode:
                return await message.reply_text(
                    "价格模式不支持，请填写：固定价、起步价、价格区间或面议。"
                )
            if not price_unit:
                return await message.reply_text(
                    "计价单位不支持，请填写：每次、每小时、每件或每次上门。"
                )
            flow["package"].update(
                {
                    "price_mode": price_mode,
                    "currency": currency,
                    "min_price": min_price,
                    "max_price": max_price,
                    "price_unit": price_unit,
                }
            )
            flow["stage"] = "custom_duration"
            await message.reply_text("自建套餐 7/9：输入“默认时长分钟|上门半径公里”；不支持上门时半径填 0。")
        elif stage == "custom_duration":
            values = [item.strip() for item in text.split("|")]
            if len(values) != 2:
                return await message.reply_text("格式不正确，例如：60|20。")
            flow["package"]["duration_minutes"] = int(values[0])
            flow["package"]["service_radius_km"] = (
                float(values[1]) if "onsite" in flow["package"]["service_modes"] else None
            )
            flow["stage"] = "custom_details"
            await message.reply_text("自建套餐 8/9：输入“包含项|不包含项|准备事项|超区说明”。")
        elif stage == "custom_details":
            values = [item.strip() for item in text.split("|")]
            if len(values) != 4:
                return await message.reply_text("格式不正确，请按 4 个字段重新输入。")
            flow["package"].update(
                {
                    "included_items": values[0],
                    "excluded_items": values[1],
                    "preparation": values[2],
                    "out_of_range_policy": values[3],
                }
            )
            flow["stage"] = "custom_travel"
            await message.reply_text(
                "自建套餐 9/9：输入“交通费模式|固定金额|每公里金额”。"
                "模式为已包含/固定交通费/按公里/报价时确认，"
                "例如：报价时确认|0|0。"
            )
        elif stage == "custom_travel":
            values = [item.strip() for item in text.split("|")]
            if len(values) != 3:
                return await message.reply_text("格式不正确，请按 3 个字段重新输入。")
            travel_mode = TRAVEL_FEE_INPUTS.get(values[0])
            if not travel_mode:
                return await message.reply_text(
                    "交通费请选择：已包含、固定交通费、按公里或报价时确认。"
                )
            flow["package"].update(
                {
                    "travel_fee": {
                        "mode": travel_mode,
                        "amount": values[1],
                        "per_km": values[2],
                    },
                    "addons": [],
                    "source_type": "custom",
                }
            )
            if flow.get("profile_mode") == "append":
                await _publish_profile(message, flow)
            else:
                flow["stage"] = "profile_headline"
                await message.reply_text("套餐信息已完成。服务资料第 1/7 步：请输入一句服务介绍（最多 80 字）。")
        elif stage.startswith("profile_"):
            profile = flow["profile"]
            next_steps = {
                "profile_headline": ("headline", "profile_bio", "第 2/7 步：请输入完整服务介绍。"),
                "profile_bio": ("bio", "profile_hours", "第 3/7 步：请输入可联系时段。"),
                "profile_hours": ("contact_hours", "profile_area", "第 4/7 步：请输入公开服务区域描述。"),
                "profile_area": ("public_area_text", "profile_venue", "第 5/7 步：请输入门店/工作地点名称。"),
                "profile_venue": ("venue_name", "profile_address", "第 6/7 步：请输入到店精确地址。"),
                "profile_address": ("venue_address", "profile_onsite", "第 7/7 步：请输入上门范围与超区说明。"),
            }
            if stage in next_steps:
                key, next_stage, prompt = next_steps[stage]
                profile[key] = text
                flow["stage"] = next_stage
                await message.reply_text(prompt)
            elif stage == "profile_onsite":
                profile["onsite_description"] = text
                await _publish_profile(message, flow)
        else:
            return
    except (ServiceDataError, ValueError) as exc:
        await message.reply_text(f"操作未完成：{exc}")
    raise StopPropagation


@app.on_callback_query(filters.regex(r"^svc:"))
async def service_callback(_, query: CallbackQuery):
    parts = str(query.data).split(":")
    try:
        action = parts[1]
        if action == "terms":
            await observe_customer(query.from_user, accepted_terms=True)
            await query.answer("已接受。")
            return await send_service_home(query)
        if action == "home":
            return await send_service_home(query)
        if action == "find":
            return await _ask_search_location(query)
        if action == "near":
            return await _show_nearby(
                query,
                longitude=None,
                latitude=None,
                page=int(parts[2]),
            )
        if action == "tech":
            return await _show_technician(query, int(parts[2]))
        if action == "req":
            return await _choose_request_mode(query, int(parts[2]), parts[3])
        if action == "mode":
            return await _request_mode(query, int(parts[2]), parts[3], parts[4])
        if action == "request_submit":
            flow = _current_flow(query.from_user.id)
            if not flow or flow["stage"] != "request_preview":
                raise ServiceDataError("request_flow_expired", "服务请求已过期，请重新发起。")
            order = await create_service_order(customer=query.from_user, **{
                key: flow.get(key)
                for key in (
                    "technician_id",
                    "package_id",
                    "service_mode",
                    "requirements",
                    "longitude",
                    "latitude",
                    "address_text",
                )
            })
            _flows.pop(query.from_user.id, None)
            await query.message.reply_text(
                f"服务请求 {order['order_id'][:8]} 已提交，等待技师报价。",
                reply_markup=_order_markup(order, query.from_user.id),
            )
            await _notify(
                order["technician_id"],
                f"收到新的服务请求 {order['order_id'][:8]}。",
                reply_markup=_order_markup(order, order["technician_id"]),
            )
            return await query.answer("已提交。")
        if action == "flow_cancel":
            _flows.pop(query.from_user.id, None)
            await query.message.reply_text("已取消当前操作。")
            return await query.answer()
        if action == "orders":
            return await _show_orders(query, role=parts[2], page=int(parts[3]))
        if action == "order":
            return await _show_order(query, parts[2])
        if action == "quote":
            await _start_quote(query, parts[2])
            return await query.answer()
        if action == "quote_submit":
            flow = _current_flow(query.from_user.id)
            if not flow or flow["stage"] != "quote_preview":
                raise ServiceDataError("quote_flow_expired", "报价流程已过期，请重新发起。")
            order, _ = await submit_service_quote(
                flow["order"]["order_id"], query.from_user.id, flow["normalized_quote"]
            )
            _flows.pop(query.from_user.id, None)
            await query.message.reply_text("最终报价已提交，等待顾客确认。")
            await _notify(order["customer_id"], f"服务单 {order['order_id'][:8]} 已收到最终报价。", reply_markup=_order_markup(order, order["customer_id"]))
            return await query.answer("已提交。")
        if action in {"cancel", "dispute", "reject"}:
            stage = {"cancel": "cancel_reason", "dispute": "dispute_reason", "reject": "reject_reason"}[action]
            _flow(query.from_user.id, stage, order_id=parts[2])
            await query.message.reply_text("请填写原因。该原因会通知对方并保留在状态轨迹中。")
            return await query.answer()
        if action == "confirm":
            order = await confirm_service_quote(parts[2], query.from_user.id)
            await query.message.reply_text("最终报价已确认。精确履约地址和双方联系入口现已解锁。")
            await _notify(order["technician_id"], f"顾客已确认服务单 {order['order_id'][:8]} 的最终报价。", reply_markup=_order_markup(order, order["technician_id"]))
            return await _show_order(query, order["order_id"])
        if action == "start":
            order = await start_service_order(parts[2], query.from_user.id)
            await _notify(order["customer_id"], f"技师已开始服务单 {order['order_id'][:8]}。")
            return await _show_order(query, order["order_id"])
        if action == "finish":
            await _issue_qr(query, parts[2])
            return await query.answer("二维码已生成。")
        if action == "claim":
            record, _ = await claim_review_qr(parts[2], query.from_user.id)
            await query.answer("服务已确认完成。")
            return await _begin_review(
                query,
                technician_id=record["technician_id"],
                source="qr_verified",
                qr_id=record["qr_id"],
            )
        if action in {"review", "review_start"}:
            if action == "review_start":
                return await query.message.reply_text("请从技师详情点击“主动评价”，或使用 /review_teacher 技师ID。")
            return await _choose_review_package(query, int(parts[2]))
        if action == "reviewpkg":
            return await _begin_review(
                query,
                technician_id=int(parts[2]),
                source="student_initiated",
                package_id=parts[3],
            )
        if action in {"single", "rate"}:
            flow = _current_flow(query.from_user.id)
            if not flow or flow["stage"] != "review_question":
                raise ServiceDataError("review_flow_expired", "评价流程已过期。")
            question = flow["template"]["questions"][flow["question_index"]]
            if question["type"] != "single_choice":
                raise ServiceDataError("invalid_review_answer", "当前问题不是单选题。")
            if action == "rate":
                option = next(
                    (
                        item
                        for item in question["options"]
                        if int(item.get("score") or 0) == int(parts[2])
                    ),
                    None,
                )
            else:
                option_index = int(parts[2])
                option = (
                    question["options"][option_index]
                    if 0 <= option_index < len(question["options"])
                    else None
                )
            if not option:
                raise ServiceDataError("invalid_review_answer", "评价选项无效。")
            flow["answers"][question["question_id"]] = option["value"]
            flow["question_index"] += 1
            await save_review_draft_progress(
                query.from_user.id,
                answers=flow["answers"],
                question_index=flow["question_index"],
            )
            await query.answer(f"已选择：{option['label']}")
            return await _ask_review_question(query)
        if action == "multi":
            flow = _current_flow(query.from_user.id)
            if not flow or flow["stage"] != "review_question":
                raise ServiceDataError("review_flow_expired", "评价流程已过期。")
            question = flow["template"]["questions"][flow["question_index"]]
            if question["type"] != "multiple_choice":
                raise ServiceDataError("invalid_review_answer", "当前问题不是多选题。")
            option_index = int(parts[2])
            if not 0 <= option_index < len(question["options"]):
                raise ServiceDataError("invalid_review_answer", "评价选项无效。")
            option_value = str(question["options"][option_index]["value"])
            selected = list(flow["answers"].get(question["question_id"]) or [])
            if option_value in selected:
                selected.remove(option_value)
            else:
                selected.append(option_value)
            flow["answers"][question["question_id"]] = selected
            await save_review_draft_progress(
                query.from_user.id,
                answers=flow["answers"],
                question_index=flow["question_index"],
            )
            await query.message.edit_reply_markup(
                _review_options_markup(question, selected)
            )
            return await query.answer("已更新选择。")
        if action == "multi_done":
            flow = _current_flow(query.from_user.id)
            if not flow or flow["stage"] != "review_question":
                raise ServiceDataError("review_flow_expired", "评价流程已过期。")
            question = flow["template"]["questions"][flow["question_index"]]
            selected = flow["answers"].get(question["question_id"]) or []
            if question.get("required") and not selected:
                return await query.answer("请至少选择一项。", show_alert=True)
            flow["question_index"] += 1
            await save_review_draft_progress(
                query.from_user.id,
                answers=flow["answers"],
                question_index=flow["question_index"],
            )
            await query.answer("已完成选择。")
            return await _ask_review_question(query)
        if action == "anonymous":
            await _submit_pending_review(query, parts[2] == "1")
            return await query.answer("评价已提交审核。")
        if action == "top":
            return await _show_top(query, parts[2], int(parts[3]))
        if action == "profile_template":
            await _start_profile_flow(query, parts[2])
            return await query.answer()
        if action == "profile_template_use":
            flow = _current_flow(query.from_user.id)
            if not flow or flow["stage"] != "template_preview":
                raise ServiceDataError("profile_flow_expired", "套餐配置流程已过期。")
            if flow.get("profile_mode") == "append":
                await _publish_profile(query, flow)
                return await query.answer("套餐已新增。")
            flow["stage"] = "profile_headline"
            await query.message.reply_text("服务资料第 1/7 步：请输入一句服务介绍（最多 80 字）。")
            return await query.answer()
        if action == "profile_edit":
            flow = _current_flow(query.from_user.id)
            if not flow or flow["stage"] != "template_preview":
                raise ServiceDataError("profile_flow_expired", "套餐配置流程已过期。")
            field = parts[2]
            prompts = {
                "name": "请输入新的套餐名称。",
                "category": "请输入新的服务分类。",
                "description": "请输入新的套餐完整说明。",
                "tags": "请输入套餐标签，多个用逗号分隔。",
                "modes": "请输入“到店”、“上门”或“两者”。",
                "price": "输入“价格模式|最低价|最高价”，例如 固定价|200|200。",
                "billing": "输入“ISO币种|计价单位”，例如 CNY|每次。",
                "duration": "输入“时长分钟|上门半径公里”，例如 60|20。",
                "details": "输入“包含项|不包含项|准备事项|超区说明|交通模式|固定金额|每公里金额”。",
                "addons": "输入“名称|单位|价格|额外分钟”，多个用分号分隔；输入“清空”可移除全部。",
            }
            if field not in prompts:
                raise ServiceDataError("invalid_profile_field", "不支持该套餐字段。")
            flow["stage"] = f"template_edit_{field}"
            await query.message.reply_text(prompts[field])
            return await query.answer()
        if action == "profile_custom":
            await _start_custom_package_flow(query)
            return await query.answer()
        if action == "profile_add":
            profile = await get_technician_service_profile(query.from_user.id)
            if len((profile.get("service_profile") or {}).get("packages", [])) >= 5:
                raise ServiceDataError("invalid_package_count", "每位技师最多发布 5 个套餐。")
            _flow(query.from_user.id, "profile_source", profile_mode="append")
            await _show_profile_sources(query)
            return await query.answer()
        if action == "profile_reconfigure":
            _flow(query.from_user.id, "profile_source", profile_mode="replace")
            await _show_profile_sources(query)
            return await query.answer()
        if action == "profile_toggle":
            profile = await get_technician_service_profile(query.from_user.id)
            values = dict(profile["service_profile"])
            values["accepting_requests"] = not values.get("accepting_requests", True)
            await publish_technician_profile(
                query.from_user.id, values, actor_id=query.from_user.id
            )
            await query.answer("接单状态已更新。")
            return await _show_profile(query)
    except (ServiceDataError, ValueError, IndexError) as exc:
        return await query.answer(str(exc), show_alert=True)
    except Exception as exc:
        log.error(f"技师服务回调失败：{type(exc).__name__}")
        return await query.answer("操作失败，请稍后重试。", show_alert=True)


start_address_redaction_sweeper()