from __future__ import annotations import hashlib import json import math import re from collections import defaultdict from datetime import UTC, datetime, timedelta from decimal import Decimal, InvalidOperation from secrets import token_urlsafe from typing import Any from uuid import uuid4 from cryptography.fernet import Fernet, InvalidToken from pymongo import ASCENDING, DESCENDING, ReturnDocument from pymongo.errors import DuplicateKeyError import wbb from wbb import control_db templatesdb = control_db.service_package_templates ordersdb = control_db.service_orders quotesdb = control_db.service_order_quotes addressesdb = control_db.service_order_addresses eventsdb = control_db.service_order_events customersdb = control_db.customer_profiles blocksdb = control_db.customer_blocks reportsdb = control_db.service_reports qrdb = control_db.teacher_review_qr_records reviewtemplatesdb = control_db.teacher_review_templates reviewdraftsdb = control_db.teacher_review_drafts reviewsdb = control_db.teacher_reviews statsdb = control_db.teacher_review_stats reviewtopicsdb = control_db.teacher_review_topics profilesdb = control_db.directory_profiles locationsdb = control_db.directory_locations membershipsdb = control_db.directory_memberships settingsdb = control_db.directory_settings SERVICE_MODES = {"at_store", "onsite"} PRICE_MODES = {"fixed", "starting_at", "range", "negotiable"} PRICE_UNITS = {"per_service", "per_hour", "per_item", "per_visit"} TRAVEL_FEE_MODES = {"included", "fixed", "per_km", "quoted"} ORDER_ACTIVE_STATUSES = { "requested", "quoted", "confirmed", "in_progress", "completion_pending", "disputed", } ORDER_TERMINAL_STATUSES = { "completed", "rejected", "canceled_customer", "canceled_technician", "voided", } ORDER_STATUSES = ORDER_ACTIVE_STATUSES | ORDER_TERMINAL_STATUSES REVIEW_SOURCES = {"qr_verified", "student_initiated"} REVIEW_STATUSES = {"pending", "approved", "rejected", "withdrawn", "voided"} DEFAULT_REVIEW_TEMPLATE = { "questions": [ { "question_id": "overall_experience", "type": "single_choice", "label": "这次服务体验怎么样？", "description": "选择最符合实际体验的一项。", "required": True, "weight": 100, "options": [ {"value": "excellent", "label": "非常满意", "score": 5}, {"value": "good", "label": "满意", "score": 4}, {"value": "average", "label": "一般", "score": 3}, {"value": "poor", "label": "不满意", "score": 2}, {"value": "bad", "label": "非常不满意", "score": 1}, ], }, { "question_id": "service_highlights", "type": "multiple_choice", "label": "哪些方面值得肯定？", "description": "可多选，也可以跳过。", "required": False, "options": [ {"value": "professional", "label": "专业可靠"}, {"value": "communication", "label": "沟通顺畅"}, {"value": "careful", "label": "服务细致"}, {"value": "transparent", "label": "价格透明"}, {"value": "punctual", "label": "准时守约"}, ], }, { "question_id": "comment", "type": "text", "label": "补充评价", "description": "可选，请勿填写联系方式或精确地址。", "required": False, "max_length": 500, }, ] } BUILTIN_PACKAGE_TEMPLATES = ( { "template_id": "builtin_quick_at_store", "name": "快速到店服务", "sort_order": -300, "package": { "package_id": "builtin_package_at_store", "name": "标准到店服务", "category": "通用服务", "description": "顾客到指定区域接受一次标准服务，具体地点和服务细节通过 Telegram 私聊沟通。", "tags": ["到店", "标准服务"], "service_modes": ["at_store"], "price_mode": "fixed", "currency": "CNY", "price_unit": "per_service", "min_price": "200", "max_price": "200", "duration_minutes": 60, "included_items": "一次标准服务", "excluded_items": "额外耗材和临时加项", "preparation": "请提前说明具体需求", "addons": [], "travel_fee": { "mode": "quoted", "amount": "0", "per_km": "0", "description": "无额外交通费；其他临时费用请提前沟通。", }, "service_radius_km": None, "out_of_range_policy": "", }, }, { "template_id": "builtin_quick_onsite", "name": "快速上门服务", "sort_order": -200, "package": { "package_id": "builtin_package_onsite", "name": "标准上门服务", "category": "通用服务", "description": "技师提供一次标准上门服务，具体地点和服务细节通过 Telegram 私聊沟通。", "tags": ["上门", "标准服务"], "service_modes": ["onsite"], "price_mode": "starting_at", "currency": "CNY", "price_unit": "per_visit", "min_price": "300", "max_price": "300", "duration_minutes": 90, "included_items": "一次标准上门服务", "excluded_items": "交通费、额外耗材和临时加项", "preparation": "请提前发送位置并说明具体需求", "addons": [], "travel_fee": { "mode": "quoted", "amount": "0", "per_km": "0", "description": "上门车费根据距离另行沟通。", }, "service_radius_km": 20, "out_of_range_policy": "超出服务半径时，请先沟通交通费和是否可以接单。", }, }, { "template_id": "builtin_quick_hourly", "name": "快速按小时服务", "sort_order": -100, "package": { "package_id": "builtin_package_hourly", "name": "按小时专业服务", "category": "通用服务", "description": "按服务时长计价，支持到店或上门，具体工作范围提前确认。", "tags": ["按小时", "到店", "上门"], "service_modes": ["at_store", "onsite"], "price_mode": "starting_at", "currency": "CNY", "price_unit": "per_hour", "min_price": "200", "max_price": "200", "duration_minutes": 60, "included_items": "一小时专业服务", "excluded_items": "交通费、额外耗材和超时服务", "preparation": "请提前说明服务内容和预计时长", "addons": [], "travel_fee": { "mode": "quoted", "amount": "0", "per_km": "0", "description": "上门车费、耗材费和超时费用另行沟通。", }, "service_radius_km": 20, "out_of_range_policy": "超出服务半径时，请先沟通交通费和是否可以接单。", }, }, ) _indexes_ready = False class ServiceDataError(ValueError): def __init__(self, code: str, message: str): super().__init__(message) self.code = code def utc_now() -> datetime: return datetime.now(UTC) def _as_utc(value: datetime) -> datetime: return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC) def _clean_text(value: Any, *, max_length: int, required: bool = False) -> str: text = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", "", str(value or "")) text = " ".join(text.strip().split()) if required and not text: raise ServiceDataError("required_field", "必填内容不能为空。") if len(text) > max_length: raise ServiceDataError("text_too_long", f"内容不能超过 {max_length} 个字符。") return text def _positive_decimal(value: Any, field: str, *, allow_zero: bool = True) -> str: try: number = Decimal(str(value or 0)).quantize(Decimal("0.01")) except (InvalidOperation, ValueError) as exc: raise ServiceDataError("invalid_price", f"{field} 必须是有效金额。") from exc if not number.is_finite(): raise ServiceDataError("invalid_price", f"{field} 必须是有效金额。") if number < 0 or (not allow_zero and number == 0): raise ServiceDataError("invalid_price", f"{field} 不能小于 0。") return format(number, "f") def _normalize_tags(values: Any) -> list[str]: if isinstance(values, str): values = values.replace("，", ",").split(",") tags = [] for value in values or []: tag = _clean_text(value, max_length=20) if tag and tag.casefold() not in {item.casefold() for item in tags}: tags.append(tag) if len(tags) > 8: raise ServiceDataError("too_many_tags", "服务标签最多 8 个。") return tags def normalize_package(value: dict[str, Any], *, package_id: str | None = None) -> dict[str, Any]: modes = list( dict.fromkeys( str(item) for item in (value.get("service_modes") or ["at_store", "onsite"]) ) ) if set(modes) - SERVICE_MODES: raise ServiceDataError("invalid_service_mode", "服务方式无效。") price_mode = str(value.get("price_mode") or "negotiable") if price_mode not in PRICE_MODES: raise ServiceDataError("invalid_price_mode", "不支持该价格模式。") price_unit = str(value.get("price_unit") or "per_service") if price_unit not in PRICE_UNITS: raise ServiceDataError("invalid_price_unit", "不支持该计价单位。") currency = str(value.get("currency") or "CNY").upper() if not re.fullmatch(r"[A-Z]{3}", currency): raise ServiceDataError("invalid_currency", "币种必须使用三位 ISO 代码。") min_price = _positive_decimal(value.get("min_price"), "最低价格") max_price = _positive_decimal(value.get("max_price"), "最高价格") if Decimal(max_price) and Decimal(max_price) < Decimal(min_price): raise ServiceDataError("invalid_price_range", "最高价格不能低于最低价格。") if price_mode == "fixed" and Decimal(min_price) != Decimal(max_price): raise ServiceDataError("invalid_fixed_price", "固定价的最低价格和最高价格必须一致。") duration = value.get("duration_minutes") if duration in (None, "", 0, "0"): duration = None else: try: duration = int(duration) except (TypeError, ValueError) as exc: raise ServiceDataError("invalid_duration", "服务时长必须是分钟数。") from exc if not 15 <= duration <= 480: raise ServiceDataError("invalid_duration", "服务时长必须在 15 到 480 分钟之间。") travel = value.get("travel_fee") if isinstance(value.get("travel_fee"), dict) else {} travel_mode = str(travel.get("mode") or "quoted") if travel_mode not in TRAVEL_FEE_MODES: raise ServiceDataError("invalid_travel_fee", "不支持该交通费模式。") addons = [] for item in value.get("addons", []) or []: if len(addons) >= 10: raise ServiceDataError("too_many_addons", "标准加项最多 10 个。") addons.append( { "addon_id": str(item.get("addon_id") or uuid4().hex), "name": _clean_text(item.get("name"), max_length=40, required=True), "unit": _clean_text(item.get("unit"), max_length=20) or "项", "price": _positive_decimal(item.get("price"), "加项价格"), "extra_minutes": max(0, min(480, int(item.get("extra_minutes") or 0))), } ) radius = value.get("service_radius_km") if radius in (None, ""): radius = None else: radius = float(radius) if not 0.5 <= radius <= 200: raise ServiceDataError("invalid_service_radius", "上门半径必须在 0.5 到 200 公里之间。") out_of_range_policy = _clean_text( value.get("out_of_range_policy"), max_length=200, ) source_type = str(value.get("source_type") or "custom") if source_type not in {"template", "custom"}: raise ServiceDataError("invalid_package_source", "套餐来源无效。") return { "package_id": str(package_id or value.get("package_id") or uuid4().hex), "name": _clean_text(value.get("name"), max_length=50, required=True), "category": _clean_text(value.get("category"), max_length=30, required=True), "description": _clean_text(value.get("description"), max_length=800, required=True), "tags": _normalize_tags(value.get("tags", [])), "service_modes": modes, "price_mode": price_mode, "currency": currency, "price_unit": price_unit, "min_price": min_price, "max_price": max_price, "price_display": _clean_text(value.get("price_display"), max_length=80), "duration_minutes": duration, "included_items": _clean_text(value.get("included_items"), max_length=500), "excluded_items": _clean_text(value.get("excluded_items"), max_length=500), "preparation": _clean_text(value.get("preparation"), max_length=500), "addons": addons, "travel_fee": { "mode": travel_mode, "amount": _positive_decimal(travel.get("amount"), "交通费"), "per_km": _positive_decimal(travel.get("per_km"), "每公里交通费"), "description": _clean_text(travel.get("description"), max_length=200), }, "service_radius_km": radius, "out_of_range_policy": out_of_range_policy, "source_type": source_type, "source_template_id": value.get("source_template_id"), "source_template_version": value.get("source_template_version"), "customized_from_template": bool(value.get("customized_from_template")), } def _normalize_profile(value: dict[str, Any]) -> dict[str, Any]: packages = [normalize_package(item) for item in value.get("packages", [])] if not 1 <= len(packages) <= 5: raise ServiceDataError("invalid_package_count", "技师必须发布 1 到 5 个套餐。") modes = sorted({mode for item in packages for mode in item["service_modes"]}) venue = value.get("venue") if isinstance(value.get("venue"), dict) else {} onsite = value.get("onsite_policy") if isinstance(value.get("onsite_policy"), dict) else {} public_area_text = _clean_text( value.get("public_area_text"), max_length=80, required=True ) venue_name = _clean_text(venue.get("name"), max_length=80) venue_address = _clean_text(venue.get("address_hint"), max_length=120) onsite_description = _clean_text(onsite.get("description"), max_length=300) return { "headline": _clean_text(value.get("headline"), max_length=80, required=True), "bio": _clean_text(value.get("bio"), max_length=800, required=True), "tags": _normalize_tags(value.get("tags", [])), "contact_hours": _clean_text(value.get("contact_hours"), max_length=120) or "请通过 Telegram 私聊沟通具体服务信息", "service_modes": modes, "accepting_requests": bool(value.get("accepting_requests", True)), "public_area_text": public_area_text, "venue": { "name": venue_name, "address_hint": venue_address, }, "onsite_policy": { "description": onsite_description, }, "packages": packages, "is_complete": True, } async def ensure_service_indexes() -> None: global _indexes_ready if _indexes_ready: return await templatesdb.create_index([("template_id", ASCENDING)], unique=True) await templatesdb.create_index([("status", ASCENDING), ("sort_order", ASCENDING)]) await ordersdb.create_index([("order_id", ASCENDING)], unique=True) await ordersdb.create_index([("customer_id", ASCENDING), ("created_at", DESCENDING)]) await ordersdb.create_index([("technician_id", ASCENDING), ("created_at", DESCENDING)]) await ordersdb.create_index([("status", ASCENDING), ("updated_at", DESCENDING)]) await quotesdb.create_index([("quote_id", ASCENDING)], unique=True) await quotesdb.create_index([("order_id", ASCENDING), ("version", DESCENDING)], unique=True) await addressesdb.create_index([("order_id", ASCENDING)], unique=True) await addressesdb.create_index([("redact_after", ASCENDING)]) await eventsdb.create_index([("event_id", ASCENDING)], unique=True) await eventsdb.create_index([("order_id", ASCENDING), ("created_at", ASCENDING)]) await customersdb.create_index([("user_id", ASCENDING)], unique=True) await blocksdb.create_index([("scope", ASCENDING), ("technician_id", ASCENDING), ("customer_id", ASCENDING)], unique=True) await reportsdb.create_index([("report_id", ASCENDING)], unique=True) await qrdb.create_index([("qr_id", ASCENDING)], unique=True) await qrdb.create_index([("token_hash", ASCENDING)], unique=True) await qrdb.create_index([("order_id", ASCENDING), ("created_at", DESCENDING)]) await reviewtemplatesdb.create_index([("version", DESCENDING)], unique=True) await reviewdraftsdb.create_index([("customer_id", ASCENDING)], unique=True) await reviewdraftsdb.create_index([("expires_at", ASCENDING)], expireAfterSeconds=0) await reviewsdb.create_index([("review_id", ASCENDING)], unique=True) await reviewsdb.create_index([("qr_id", ASCENDING)], unique=True, sparse=True) await reviewsdb.create_index([("technician_id", ASCENDING), ("status", ASCENDING), ("created_at", DESCENDING)]) await statsdb.create_index([("technician_id", ASCENDING), ("category", ASCENDING)], unique=True) await reviewtopicsdb.create_index([("technician_id", ASCENDING)], unique=True) await reviewtopicsdb.create_index([("forum_chat_id", ASCENDING), ("message_thread_id", ASCENDING)]) _indexes_ready = True async def record_service_event( order_id: str, event_type: str, *, actor_id: int | str, reason: str = "", metadata: dict[str, Any] | None = None, ) -> dict[str, Any]: await ensure_service_indexes() document = { "event_id": uuid4().hex, "order_id": str(order_id), "event_type": str(event_type), "actor_id": actor_id, "reason": _clean_text(reason, max_length=500), "metadata": metadata or {}, "created_at": utc_now(), } await eventsdb.insert_one(document) return document async def record_service_funnel_event(user_id: int, event_type: str) -> dict[str, Any]: if event_type not in {"directory_viewed", "technician_viewed", "request_started"}: raise ServiceDataError("invalid_funnel_event", "不支持该履约漏斗事件。") return await record_service_event( "", event_type, actor_id=int(user_id), metadata={"funnel": True}, ) async def observe_customer(user: Any, *, accepted_terms: bool = False) -> dict[str, Any]: await ensure_service_indexes() now = utc_now() display_name = " ".join( value for value in (str(getattr(user, "first_name", "") or "").strip(), str(getattr(user, "last_name", "") or "").strip()) if value ) or f"用户 {user.id}" updates: dict[str, Any] = { "username": getattr(user, "username", None), "display_name": display_name, "updated_at": now, } if accepted_terms: updates["terms_accepted_at"] = now await customersdb.update_one( {"user_id": int(user.id)}, {"$set": updates, "$setOnInsert": {"created_at": now, "blocked": False}}, upsert=True, ) return await customersdb.find_one({"user_id": int(user.id)}) or {} async def require_customer_allowed(user_id: int) -> dict[str, Any]: await ensure_service_indexes() customer = await customersdb.find_one({"user_id": int(user_id)}) if not customer or not customer.get("terms_accepted_at"): raise ServiceDataError("terms_required", "请先接受服务规则和隐私说明。") if customer.get("blocked"): raise ServiceDataError("customer_blocked", "你的服务功能已被暂停。") global_block = await blocksdb.find_one({"scope": "global", "customer_id": int(user_id), "active": True}) if global_block: raise ServiceDataError("customer_blocked", "你的服务功能已被暂停。") return customer async def get_service_settings() -> dict[str, Any]: stored = await settingsdb.find_one({"settings_id": "global"}) or {} forum_chat_id = str( stored.get("service_review_forum_chat_id") or getattr(wbb, "SERVICE_REVIEW_FORUM_CHAT_ID", "") or "" ).strip() forum_username = str( stored.get("service_review_forum_username") or getattr(wbb, "SERVICE_REVIEW_FORUM_USERNAME", "") or "" ).strip().lstrip("@") return { "customer_max_open_orders": max( 1, min( 20, int( stored.get("service_customer_max_open_orders") or getattr(wbb, "SERVICE_CUSTOMER_MAX_OPEN_ORDERS", 3) or 3 ), ), ), "customer_daily_request_limit": max( 1, min( 100, int( stored.get("service_customer_daily_request_limit") or getattr(wbb, "SERVICE_CUSTOMER_DAILY_REQUEST_LIMIT", 10) or 10 ), ), ), "review_qr_expiry_hours": max( 1, min( 168, int( stored.get("service_review_qr_expiry_hours") or getattr(wbb, "SERVICE_REVIEW_QR_EXPIRY_HOURS", 24) or 24 ), ), ), "review_forum_chat_id": forum_chat_id, "review_forum_username": forum_username, } async def set_service_settings(values: dict[str, Any]) -> dict[str, Any]: current = await get_service_settings() try: normalized = { "customer_max_open_orders": max( 1, min(20, int(values.get("customer_max_open_orders", current["customer_max_open_orders"]))), ), "customer_daily_request_limit": max( 1, min( 100, int( values.get( "customer_daily_request_limit", current["customer_daily_request_limit"], ) ), ), ), "review_qr_expiry_hours": max( 1, min(168, int(values.get("review_qr_expiry_hours", current["review_qr_expiry_hours"]))), ), "review_forum_chat_id": _clean_text( values.get("review_forum_chat_id", current["review_forum_chat_id"]), max_length=40, ), "review_forum_username": _clean_text( values.get( "review_forum_username", current["review_forum_username"], ), max_length=64, ).lstrip("@"), } except (TypeError, ValueError) as exc: raise ServiceDataError("invalid_service_settings", "服务设置格式无效。") from exc await settingsdb.update_one( {"settings_id": "global"}, { "$set": { "service_customer_max_open_orders": normalized["customer_max_open_orders"], "service_customer_daily_request_limit": normalized["customer_daily_request_limit"], "service_review_qr_expiry_hours": normalized["review_qr_expiry_hours"], "service_review_forum_chat_id": normalized["review_forum_chat_id"], "service_review_forum_username": normalized["review_forum_username"], "updated_at": utc_now(), }, "$setOnInsert": {"created_at": utc_now()}, }, upsert=True, ) return normalized async def list_package_templates(*, include_archived: bool = False) -> list[dict[str, Any]]: await ensure_service_indexes() filters = {} if include_archived else {"status": {"$ne": "archived"}} return await templatesdb.find(filters).sort([("sort_order", ASCENDING), ("created_at", ASCENDING)]).to_list(length=500) async def ensure_builtin_package_templates() -> list[dict[str, Any]]: """Seed editable starter templates once without overwriting administrator changes.""" await ensure_service_indexes() now = utc_now() template_ids = [] for item in BUILTIN_PACKAGE_TEMPLATES: template_id = str(item["template_id"]) template_ids.append(template_id) package = normalize_package( item["package"], package_id=str(item["package"]["package_id"]), ) package["source_type"] = "template" await templatesdb.update_one( {"template_id": template_id}, { "$setOnInsert": { "template_id": template_id, "version": 1, "name": item["name"], "admin_note": "系统内置快速模板，可在后台编辑、停用或归档。", "package": package, "status": "enabled", "sort_order": int(item["sort_order"]), "builtin": True, "created_by": "system", "created_at": now, "updated_at": now, } }, upsert=True, ) return await templatesdb.find({"template_id": {"$in": template_ids}}).sort( [("sort_order", ASCENDING)] ).to_list(length=len(template_ids)) async def create_package_template(values: dict[str, Any], *, actor_id: str) -> dict[str, Any]: await ensure_service_indexes() now = utc_now() package = normalize_package(values.get("package") or values) package["source_type"] = "template" document = { "template_id": uuid4().hex, "version": 1, "name": _clean_text(values.get("template_name") or package["name"], max_length=80, required=True), "admin_note": _clean_text(values.get("admin_note"), max_length=500), "package": package, "status": "enabled" if values.get("enabled", True) else "disabled", "sort_order": int(values.get("sort_order") or 0), "created_by": str(actor_id), "created_at": now, "updated_at": now, } await templatesdb.insert_one(document) return document async def update_package_template(template_id: str, values: dict[str, Any], *, actor_id: str) -> dict[str, Any]: current = await templatesdb.find_one({"template_id": str(template_id)}) if not current: raise ServiceDataError("template_not_found", "未找到套餐模板。") package = normalize_package(values.get("package") or values, package_id=current["package"]["package_id"]) package["source_type"] = "template" updated = await templatesdb.find_one_and_update( {"template_id": str(template_id)}, { "$set": { "name": _clean_text(values.get("template_name") or package["name"], max_length=80, required=True), "admin_note": _clean_text(values.get("admin_note"), max_length=500), "package": package, "sort_order": int(values.get("sort_order") or 0), "updated_by": str(actor_id), "updated_at": utc_now(), }, "$inc": {"version": 1}, }, return_document=ReturnDocument.AFTER, ) return updated or {} async def set_package_template_status(template_id: str, action: str, *, actor_id: str) -> dict[str, Any]: status_map = {"enable": "enabled", "disable": "disabled", "archive": "archived"} if action not in status_map: raise ServiceDataError("invalid_template_action", "不支持该模板操作。") updated = await templatesdb.find_one_and_update( {"template_id": str(template_id)}, {"$set": {"status": status_map[action], "updated_by": str(actor_id), "updated_at": utc_now()}}, return_document=ReturnDocument.AFTER, ) if not updated: raise ServiceDataError("template_not_found", "未找到套餐模板。") return updated async def duplicate_package_template(template_id: str, *, actor_id: str) -> dict[str, Any]: current = await templatesdb.find_one({"template_id": str(template_id)}) if not current: raise ServiceDataError("template_not_found", "未找到套餐模板。") values = { "template_name": f"{current['name']}（副本）", "admin_note": current.get("admin_note", ""), "sort_order": current.get("sort_order", 0), "enabled": False, "package": current["package"], } return await create_package_template(values, actor_id=actor_id) async def publish_technician_profile(user_id: int, values: dict[str, Any], *, actor_id: int | str) -> dict[str, Any]: await ensure_service_indexes() profile = await profilesdb.find_one({"user_id": int(user_id)}) if not profile or profile.get("application_status") != "approved": raise ServiceDataError("technician_required", "只有已认证技师可以发布服务资料。") if not profile.get("username"): raise ServiceDataError("username_required", "技师必须设置有效的 Telegram 用户名。") if not await membershipsdb.find_one({"user_id": int(user_id), "active": True}): raise ServiceDataError("technician_membership_required", "技师必须仍是受管群当前成员。") normalized = _normalize_profile(values) now = utc_now() normalized.update({"updated_at": now, "updated_by": actor_id, "version": int((profile.get("service_profile") or {}).get("version") or 0) + 1}) await profilesdb.update_one( {"user_id": int(user_id)}, {"$set": {"service_profile": normalized, "updated_at": now}}, ) return await profilesdb.find_one({"user_id": int(user_id)}) or {} async def get_technician_service_profile(user_id: int) -> dict[str, Any]: profile = await profilesdb.find_one({"user_id": int(user_id)}) if not profile: raise ServiceDataError("technician_not_found", "未找到技师资料。") return profile async def get_technician_self_service_context(user_id: int) -> dict[str, Any]: profile = await profilesdb.find_one({"user_id": int(user_id)}) if not profile: raise ServiceDataError("technician_not_found", "未找到技师申请资料。") membership_active = bool( await membershipsdb.find_one({"user_id": int(user_id), "active": True}) ) issues = [] if profile.get("application_status") != "approved": issues.append({"code": "approval_required", "message": "技师认证尚未通过。"}) if not profile.get("username"): issues.append( { "code": "username_required", "message": "请先在 Telegram 设置用户名。", } ) if not membership_active: issues.append( { "code": "membership_required", "message": "当前不在受管群成员名单中。", } ) return { "user_id": int(user_id), "display_name": profile.get("display_name") or f"技师 {user_id}", "username": profile.get("username"), "eligibility": {"ready": not issues, "issues": issues}, "service_profile": profile.get("service_profile") or None, "review_topic": await get_technician_review_topic(int(user_id)), } async def set_technician_accepting_requests( user_id: int, accepting_requests: bool, ) -> dict[str, Any]: profile = await get_technician_service_profile(user_id) values = dict(profile.get("service_profile") or {}) if not values.get("is_complete"): raise ServiceDataError("technician_profile_required", "请先发布至少一个服务套餐。") values["accepting_requests"] = bool(accepting_requests) return await publish_technician_profile( user_id, values, actor_id=user_id, ) async def instantiate_package_template(template_id: str) -> dict[str, Any]: template = await templatesdb.find_one( {"template_id": str(template_id), "status": "enabled"} ) if not template: raise ServiceDataError("template_not_found", "套餐模板不存在或已停用。") package = dict(template["package"]) package.update( { "package_id": uuid4().hex, "source_type": "template", "source_template_id": template["template_id"], "source_template_version": template["version"], "customized_from_template": False, } ) return package def _fernet() -> Fernet: raw = str(getattr(wbb, "SERVICE_ADDRESS_ENCRYPTION_KEY", "") or "").strip() if not raw: raise ServiceDataError("address_encryption_required", "服务地址加密密钥尚未配置。") try: return Fernet(raw.encode()) except (ValueError, TypeError) as exc: raise ServiceDataError("invalid_address_encryption_key", "服务地址加密密钥无效。") from exc def _encrypt_address(payload: dict[str, Any]) -> str: return _fernet().encrypt(json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode()).decode() def _decrypt_address(token: str) -> dict[str, Any]: try: return json.loads(_fernet().decrypt(str(token).encode()).decode()) except (InvalidToken, ValueError, json.JSONDecodeError) as exc: raise ServiceDataError("address_unavailable", "精确地址无法解密。") from exc async def _schedule_address_redaction(order_id: str, closed_at: datetime) -> None: retention = max( 1, min(365, int(getattr(wbb, "SERVICE_ADDRESS_RETENTION_DAYS", 7) or 7)), ) await addressesdb.update_one( {"order_id": str(order_id), "redacted": False}, { "$set": { "redact_after": closed_at + timedelta(days=retention), "updated_at": closed_at, } }, ) def _distance_meters(lon1: float, lat1: float, lon2: float, lat2: float) -> float: radius = 6_371_000.0 phi1, phi2 = math.radians(lat1), math.radians(lat2) delta_phi = math.radians(lat2 - lat1) delta_lambda = math.radians(lon2 - lon1) a = math.sin(delta_phi / 2) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(delta_lambda / 2) ** 2 return radius * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a)) def distance_band(distance_meters: float) -> str: km = max(0.0, distance_meters / 1000) for threshold in (1, 3, 5, 10, 20, 50): if km <= threshold: return f"{threshold} 公里内" return "50 公里以上" async def _get_package(technician_id: int, package_id: str) -> tuple[dict[str, Any], dict[str, Any]]: profile = await profilesdb.find_one({"user_id": int(technician_id)}) if not profile or profile.get("application_status") != "approved": raise ServiceDataError("technician_not_found", "未找到已认证技师。") if not profile.get("listed") or not profile.get("username"): raise ServiceDataError("technician_unavailable", "该技师暂未公开接单。") if not await membershipsdb.find_one({"user_id": int(technician_id), "active": True}): raise ServiceDataError("technician_unavailable", "该技师当前不满足接单资格。") service_profile = profile.get("service_profile") or {} if not service_profile.get("is_complete") or not service_profile.get("accepting_requests", True): raise ServiceDataError("technician_unavailable", "该技师暂未接单。") package = next((item for item in service_profile.get("packages", []) if str(item.get("package_id")) == str(package_id)), None) if not package: raise ServiceDataError("package_not_found", "未找到该服务套餐。") return profile, package async def list_available_technicians( *, longitude: float | None = None, latitude: float | None = None, max_distance_meters: float | None = None, query: str = "", page: int = 1, page_size: int = 10, ) -> tuple[list[dict[str, Any]], int]: use_distance = longitude is not None and latitude is not None lon = float(longitude) if longitude is not None else 0.0 lat = float(latitude) if latitude is not None else 0.0 if use_distance and (not -180 <= lon <= 180 or not -90 <= lat <= 90): raise ServiceDataError("invalid_coordinates", "位置坐标无效。") filters: dict[str, Any] = { "application_status": "approved", "listed": True, "username": {"$nin": [None, ""]}, "service_profile.is_complete": True, "service_profile.accepting_requests": True, } if query: pattern = re.compile(re.escape(query.lstrip("@")), re.IGNORECASE) filters["$or"] = [ {"username": pattern}, {"display_name": pattern}, {"service_profile.tags": pattern}, {"service_profile.packages.category": pattern}, ] active_members = { int(item["user_id"]) async for item in membershipsdb.find({"active": True}, {"user_id": 1}) } profiles = { int(item["user_id"]): item async for item in profilesdb.find(filters) if int(item["user_id"]) in active_members } values: list[dict[str, Any]] = [] locations = {} if profiles and use_distance: locations = { int(location["user_id"]): location async for location in locationsdb.find( {"user_id": {"$in": list(profiles)}} ) } for technician_id, profile in profiles.items(): service_profile = profile.get("service_profile") or {} value = { "user_id": int(profile["user_id"]), "username": profile.get("username"), "display_name": profile.get("display_name"), "headline": service_profile.get("headline"), "bio": service_profile.get("bio"), "tags": service_profile.get("tags", []), "public_area_text": service_profile.get("public_area_text"), "service_modes": service_profile.get("service_modes", []), "packages": service_profile.get("packages", []), "distance_meters": None, "distance_band": "", } if use_distance: location = locations.get(technician_id) if not location: continue point = location.get("point", {}).get("coordinates") or [ location.get("longitude"), location.get("latitude"), ] if len(point) != 2 or point[0] is None or point[1] is None: continue distance = _distance_meters(lon, lat, float(point[0]), float(point[1])) if max_distance_meters is not None and distance > float(max_distance_meters): continue value.update( { "distance_meters": round(distance, 2), "distance_band": distance_band(distance), } ) values.append(value) if use_distance: values.sort(key=lambda item: (item["distance_meters"], item["user_id"])) else: values.sort( key=lambda item: ( str(item.get("public_area_text") or "").casefold(), str(item.get("display_name") or "").casefold(), item["user_id"], ) ) total = len(values) start = (max(1, page) - 1) * max(1, page_size) return values[start : start + max(1, page_size)], total async def create_service_order( *, customer: Any, technician_id: int, package_id: str, service_mode: str, requirements: str, longitude: float | None = None, latitude: float | None = None, address_text: str = "", ) -> dict[str, Any]: await ensure_service_indexes() await observe_customer(customer) await require_customer_allowed(int(customer.id)) if int(customer.id) == int(technician_id): raise ServiceDataError("self_service_not_allowed", "不能向自己发起服务请求。") personal_block = await blocksdb.find_one({"scope": "technician", "technician_id": int(technician_id), "customer_id": int(customer.id), "active": True}) if personal_block: raise ServiceDataError("customer_blocked", "该技师暂不接受你的服务请求。") service_settings = await get_service_settings() max_open = service_settings["customer_max_open_orders"] open_count = await ordersdb.count_documents({"customer_id": int(customer.id), "status": {"$in": list(ORDER_ACTIVE_STATUSES)}}) if open_count >= max_open: raise ServiceDataError("too_many_open_orders", f"最多同时保留 {max_open} 个未关闭服务单。") day_start = utc_now().replace(hour=0, minute=0, second=0, microsecond=0) daily_limit = service_settings["customer_daily_request_limit"] if await ordersdb.count_documents({"customer_id": int(customer.id), "created_at": {"$gte": day_start}}) >= daily_limit: raise ServiceDataError("daily_request_limit", "今天发起的服务请求已达到上限。") profile, package = await _get_package(int(technician_id), str(package_id)) if service_mode not in package["service_modes"]: raise ServiceDataError("service_mode_unavailable", "该套餐不支持所选服务方式。") technician_location = await locationsdb.find_one({"user_id": int(technician_id)}) if not technician_location: raise ServiceDataError("technician_location_required", "技师尚未配置服务位置。") address_payload: dict[str, Any] | None = None distance = 0.0 if service_mode == "onsite": if longitude is None or latitude is None: raise ServiceDataError("customer_location_required", "上门服务必须提供位置。") lon, lat = float(longitude), float(latitude) if not -180 <= lon <= 180 or not -90 <= lat <= 90: raise ServiceDataError("invalid_coordinates", "位置坐标无效。") address = _clean_text(address_text, max_length=300, required=True) point = technician_location.get("point", {}).get("coordinates") or [technician_location.get("longitude"), technician_location.get("latitude")] distance = _distance_meters(float(point[0]), float(point[1]), lon, lat) radius = float(package.get("service_radius_km") or 0) if radius and distance > radius * 1000: raise ServiceDataError("outside_service_radius", "顾客位置超出该套餐的上门服务范围。") address_payload = {"kind": "customer", "longitude": lon, "latitude": lat, "address_text": address} else: if longitude is None or latitude is None: raise ServiceDataError( "customer_location_required", "到店服务必须先通过附近查找提供位置。" ) lon, lat = float(longitude), float(latitude) if not -180 <= lon <= 180 or not -90 <= lat <= 90: raise ServiceDataError("invalid_coordinates", "位置坐标无效。") point = technician_location.get("point", {}).get("coordinates") or [ technician_location.get("longitude"), technician_location.get("latitude"), ] distance = _distance_meters(float(point[0]), float(point[1]), lon, lat) venue = (profile.get("service_profile") or {}).get("venue") or {} address_payload = { "kind": "venue", "longitude": technician_location.get("longitude"), "latitude": technician_location.get("latitude"), "address_text": _clean_text(venue.get("address_hint"), max_length=300, required=True), } now = utc_now() order_id = uuid4().hex order = { "order_id": order_id, "customer_id": int(customer.id), "customer_name": _clean_text(getattr(customer, "first_name", ""), max_length=80) or f"用户 {customer.id}", "technician_id": int(technician_id), "technician_name": profile.get("display_name") or f"技师 {technician_id}", "package_id": str(package_id), "package_snapshot": package, "category": package["category"], "service_mode": service_mode, "requirements": _clean_text(requirements, max_length=800, required=True), "distance_meters": round(distance, 2), "distance_band": distance_band(distance), "status": "requested", "version": 1, "created_at": now, "updated_at": now, } encrypted_address = _encrypt_address(address_payload) await ordersdb.insert_one(order) try: await addressesdb.insert_one( { "order_id": order_id, "encrypted_payload": encrypted_address, "redacted": False, "redact_after": None, "created_at": now, "updated_at": now, } ) except Exception: await ordersdb.delete_one({"order_id": order_id, "status": "requested"}) raise await record_service_event(order_id, "order_requested", actor_id=int(customer.id)) return order def normalize_quote(value: dict[str, Any], *, package: dict[str, Any]) -> dict[str, Any]: currency = str(value.get("currency") or package.get("currency") or "CNY").upper() if currency != str(package.get("currency") or currency): raise ServiceDataError("currency_mismatch", "最终报价币种必须与套餐一致。") base = _positive_decimal(value.get("base_amount"), "服务金额") travel = _positive_decimal(value.get("travel_fee"), "交通费") discount = _positive_decimal(value.get("discount"), "优惠金额") addons = [] addon_total = Decimal("0") for item in value.get("addons", []) or []: quantity = max(1, min(999, int(item.get("quantity") or 1))) unit_price = Decimal(_positive_decimal(item.get("unit_price"), "加项单价")) addons.append({"name": _clean_text(item.get("name"), max_length=60, required=True), "quantity": quantity, "unit_price": format(unit_price, "f")}) addon_total += unit_price * quantity total = Decimal(base) + Decimal(travel) + addon_total - Decimal(discount) if total < 0: raise ServiceDataError("invalid_total", "最终报价总额不能小于 0。") provided = value.get("total_amount") if provided not in (None, "") and Decimal(_positive_decimal(provided, "总价")) != total: raise ServiceDataError("quote_total_mismatch", "报价明细与总价不一致。") return { "currency": currency, "base_amount": base, "travel_fee": travel, "addons": addons, "discount": discount, "total_amount": format(total.quantize(Decimal("0.01")), "f"), "note": _clean_text(value.get("note"), max_length=500), } async def submit_service_quote(order_id: str, technician_id: int, values: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]: await ensure_service_indexes() order = await ordersdb.find_one({"order_id": str(order_id)}) if not order or int(order["technician_id"]) != int(technician_id): raise ServiceDataError("order_not_found", "未找到可报价服务单。") if order["status"] not in {"requested", "quoted"}: raise ServiceDataError("invalid_order_state", "当前服务单不能修改报价。") quote = normalize_quote(values, package=order["package_snapshot"]) version = int(order.get("quote_version") or 0) + 1 quote.update({"quote_id": uuid4().hex, "order_id": str(order_id), "version": version, "created_by": int(technician_id), "created_at": utc_now()}) try: await quotesdb.insert_one(quote) except DuplicateKeyError as exc: raise ServiceDataError( "concurrent_order_update", "服务单状态已变化，请刷新后重试。" ) from exc updated = await ordersdb.find_one_and_update( { "order_id": str(order_id), "status": {"$in": ["requested", "quoted"]}, "version": order["version"], }, { "$set": { "status": "quoted", "current_quote_id": quote["quote_id"], "quote_snapshot": quote, "quote_version": version, "updated_at": utc_now(), }, "$inc": {"version": 1}, }, return_document=ReturnDocument.AFTER, ) if not updated: await quotesdb.delete_one({"quote_id": quote["quote_id"]}) raise ServiceDataError("concurrent_order_update", "服务单状态已变化，请刷新后重试。") await record_service_event(order_id, "quote_submitted", actor_id=int(technician_id), metadata={"quote_id": quote["quote_id"], "version": version}) return updated, quote async def confirm_service_quote(order_id: str, customer_id: int) -> dict[str, Any]: order = await ordersdb.find_one_and_update( {"order_id": str(order_id), "customer_id": int(customer_id), "status": "quoted"}, {"$set": {"status": "confirmed", "confirmed_at": utc_now(), "updated_at": utc_now()}, "$inc": {"version": 1}}, return_document=ReturnDocument.AFTER, ) if not order: raise ServiceDataError("invalid_order_state", "当前报价无法确认。") await record_service_event(order_id, "quote_confirmed", actor_id=int(customer_id)) return order async def reject_service_order(order_id: str, technician_id: int, reason: str) -> dict[str, Any]: now = utc_now() order = await ordersdb.find_one_and_update( {"order_id": str(order_id), "technician_id": int(technician_id), "status": {"$in": ["requested", "quoted"]}}, {"$set": {"status": "rejected", "closed_at": now, "updated_at": now, "close_reason": _clean_text(reason, max_length=500, required=True)}, "$inc": {"version": 1}}, return_document=ReturnDocument.AFTER, ) if not order: raise ServiceDataError("invalid_order_state", "当前服务单无法拒绝。") await _schedule_address_redaction(order_id, now) await record_service_event(order_id, "order_rejected", actor_id=int(technician_id), reason=reason) return order async def start_service_order(order_id: str, technician_id: int) -> dict[str, Any]: order = await ordersdb.find_one_and_update( {"order_id": str(order_id), "technician_id": int(technician_id), "status": "confirmed"}, {"$set": {"status": "in_progress", "started_at": utc_now(), "updated_at": utc_now()}, "$inc": {"version": 1}}, return_document=ReturnDocument.AFTER, ) if not order: raise ServiceDataError("invalid_order_state", "只有已确认服务单可以开始。") await record_service_event(order_id, "service_started", actor_id=int(technician_id)) return order async def cancel_service_order(order_id: str, actor_id: int, reason: str) -> dict[str, Any]: order = await ordersdb.find_one({"order_id": str(order_id)}) if not order or int(actor_id) not in {int(order["customer_id"]), int(order["technician_id"])}: raise ServiceDataError("order_not_found", "未找到该服务单。") if order["status"] not in ORDER_ACTIVE_STATUSES - {"disputed"}: raise ServiceDataError("invalid_order_state", "当前服务单不能取消。") status = "canceled_customer" if int(actor_id) == int(order["customer_id"]) else "canceled_technician" now = utc_now() updated = await ordersdb.find_one_and_update( {"order_id": str(order_id), "status": order["status"], "version": order["version"]}, {"$set": {"status": status, "closed_at": now, "updated_at": now, "close_reason": _clean_text(reason, max_length=500, required=True)}, "$inc": {"version": 1}}, return_document=ReturnDocument.AFTER, ) if not updated: raise ServiceDataError("concurrent_order_update", "服务单状态已变化，请刷新后重试。") await _schedule_address_redaction(order_id, now) await qrdb.update_many({"order_id": str(order_id), "status": "issued"}, {"$set": {"status": "revoked", "revoked_at": now, "revoke_reason": "服务单已取消"}}) await record_service_event(order_id, "order_canceled", actor_id=int(actor_id), reason=reason, metadata={"status": status}) return updated async def dispute_service_order(order_id: str, actor_id: int, reason: str) -> dict[str, Any]: current = await ordersdb.find_one( { "order_id": str(order_id), "status": {"$in": list(ORDER_ACTIVE_STATUSES - {"disputed"})}, "$or": [{"customer_id": int(actor_id)}, {"technician_id": int(actor_id)}], } ) if not current: raise ServiceDataError("invalid_order_state", "当前服务单不能发起争议。") order = await ordersdb.find_one_and_update( {"order_id": str(order_id), "status": current["status"], "version": current["version"]}, {"$set": {"status": "disputed", "status_before_dispute": current["status"], "dispute_reason": _clean_text(reason, max_length=500, required=True), "updated_at": utc_now()}, "$inc": {"version": 1}}, return_document=ReturnDocument.AFTER, ) if not order: raise ServiceDataError("invalid_order_state", "当前服务单不能发起争议。") await qrdb.update_many({"order_id": str(order_id), "status": "issued"}, {"$set": {"status": "revoked", "revoked_at": utc_now(), "revoke_reason": "服务单争议"}}) await record_service_event(order_id, "order_disputed", actor_id=int(actor_id), reason=reason) return order async def resolve_service_dispute( order_id: str, *, actor_id: str, action: str, reason: str, ) -> dict[str, Any]: if action == "void": return await admin_void_service_order(order_id, actor_id=actor_id, reason=reason) if action != "resume": raise ServiceDataError("invalid_dispute_action", "不支持该争议处理操作。") current = await ordersdb.find_one({"order_id": str(order_id), "status": "disputed"}) if not current: raise ServiceDataError("invalid_order_state", "该服务单当前不在争议中。") resume_status = str(current.get("status_before_dispute") or "confirmed") if resume_status == "completion_pending": resume_status = "in_progress" if resume_status not in {"requested", "quoted", "confirmed", "in_progress"}: resume_status = "confirmed" updated = await ordersdb.find_one_and_update( {"order_id": str(order_id), "status": "disputed", "version": current["version"]}, { "$set": { "status": resume_status, "dispute_resolved_at": utc_now(), "dispute_resolution_reason": _clean_text(reason, max_length=500, required=True), "updated_at": utc_now(), }, "$unset": {"status_before_dispute": ""}, "$inc": {"version": 1}, }, return_document=ReturnDocument.AFTER, ) if not updated: raise ServiceDataError("concurrent_order_update", "服务单状态已变化，请刷新后重试。") await record_service_event(order_id, "dispute_resolved", actor_id=actor_id, reason=reason, metadata={"status": resume_status}) return updated async def admin_void_service_order(order_id: str, *, actor_id: str, reason: str) -> dict[str, Any]: now = utc_now() order = await ordersdb.find_one_and_update( {"order_id": str(order_id), "status": {"$ne": "voided"}}, {"$set": {"status": "voided", "closed_at": now, "updated_at": now, "close_reason": _clean_text(reason, max_length=500, required=True)}, "$inc": {"version": 1}}, return_document=ReturnDocument.AFTER, ) if not order: raise ServiceDataError("order_not_found", "未找到可作废服务单。") await _schedule_address_redaction(order_id, now) await qrdb.update_many({"order_id": str(order_id), "status": {"$ne": "voided"}}, {"$set": {"status": "voided", "revoked_at": now}}) await reviewsdb.update_many({"order_id": str(order_id), "status": "approved"}, {"$set": {"status": "voided", "moderated_at": now, "moderation_reason": reason, "updated_at": now}}) await recompute_review_stats() await record_service_event(order_id, "order_voided", actor_id=actor_id, reason=reason) return order async def issue_review_qr(order_id: str, technician_id: int) -> tuple[dict[str, Any], str]: await ensure_service_indexes() order = await ordersdb.find_one({"order_id": str(order_id), "technician_id": int(technician_id)}) if order and order.get("status") == "completion_pending": previous = await qrdb.find_one( {"order_id": str(order_id), "status": "issued"}, sort=[("created_at", DESCENDING)], ) if previous and _as_utc(previous["expires_at"]) <= utc_now(): now = utc_now() expired = await qrdb.find_one_and_update( {"qr_id": previous["qr_id"], "status": "issued"}, {"$set": {"status": "expired", "expired_at": now}}, return_document=ReturnDocument.AFTER, ) if expired: await ordersdb.update_one( { "order_id": str(order_id), "status": "completion_pending", "current_review_qr_id": previous["qr_id"], }, { "$set": {"status": "in_progress", "updated_at": now}, "$unset": {"current_review_qr_id": ""}, "$inc": {"version": 1}, }, ) order = await ordersdb.find_one( {"order_id": str(order_id), "technician_id": int(technician_id)} ) if not order or order.get("status") != "in_progress": raise ServiceDataError("invalid_order_state", "只有服务中的订单可以生成评价二维码。") if await qrdb.find_one({"order_id": str(order_id), "status": {"$in": ["issued", "claimed"]}}): raise ServiceDataError("review_qr_exists", "该服务单已经生成评价二维码。") token = token_urlsafe(18) now = utc_now() hours = (await get_service_settings())["review_qr_expiry_hours"] document = { "qr_id": uuid4().hex, "order_id": str(order_id), "technician_id": int(technician_id), "customer_id": int(order["customer_id"]), "package_snapshot": order["package_snapshot"], "quote_snapshot": order.get("quote_snapshot"), "token_hash": hashlib.sha256(token.encode()).hexdigest(), "status": "issued", "expires_at": now + timedelta(hours=hours), "created_at": now, } try: await qrdb.insert_one(document) except DuplicateKeyError as exc: raise ServiceDataError("review_qr_exists", "该服务单已经生成评价二维码。") from exc updated = await ordersdb.find_one_and_update( { "order_id": str(order_id), "status": "in_progress", "version": order["version"], }, { "$set": { "status": "completion_pending", "current_review_qr_id": document["qr_id"], "completion_requested_at": now, "updated_at": now, }, "$inc": {"version": 1}, }, return_document=ReturnDocument.AFTER, ) if not updated: await qrdb.update_one( {"qr_id": document["qr_id"], "status": "issued"}, { "$set": { "status": "voided", "revoked_at": utc_now(), "revoke_reason": "服务单状态已变化", } }, ) raise ServiceDataError("concurrent_order_update", "服务单状态已变化，请刷新后重试。") await record_service_event(order_id, "review_qr_issued", actor_id=int(technician_id), metadata={"qr_id": document["qr_id"]}) return document, token async def preview_review_qr(token: str, customer_id: int) -> dict[str, Any]: document = await qrdb.find_one({"token_hash": hashlib.sha256(str(token).encode()).hexdigest()}) if not document: raise ServiceDataError("review_qr_unavailable", "评价二维码无效、已使用或已过期。") if int(document["customer_id"]) != int(customer_id): raise ServiceDataError("review_qr_customer_mismatch", "该二维码不属于当前顾客。") if document.get("status") == "claimed": if await reviewsdb.find_one({"qr_id": document["qr_id"]}): raise ServiceDataError("review_qr_unavailable", "评价二维码无效、已使用或已过期。") return document if document.get("status") != "issued": raise ServiceDataError("review_qr_unavailable", "评价二维码无效、已使用或已过期。") if _as_utc(document["expires_at"]) <= utc_now(): now = utc_now() expired = await qrdb.find_one_and_update( {"qr_id": document["qr_id"], "status": "issued"}, {"$set": {"status": "expired", "expired_at": now}}, return_document=ReturnDocument.AFTER, ) if expired: await ordersdb.update_one( { "order_id": document["order_id"], "status": "completion_pending", "current_review_qr_id": document["qr_id"], }, { "$set": {"status": "in_progress", "updated_at": now}, "$unset": {"current_review_qr_id": ""}, "$inc": {"version": 1}, }, ) raise ServiceDataError("review_qr_unavailable", "评价二维码无效、已使用或已过期。") return document async def claim_review_qr(token: str, customer_id: int) -> tuple[dict[str, Any], dict[str, Any]]: token_hash = hashlib.sha256(str(token).encode()).hexdigest() now = utc_now() await preview_review_qr(token, customer_id) document = await qrdb.find_one_and_update( {"token_hash": token_hash, "customer_id": int(customer_id), "status": "issued"}, {"$set": {"status": "claimed", "claimed_at": now}}, return_document=ReturnDocument.AFTER, ) if not document: raise ServiceDataError("review_qr_unavailable", "评价二维码无效、已使用或已过期。") order = await ordersdb.find_one_and_update( {"order_id": document["order_id"], "status": "completion_pending", "customer_id": int(customer_id)}, {"$set": {"status": "completed", "completed_at": now, "closed_at": now, "updated_at": now}, "$inc": {"version": 1}}, return_document=ReturnDocument.AFTER, ) if not order: await qrdb.update_one({"qr_id": document["qr_id"]}, {"$set": {"status": "voided"}}) raise ServiceDataError("invalid_order_state", "服务单状态已变化，无法完成评价。") await _schedule_address_redaction(order["order_id"], now) await record_service_event(order["order_id"], "service_completed", actor_id=int(customer_id), metadata={"qr_id": document["qr_id"]}) return document, order def _legacy_review_questions(values: dict[str, Any]) -> list[dict[str, Any]]: questions = [] for item in values.get("rating_questions", []): questions.append( { **item, "type": "single_choice", "options": [ {"value": str(score), "label": f"{score} 分", "score": score} for score in range(5, 0, -1) ], } ) for item in values.get("text_questions", []): questions.append({**item, "type": "text"}) return questions def normalize_review_template(values: dict[str, Any]) -> dict[str, Any]: raw_questions = values.get("questions") if raw_questions is None: raw_questions = _legacy_review_questions(values) if not isinstance(raw_questions, list) or not 1 <= len(raw_questions) <= 8: raise ServiceDataError( "invalid_review_questions", "评价模板必须包含 1 到 8 个问题。", ) questions = [] question_ids: set[str] = set() scoring_weight = 0 for raw in raw_questions: if not isinstance(raw, dict): raise ServiceDataError("invalid_review_question", "评价问题格式无效。") question_type = str(raw.get("type") or "").strip() if question_type == "input": question_type = "text" if question_type not in {"single_choice", "multiple_choice", "text"}: raise ServiceDataError( "invalid_review_question_type", "评价问题仅支持单选、多选和输入框。", ) question_id = _clean_text( raw.get("question_id") or uuid4().hex[:12], max_length=32, required=True, ) if question_id in question_ids: raise ServiceDataError("duplicate_review_question", "评价问题 ID 不能重复。") question_ids.add(question_id) question = { "question_id": question_id, "type": question_type, "label": _clean_text(raw.get("label"), max_length=50, required=True), "description": _clean_text(raw.get("description"), max_length=160), "required": bool(raw.get("required")), } if question_type == "text": question["max_length"] = max( 50, min(1000, int(raw.get("max_length") or 500)), ) else: raw_options = raw.get("options") if not isinstance(raw_options, list) or not 2 <= len(raw_options) <= 8: raise ServiceDataError( "invalid_review_options", "单选或多选问题必须设置 2 到 8 个选项。", ) options = [] option_values: set[str] = set() for index, raw_option in enumerate(raw_options): if isinstance(raw_option, str): raw_option = {"value": str(index + 1), "label": raw_option} if not isinstance(raw_option, dict): raise ServiceDataError( "invalid_review_options", "评价选项格式无效。", ) option_value = _clean_text( raw_option.get("value") or str(index + 1), max_length=24, required=True, ) if option_value in option_values: raise ServiceDataError( "duplicate_review_option", "同一问题的选项值不能重复。", ) option_values.add(option_value) option = { "value": option_value, "label": _clean_text( raw_option.get("label"), max_length=24, required=True, ), } if question_type == "single_choice": score = int(raw_option.get("score") or 0) if not 1 <= score <= 5: raise ServiceDataError( "invalid_review_option_score", "单选项评分必须在 1 到 5 之间。", ) option["score"] = score options.append(option) question["options"] = options if question_type == "single_choice": weight = int(raw.get("weight") or 0) if weight <= 0: raise ServiceDataError( "invalid_review_weight", "单选评分问题的权重必须大于 0。", ) question["weight"] = weight scoring_weight += weight questions.append(question) if scoring_weight != 100: raise ServiceDataError( "invalid_review_weight", "全部单选评分问题的权重合计必须为 100%。", ) return {"questions": questions} async def get_active_review_template() -> dict[str, Any]: await ensure_service_indexes() current = await reviewtemplatesdb.find_one({"status": "active"}, sort=[("version", DESCENDING)]) if current: return {**current, **normalize_review_template(current)} now = utc_now() document = {"template_id": uuid4().hex, "version": 1, "status": "active", **normalize_review_template(DEFAULT_REVIEW_TEMPLATE), "created_at": now, "published_at": now} try: await reviewtemplatesdb.insert_one(document) except DuplicateKeyError: return await reviewtemplatesdb.find_one({"status": "active"}, sort=[("version", DESCENDING)]) or document return document async def publish_review_template(values: dict[str, Any], *, actor_id: str) -> dict[str, Any]: normalized = normalize_review_template(values) await ensure_service_indexes() while True: current = await reviewtemplatesdb.find_one(sort=[("version", DESCENDING)]) version = int((current or {}).get("version") or 0) + 1 now = utc_now() document = { "template_id": uuid4().hex, "version": version, "status": "active", **normalized, "published_by": str(actor_id), "created_at": now, "published_at": now, } try: await reviewtemplatesdb.insert_one(document) break except DuplicateKeyError: continue await reviewtemplatesdb.update_many( {"status": "active", "template_id": {"$ne": document["template_id"]}}, {"$set": {"status": "retired", "retired_at": now}}, ) return document async def begin_review_draft( *, customer_id: int, technician_id: int, source: str, template: dict[str, Any], package_id: str | None = None, qr_id: str | None = None, ) -> dict[str, Any]: await ensure_service_indexes() now = utc_now() current = await reviewdraftsdb.find_one({"customer_id": int(customer_id)}) identity = { "technician_id": int(technician_id), "source": str(source), "package_id": str(package_id) if package_id else None, "qr_id": str(qr_id) if qr_id else None, } if ( current and all(current.get(key) == value for key, value in identity.items()) and _as_utc(current["expires_at"]) > now ): snapshot = current.get("template_snapshot") or {} if "questions" not in snapshot: normalized_snapshot = { "template_id": snapshot.get("template_id") or template["template_id"], "version": snapshot.get("version") or template["version"], "questions": normalize_review_template(snapshot)["questions"], } question_index = int(current.get("rating_index") or 0) + int( current.get("text_index") or 0 ) await reviewdraftsdb.update_one( {"draft_id": current["draft_id"]}, { "$set": { "template_snapshot": normalized_snapshot, "question_index": question_index, "updated_at": now, }, "$unset": {"rating_index": "", "text_index": ""}, }, ) current.update( { "template_snapshot": normalized_snapshot, "question_index": question_index, } ) return current normalized_template = normalize_review_template(template) document = { "draft_id": uuid4().hex, "customer_id": int(customer_id), **identity, "template_snapshot": { "template_id": template["template_id"], "version": template["version"], "questions": normalized_template["questions"], }, "question_index": 0, "answers": {}, "created_at": now, "updated_at": now, "expires_at": now + timedelta(days=30), } await reviewdraftsdb.update_one( {"customer_id": int(customer_id)}, {"$set": document}, upsert=True ) return document async def save_review_draft_progress( customer_id: int, *, answers: dict[str, Any], question_index: int | None = None, rating_index: int = 0, text_index: int = 0, ) -> None: current_index = ( max(0, int(question_index)) if question_index is not None else max(0, int(rating_index) + int(text_index)) ) await reviewdraftsdb.update_one( {"customer_id": int(customer_id)}, { "$set": { "answers": dict(answers), "question_index": current_index, "updated_at": utc_now(), } }, ) async def delete_review_draft(customer_id: int) -> None: await reviewdraftsdb.delete_one({"customer_id": int(customer_id)}) def _score_review( template: dict[str, Any], answers: dict[str, Any], ) -> tuple[float, dict[str, int], dict[str, Any], dict[str, str]]: questions = normalize_review_template(template)["questions"] ratings: dict[str, int] = {} choices: dict[str, Any] = {} texts: dict[str, str] = {} total = 0.0 for question in questions: raw = answers.get(question["question_id"]) if question["type"] == "text": text = _clean_text( raw, max_length=question["max_length"], required=question["required"], ) if text: texts[question["question_id"]] = text continue options = { str(option["value"]): option for option in question["options"] } if question["type"] == "single_choice": if raw in (None, "") and not question["required"]: continue option = options.get(str(raw)) if not option: raise ServiceDataError("invalid_review_answer", "请选择有效的单选项。") score = int(option["score"]) ratings[question["question_id"]] = score choices[question["question_id"]] = option["label"] total += score * question["weight"] / 100 continue selected = raw if isinstance(raw, list) else ([] if raw in (None, "") else [raw]) selected_values = list(dict.fromkeys(str(item) for item in selected)) if question["required"] and not selected_values: raise ServiceDataError("required_field", f"{question['label']} 为必选项。") if set(selected_values) - set(options): raise ServiceDataError("invalid_review_answer", "请选择有效的多选项。") if selected_values: choices[question["question_id"]] = [ options[value]["label"] for value in selected_values ] return round(total, 4), ratings, choices, texts async def submit_review( *, customer: Any, technician_id: int, source: str, answers: dict[str, Any], anonymous: bool, package_id: str | None = None, qr_id: str | None = None, template_snapshot: dict[str, Any] | None = None, ) -> dict[str, Any]: await observe_customer(customer) await require_customer_allowed(int(customer.id)) if source not in REVIEW_SOURCES: raise ServiceDataError("invalid_review_source", "评价来源无效。") profile = await profilesdb.find_one({"user_id": int(technician_id), "application_status": "approved"}) if not profile: raise ServiceDataError("technician_not_found", "未找到已认证技师。") if int(customer.id) == int(technician_id): raise ServiceDataError("self_review_not_allowed", "不能评价自己。") order_id = None package_snapshot = None if source == "qr_verified": qr = await qrdb.find_one({"qr_id": str(qr_id), "status": "claimed", "customer_id": int(customer.id), "technician_id": int(technician_id)}) if not qr: raise ServiceDataError("review_qr_required", "缺少已完成服务单的评价资格。") if await reviewsdb.find_one({"qr_id": str(qr_id)}): raise ServiceDataError("review_already_submitted", "该服务单已经提交评价。") order_id = qr["order_id"] package_snapshot = qr["package_snapshot"] else: service_profile = profile.get("service_profile") or {} package_snapshot = next((item for item in service_profile.get("packages", []) if str(item.get("package_id")) == str(package_id)), None) if package_id and not package_snapshot: raise ServiceDataError("package_not_found", "未找到所选套餐。") if not package_snapshot: packages = service_profile.get("packages") or [] primary_package = packages[0] if packages else {} package_snapshot = { "package_id": None, "name": "技师服务", "category": primary_package.get("category") or "其他", } template = template_snapshot or await get_active_review_template() normalized_template = normalize_review_template(template) score, rating_answers, choice_answers, text_answers = _score_review( normalized_template, answers, ) normalized_content = "\n".join( str(value).casefold() for _, value in sorted( {**choice_answers, **text_answers}.items() ) if value ) now = utc_now() document = { "review_id": uuid4().hex, "technician_id": int(technician_id), "technician_name": profile.get("display_name") or f"技师 {technician_id}", "customer_id": int(customer.id), "customer_name": _clean_text(getattr(customer, "first_name", ""), max_length=80) or f"用户 {customer.id}", "source": source, "package_snapshot": package_snapshot, "category": package_snapshot.get("category") or "其他", "template_snapshot": { "template_id": template["template_id"], "version": template["version"], "questions": normalized_template["questions"], }, "rating_answers": rating_answers, "choice_answers": choice_answers, "text_answers": text_answers, "content_fingerprint": ( hashlib.sha256(normalized_content.encode()).hexdigest() if normalized_content else None ), "score": score, "anonymous": bool(anonymous), "status": "pending", "created_at": now, "updated_at": now, } if qr_id: document["qr_id"] = str(qr_id) document["order_id"] = order_id try: await reviewsdb.insert_one(document) except DuplicateKeyError as exc: if qr_id: raise ServiceDataError( "review_already_submitted", "该服务单已经提交评价。" ) from exc raise return document async def moderate_review(review_id: str, action: str, *, actor_id: str, reason: str = "") -> dict[str, Any]: status_map = {"approve": "approved", "reject": "rejected", "void": "voided"} source_status = {"approve": "pending", "reject": "pending", "void": "approved"} if action not in status_map: raise ServiceDataError("invalid_review_action", "不支持该评价操作。") current = await reviewsdb.find_one({"review_id": str(review_id)}) if not current: raise ServiceDataError("review_not_found", "未找到评价。") if current["status"] != source_status[action]: raise ServiceDataError("invalid_review_state", "当前评价状态不能执行该审核操作。") if action in {"reject", "void"} and not _clean_text(reason, max_length=500): raise ServiceDataError("reason_required", "拒绝或作废必须填写原因。") updated = await reviewsdb.find_one_and_update( {"review_id": str(review_id), "status": source_status[action]}, {"$set": {"status": status_map[action], "moderated_by": str(actor_id), "moderated_at": utc_now(), "moderation_reason": _clean_text(reason, max_length=500), "updated_at": utc_now()}}, return_document=ReturnDocument.AFTER, ) if not updated: raise ServiceDataError("concurrent_review_update", "评价状态已变化，请刷新后重试。") await recompute_review_stats() return updated async def recompute_review_stats() -> None: await ensure_service_indexes() approved = await reviewsdb.find({"status": "approved"}).to_list(length=100000) grouped: dict[tuple[int, str], list[dict[str, Any]]] = defaultdict(list) category_scores: dict[str, list[float]] = defaultdict(list) for review in approved: key = (int(review["technician_id"]), str(review.get("category") or "其他")) grouped[key].append(review) category_scores[key[1]].append(float(review["score"])) now = utc_now() await statsdb.delete_many({}) documents = [] for (technician_id, category), values in grouped.items(): count = len(values) average = sum(float(item["score"]) for item in values) / count category_average = sum(category_scores[category]) / len(category_scores[category]) rank_score = count / (count + 5) * average + 5 / (count + 5) * category_average dimensions: dict[str, list[int]] = defaultdict(list) for review in values: for key, score in review.get("rating_answers", {}).items(): dimensions[key].append(int(score)) documents.append( { "technician_id": technician_id, "category": category, "review_count": count, "average_score": round(average, 4), "category_average": round(category_average, 4), "rank_score": round(rank_score, 6), "eligible": count >= 1, "dimension_averages": {key: round(sum(items) / len(items), 4) for key, items in dimensions.items()}, "last_review_at": max(item.get("moderated_at") or item["created_at"] for item in values), "updated_at": now, } ) if documents: await statsdb.insert_many(documents) async def list_leaderboard(*, category: str = "", page: int = 1, page_size: int = 20) -> tuple[list[dict[str, Any]], int]: await ensure_service_indexes() filters: dict[str, Any] = {"review_count": {"$gte": 1}} if category: filters["category"] = str(category) visible_ids = [ int(item["user_id"]) async for item in profilesdb.find( { "application_status": "approved", "listed": True, "username": {"$nin": [None, ""]}, }, {"user_id": 1}, ) ] filters["technician_id"] = {"$in": visible_ids} total = await statsdb.count_documents(filters) items = await statsdb.find(filters).sort([("rank_score", DESCENDING), ("review_count", DESCENDING), ("last_review_at", DESCENDING)]).skip((max(1, page) - 1) * page_size).limit(page_size).to_list(length=page_size) profiles = {int(item["user_id"]): item async for item in profilesdb.find({"user_id": {"$in": [item["technician_id"] for item in items]}, "application_status": "approved", "listed": True})} visible = [{**item, "display_name": profiles[item["technician_id"]].get("display_name"), "username": profiles[item["technician_id"]].get("username")} for item in items if item["technician_id"] in profiles] return visible, total async def list_public_reviews( technician_id: int, *, page: int = 1, page_size: int = 10, ) -> tuple[list[dict[str, Any]], int]: filters = {"technician_id": int(technician_id), "status": "approved"} total = await reviewsdb.count_documents(filters) items = await reviewsdb.find(filters).sort("moderated_at", DESCENDING).skip( (max(1, page) - 1) * page_size ).limit(page_size).to_list(length=page_size) public_items = [] for item in items: public_items.append( { "review_id": item["review_id"], "source": item["source"], "source_label": ( "完成服务单评价" if item["source"] == "qr_verified" else "用户主动评价" ), "customer_name": "匿名顾客" if item.get("anonymous", True) else item.get("customer_name"), "package_name": (item.get("package_snapshot") or {}).get("name"), "category": item.get("category"), "score": item.get("score"), "rating_answers": item.get("rating_answers", {}), "choice_answers": item.get("choice_answers", {}), "text_answers": item.get("text_answers", {}), "template_snapshot": item.get("template_snapshot", {}), "approved_at": item.get("moderated_at"), "topic_url": item.get("topic_url"), } ) return public_items, total async def get_technician_review_topic( technician_id: int, ) -> dict[str, Any] | None: await ensure_service_indexes() return await reviewtopicsdb.find_one( {"technician_id": int(technician_id), "status": "active"} ) async def save_technician_review_topic( *, technician_id: int, forum_chat_id: str, forum_username: str, message_thread_id: int, topic_name: str, topic_url: str, ) -> dict[str, Any]: await ensure_service_indexes() now = utc_now() await reviewtopicsdb.update_one( {"technician_id": int(technician_id)}, { "$set": { "forum_chat_id": str(forum_chat_id), "forum_username": str(forum_username), "message_thread_id": int(message_thread_id), "topic_name": _clean_text( topic_name, max_length=128, required=True, ), "topic_url": _clean_text( topic_url, max_length=300, required=True, ), "status": "active", "updated_at": now, }, "$setOnInsert": { "technician_id": int(technician_id), "created_at": now, }, }, upsert=True, ) return await get_technician_review_topic(int(technician_id)) or {} async def mark_review_topic_published( review_id: str, *, topic: dict[str, Any], message_id: int, ) -> dict[str, Any]: updated = await reviewsdb.find_one_and_update( {"review_id": str(review_id), "status": "approved"}, { "$set": { "topic_url": topic["topic_url"], "topic_message_id": int(message_id), "topic_published_at": utc_now(), "updated_at": utc_now(), }, "$unset": {"topic_publish_error": ""}, }, return_document=ReturnDocument.AFTER, ) return updated or {} async def get_review(review_id: str) -> dict[str, Any] | None: return await reviewsdb.find_one({"review_id": str(review_id)}) async def mark_review_topic_publish_error( review_id: str, error: str, ) -> None: await reviewsdb.update_one( {"review_id": str(review_id), "status": "approved"}, { "$set": { "topic_publish_error": _clean_text(error, max_length=300), "updated_at": utc_now(), } }, ) async def get_technician_review_summary(technician_id: int) -> dict[str, Any]: items = await reviewsdb.find( {"technician_id": int(technician_id), "status": "approved"}, {"score": 1, "source": 1}, ).to_list(length=100000) count = len(items) return { "review_count": count, "average_score": round( sum(float(item["score"]) for item in items) / count, 4, ) if count else 0, "source_counts": { source: sum(1 for item in items if item.get("source") == source) for source in REVIEW_SOURCES }, "ranking_title": "审核评价排行", } async def list_service_orders(*, query: str = "", status: str = "", page: int = 1, page_size: int = 20) -> tuple[list[dict[str, Any]], int]: await ensure_service_indexes() filters: dict[str, Any] = {} if status: filters["status"] = status if query: if query.isdigit(): filters["$or"] = [{"customer_id": int(query)}, {"technician_id": int(query)}] else: pattern = re.compile(re.escape(query), re.IGNORECASE) filters["$or"] = [{"order_id": pattern}, {"customer_name": pattern}, {"technician_name": pattern}] total = await ordersdb.count_documents(filters) items = await ordersdb.find(filters).sort("created_at", DESCENDING).skip((max(1, page) - 1) * page_size).limit(page_size).to_list(length=page_size) return items, total async def list_actor_service_orders( actor_id: int, *, role: str = "all", page: int = 1, page_size: int = 20, ) -> tuple[list[dict[str, Any]], int]: if role == "customer": filters: dict[str, Any] = {"customer_id": int(actor_id)} elif role == "technician": filters = {"technician_id": int(actor_id)} else: filters = { "$or": [ {"customer_id": int(actor_id)}, {"technician_id": int(actor_id)}, ] } total = await ordersdb.count_documents(filters) items = await ordersdb.find(filters).sort("created_at", DESCENDING).skip( (max(1, page) - 1) * page_size ).limit(page_size).to_list(length=page_size) return items, total async def get_admin_service_order(order_id: str) -> dict[str, Any]: order = await ordersdb.find_one({"order_id": str(order_id)}) if not order: raise ServiceDataError("order_not_found", "未找到服务单。") result = dict(order) result["quotes"] = await quotesdb.find({"order_id": str(order_id)}).sort( "version", ASCENDING ).to_list(length=100) result["events"] = await eventsdb.find({"order_id": str(order_id)}).sort( "created_at", ASCENDING ).to_list(length=500) result["qr_records"] = await qrdb.find({"order_id": str(order_id)}).sort( "created_at", DESCENDING ).to_list(length=100) return result async def list_review_qr_records( *, status: str = "", query: str = "", page: int = 1, page_size: int = 20, ) -> tuple[list[dict[str, Any]], int]: filters: dict[str, Any] = {} if status: filters["status"] = status if query: filters["$or"] = [ {"order_id": re.compile(re.escape(query), re.IGNORECASE)}, {"qr_id": re.compile(re.escape(query), re.IGNORECASE)}, {"customer_id": int(query) if query.isdigit() else -1}, {"technician_id": int(query) if query.isdigit() else -1}, ] total = await qrdb.count_documents(filters) items = await qrdb.find(filters).sort("created_at", DESCENDING).skip( (max(1, page) - 1) * page_size ).limit(page_size).to_list(length=page_size) for item in items: item.pop("token_hash", None) return items, total async def list_reviews(*, status: str = "", source: str = "", query: str = "", page: int = 1, page_size: int = 20) -> tuple[list[dict[str, Any]], int]: await ensure_service_indexes() filters: dict[str, Any] = {} if status: filters["status"] = status if source: filters["source"] = source if query: pattern = re.compile(re.escape(query), re.IGNORECASE) filters["$or"] = [{"technician_name": pattern}, {"customer_name": pattern}, {"review_id": pattern}] total = await reviewsdb.count_documents(filters) items = await reviewsdb.find(filters).sort("created_at", DESCENDING).skip((max(1, page) - 1) * page_size).limit(page_size).to_list(length=page_size) for item in items: item["customer_review_count"] = await reviewsdb.count_documents({"customer_id": item["customer_id"], "technician_id": item["technician_id"]}) item["customer_review_count_30d"] = await reviewsdb.count_documents({"customer_id": item["customer_id"], "technician_id": item["technician_id"], "created_at": {"$gte": utc_now() - timedelta(days=30)}}) item["customer_review_count_90d"] = await reviewsdb.count_documents({"customer_id": item["customer_id"], "technician_id": item["technician_id"], "created_at": {"$gte": utc_now() - timedelta(days=90)}}) item["similar_content_count"] = ( await reviewsdb.count_documents( { "customer_id": item["customer_id"], "technician_id": item["technician_id"], "content_fingerprint": item["content_fingerprint"], } ) if item.get("content_fingerprint") else 0 ) return items, total async def get_order_for_actor(order_id: str, actor_id: int, *, reveal_address: bool = False) -> dict[str, Any]: order = await ordersdb.find_one({"order_id": str(order_id)}) if not order or int(actor_id) not in {int(order["customer_id"]), int(order["technician_id"])}: raise ServiceDataError("order_not_found", "未找到服务单。") result = dict(order) can_reveal = order["status"] in {"confirmed", "in_progress", "completion_pending", "completed", "disputed"} if reveal_address and can_reveal: address = await addressesdb.find_one({"order_id": str(order_id), "redacted": False}) if address: result["exact_address"] = _decrypt_address(address["encrypted_payload"]) return result async def admin_reveal_order_address(order_id: str, *, actor_id: str, reason: str) -> dict[str, Any]: reason = _clean_text(reason, max_length=500, required=True) order = await ordersdb.find_one({"order_id": str(order_id)}) address = await addressesdb.find_one({"order_id": str(order_id), "redacted": False}) if not order or not address: raise ServiceDataError("address_unavailable", "精确地址已脱敏或不存在。") await record_service_event(order_id, "admin_address_revealed", actor_id=actor_id, reason=reason) return _decrypt_address(address["encrypted_payload"]) async def redact_expired_addresses() -> int: now = utc_now() candidates = await addressesdb.find({"redacted": False, "redact_after": {"$lte": now}}).to_list(length=1000) count = 0 for item in candidates: order = await ordersdb.find_one({"order_id": item["order_id"]}) if not order or order.get("status") not in ORDER_TERMINAL_STATUSES: continue result = await addressesdb.update_one({"_id": item["_id"], "redacted": False}, {"$set": {"redacted": True, "redacted_at": now, "updated_at": now}, "$unset": {"encrypted_payload": ""}}) if result.modified_count: await ordersdb.update_one( {"order_id": item["order_id"]}, { "$set": {"address_redacted": True, "updated_at": now}, "$unset": {"distance_meters": ""}, }, ) count += result.modified_count return count async def set_customer_block(*, customer_id: int, scope: str, technician_id: int | None, active: bool, actor_id: int | str, reason: str) -> dict[str, Any]: if scope not in {"global", "technician"}: raise ServiceDataError("invalid_block_scope", "拉黑范围无效。") if scope == "technician" and technician_id is None: raise ServiceDataError("technician_required", "技师拉黑必须指定技师。") now = utc_now() await blocksdb.update_one( {"scope": scope, "technician_id": int(technician_id or 0), "customer_id": int(customer_id)}, {"$set": {"active": bool(active), "reason": _clean_text(reason, max_length=500, required=active), "actor_id": actor_id, "updated_at": now}, "$setOnInsert": {"created_at": now}}, upsert=True, ) return await blocksdb.find_one({"scope": scope, "technician_id": int(technician_id or 0), "customer_id": int(customer_id)}) or {} async def set_technician_customer_block( *, technician_id: int, customer_id: int, active: bool, reason: str, ) -> dict[str, Any]: profile = await profilesdb.find_one( {"user_id": int(technician_id), "application_status": "approved"} ) if not profile: raise ServiceDataError("technician_required", "只有已认证技师可以管理个人拉黑。") if not await ordersdb.find_one( {"technician_id": int(technician_id), "customer_id": int(customer_id)} ): raise ServiceDataError("customer_relationship_required", "只能拉黑曾向你发起服务请求的顾客。") return await set_customer_block( customer_id=int(customer_id), scope="technician", technician_id=int(technician_id), active=active, actor_id=int(technician_id), reason=reason, ) async def list_customer_blocks( *, scope: str = "global", active: bool | None = None, page: int = 1, page_size: int = 20, ) -> tuple[list[dict[str, Any]], int]: filters: dict[str, Any] = {"scope": scope} if active is not None: filters["active"] = active total = await blocksdb.count_documents(filters) items = await blocksdb.find(filters).sort("updated_at", DESCENDING).skip( (max(1, page) - 1) * page_size ).limit(page_size).to_list(length=page_size) return items, total async def create_service_report(*, reporter_id: int, target_type: str, target_id: str, reason: str) -> dict[str, Any]: if target_type not in {"technician", "customer", "order", "review"}: raise ServiceDataError("invalid_report_target", "举报对象无效。") document = {"report_id": uuid4().hex, "reporter_id": int(reporter_id), "target_type": target_type, "target_id": str(target_id), "reason": _clean_text(reason, max_length=800, required=True), "status": "open", "created_at": utc_now()} await reportsdb.insert_one(document) return document async def list_service_reports( *, status: str = "", page: int = 1, page_size: int = 20, ) -> tuple[list[dict[str, Any]], int]: filters = {"status": status} if status else {} total = await reportsdb.count_documents(filters) items = await reportsdb.find(filters).sort("created_at", DESCENDING).skip( (max(1, page) - 1) * page_size ).limit(page_size).to_list(length=page_size) return items, total async def moderate_service_report( report_id: str, *, action: str, actor_id: str, reason: str, ) -> dict[str, Any]: status_map = {"resolve": "resolved", "dismiss": "dismissed"} if action not in status_map: raise ServiceDataError("invalid_report_action", "不支持该举报操作。") updated = await reportsdb.find_one_and_update( {"report_id": str(report_id), "status": "open"}, { "$set": { "status": status_map[action], "handled_by": str(actor_id), "handled_at": utc_now(), "handling_reason": _clean_text(reason, max_length=500, required=True), } }, return_document=ReturnDocument.AFTER, ) if not updated: raise ServiceDataError("report_not_found", "举报不存在或已处理。") return updated async def fulfillment_metrics() -> dict[str, Any]: await ensure_service_indexes() counts = {status: await ordersdb.count_documents({"status": status}) for status in ORDER_STATUSES} valid_requests = sum(counts.values()) - counts["voided"] quoted = await ordersdb.count_documents( {"status": {"$ne": "voided"}, "current_quote_id": {"$exists": True}} ) confirmed = await ordersdb.count_documents( {"status": {"$ne": "voided"}, "confirmed_at": {"$exists": True}} ) completed = counts["completed"] canceled = counts["canceled_customer"] + counts["canceled_technician"] directory_views = await eventsdb.count_documents({"event_type": "directory_viewed"}) technician_views = await eventsdb.count_documents({"event_type": "technician_viewed"}) request_starts = await eventsdb.count_documents({"event_type": "request_started"}) return { "counts": counts, "valid_requests": valid_requests, "quoted_requests": quoted, "confirmed_requests": confirmed, "directory_views": directory_views, "technician_views": technician_views, "request_starts": request_starts, "request_submission_rate": min(1.0, round(valid_requests / request_starts, 4)) if request_starts else 0, "quote_rate": round(quoted / valid_requests, 4) if valid_requests else 0, "customer_confirmation_rate": round(confirmed / quoted, 4) if quoted else 0, "match_success_rate": round(confirmed / valid_requests, 4) if valid_requests else 0, "fulfillment_rate": round(completed / confirmed, 4) if confirmed else 0, "cancellation_rate": round(canceled / valid_requests, 4) if valid_requests else 0, "cancellation_distribution": { "customer": counts["canceled_customer"], "technician": counts["canceled_technician"], }, "scope_note": "仅统计平台服务单；直接 Telegram 私聊成交不在统计范围。", }