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

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

from wbb import BOT_USERNAME, app, log
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_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 - 查找附近技师并发起服务请求。
/my_services - 查看我的服务单。
/technician_profile - 管理技师服务资料和套餐。
/review_teacher - 主动评价技师。
/technician_top - 查看审核评价排行。
/review_qr - 为服务中的订单生成完成评价二维码。
/block_customer - 技师拉黑曾发起请求的顾客。
/report_service - 举报服务单。
旧 /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": "最终报价确认",
}
_flows: dict[int, dict[str, Any]] = {}
_address_redaction_task: asyncio.Task[None] | None = None


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"
        "平台只记录服务请求、报价、状态与评价，不处理支付和退款。"
        "时间由双方私聊协商。精确地址仅在报价确认后向履约所需一方开放，"
        "订单关闭 7 天后自动脱敏。\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:orders:all:1"),
                    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
    _flow(target.from_user.id, "search_location")
    await _send_message(
        target,
        "请发送当前位置用于计算距离。该位置只保存在当前交互内存中，不会作为精确地址入库。",
        reply_markup=_location_keyboard("发送当前位置"),
    )


async def _show_nearby(
    target: Message | CallbackQuery,
    *,
    longitude: float,
    latitude: float,
    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:
        modes = " / ".join(MODE_LABELS.get(mode, mode) for mode in item["service_modes"])
        lines.append(
            f"\n<b>{escape(item.get('display_name') or str(item['user_id']))}</b> · "
            f"{escape(item['distance_band'])}\n"
            f"{escape(item.get('public_area_text') or '区域由技师填写')} · {escape(modes)}\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}")])
    flow = _current_flow(target.from_user.id)
    if flow is not None:
        flow.update({"longitude": longitude, "latitude": latitude, "stage": "search_results"})
    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 = []
    for package in service_profile.get("packages", []):
        modes = "/".join(MODE_LABELS.get(mode, mode) for mode in package["service_modes"])
        package_lines = [
            f"\n<b>{escape(package['name'])}</b> · {escape(modes)}\n"
            f"{escape(_short(package['description'], 100))}\n"
            f"参考价：{escape(_package_price_text(package))} · "
            f"约 {package.get('duration_minutes') or '未设'} 分钟",
            f"交通费：{escape(_travel_fee_text(package))}",
        ]
        if "onsite" in package["service_modes"]:
            package_lines.append(
                f"上门范围：{package.get('service_radius_km') or '-'} 公里；"
                f"{escape(_short(package.get('out_of_range_policy'), 60))}"
            )
        if package.get("included_items"):
            package_lines.append(f"包含：{escape(_short(package['included_items'], 50))}")
        if package.get("excluded_items"):
            package_lines.append(f"不含：{escape(_short(package['excluded_items'], 40))}")
        addons = package.get("addons") or []
        if addons:
            addon_text = "、".join(
                f"{item['name']} {package['currency']} {item['price']}/{item['unit']}"
                for item in addons[:5]
            )
            package_lines.append(f"标准加项：{escape(_short(addon_text, 70))}")
        if package.get("preparation"):
            package_lines.append(f"准备：{escape(_short(package['preparation'], 40))}")
        lines.append("\n".join(package_lines))
        rows.append(
            [
                InlineKeyboardButton(
                    f"发起「{package['name'][:12]}」请求",
                    callback_data=f"svc:req:{technician_id}:{package['package_id']}",
                )
            ]
        )
    rows.append(
        [
            InlineKeyboardButton(
                "主动评价该技师", callback_data=f"svc:review:{technician_id}"
            ),
            InlineKeyboardButton(
                "Telegram 私聊（次入口）",
                url=f"https://t.me/{profile['username']}",
            ),
        ]
    )
    public_reviews, _ = await list_public_reviews(technician_id, page_size=3)
    if public_reviews:
        lines.append("\n<b>最近审核评价</b>")
        for review in public_reviews:
            comment = "；".join(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_rating",
        technician_id=technician_id,
        source=source,
        package_id=package_id,
        qr_id=qr_id,
        template=draft["template_snapshot"],
        rating_index=int(draft.get("rating_index") or 0),
        text_index=int(draft.get("text_index") or 0),
        answers=dict(draft.get("answers") or {}),
    )
    await _ask_rating(target)


async def _ask_rating(target: Message | CallbackQuery) -> None:
    flow = _current_flow(target.from_user.id)
    if not flow:
        return
    ratings = flow["template"]["rating_questions"]
    index = flow["rating_index"]
    if index >= len(ratings):
        return await _ask_review_text(target)
    question = ratings[index]
    await _send_message(
        target,
        f"评价 {index + 1}/{len(ratings)}：<b>{escape(question['label'])}</b>\n"
        f"{escape(question.get('description') or '')}",
        parse_mode=ParseMode.HTML,
        reply_markup=InlineKeyboardMarkup(
            [[InlineKeyboardButton(f"{score} 分", callback_data=f"svc:rate:{score}") for score in range(1, 6)]]
        ),
    )


async def _ask_review_text(target: Message | CallbackQuery) -> None:
    flow = _current_flow(target.from_user.id)
    if not flow:
        return
    texts = flow["template"]["text_questions"]
    index = flow["text_index"]
    if index >= len(texts):
        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 = texts[index]
    flow["stage"] = "review_text"
    await _send_message(
        target,
        f"文字评价 {index + 1}/{len(texts)}：{question['label']}\n"
        f"最多 {question['max_length']} 字。"
        + ("" if question.get("required") else " 可发送“跳过”。"),
    )


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 _send_message(
        target,
        f"请选择要评价的 {profile.get('display_name') or technician_id} 套餐。主动评价无需服务单，可重复提交，审核通过后同等计分。",
        reply_markup=InlineKeyboardMarkup(
            [
                [
                    InlineKeyboardButton(
                        package["name"],
                        callback_data=f"svc:reviewpkg:{technician_id}:{package['package_id']}",
                    )
                ]
                for package in packages
            ]
        ),
    )


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>",
        "主动评价与完成服务单评价审核通过后同等计分；至少 3 条批准评价入榜。",
        "公式：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暂时没有满足 3 条审核评价门槛的技师。")
    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, "只有已通过管理员认证的技师可以发布服务资料。")
    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 = [
            [
                InlineKeyboardButton(
                    "暂停接单" if service_profile.get("accepting_requests") else "开始接单",
                    callback_data="svc:profile_toggle",
                )
            ]
        ]
        if len(package_items) < 5:
            rows.append(
                [InlineKeyboardButton("新增套餐", callback_data="svc:profile_add")]
            )
        rows.append(
            [
                InlineKeyboardButton(
                    "重建资料与套餐", 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),
        )
    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
    )
    _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"]["text_questions"][flow["text_index"]]
            if text != "跳过" or question.get("required"):
                flow["answers"][question["question_id"]] = "" if text == "跳过" else text
            flow["text_index"] += 1
            await save_review_draft_progress(
                message.from_user.id,
                answers=flow["answers"],
                rating_index=flow["rating_index"],
                text_index=flow["text_index"],
            )
            await _ask_review_text(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("格式为：价格模式|最低价|最高价。")
            flow["package"].update(
                {
                    "price_mode": values[0],
                    "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币种|计价单位。")
            flow["package"].update(
                {
                    "currency": values[0],
                    "price_unit": values[1],
                    "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"
                "价格模式：fixed/starting_at/range/negotiable；计价单位：per_service/per_hour/per_item/per_visit。\n"
                "示例：fixed|CNY|200|200|per_service"
            )
        elif stage == "custom_price":
            values = [item.strip() for item in text.split("|")]
            if len(values) != 5:
                return await message.reply_text("格式不正确，请按 5 个字段重新输入。")
            price_mode, currency, min_price, max_price, price_unit = values
            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：输入“交通费模式|固定金额|每公里金额”。"
                "模式为 included/fixed/per_km/quoted，例如：quoted|0|0。"
            )
        elif stage == "custom_travel":
            values = [item.strip() for item in text.split("|")]
            if len(values) != 3:
                return await message.reply_text("格式不正确，请按 3 个字段重新输入。")
            flow["package"].update(
                {
                    "travel_fee": {
                        "mode": values[0],
                        "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":
            flow = _current_flow(query.from_user.id)
            if not flow or "longitude" not in flow:
                return await query.answer("位置已过期，请重新发送。", show_alert=True)
            return await _show_nearby(
                query,
                longitude=flow["longitude"],
                latitude=flow["latitude"],
                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 == "rate":
            flow = _current_flow(query.from_user.id)
            if not flow or flow["stage"] != "review_rating":
                raise ServiceDataError("review_flow_expired", "评价流程已过期。")
            question = flow["template"]["rating_questions"][flow["rating_index"]]
            flow["answers"][question["question_id"]] = int(parts[2])
            flow["rating_index"] += 1
            await save_review_draft_progress(
                query.from_user.id,
                answers=flow["answers"],
                rating_index=flow["rating_index"],
                text_index=flow["text_index"],
            )
            await query.answer(f"已选择 {parts[2]} 分。")
            return await _ask_rating(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": "输入“价格模式|最低价|最高价”，例如 fixed|200|200。",
                "billing": "输入“ISO币种|计价单位”，例如 CNY|per_service。",
                "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()