TelegramBotAdmin/tests/test_service_fulfillment.py

from __future__ import annotations

import asyncio
from datetime import timedelta
from types import SimpleNamespace

import pytest
from aiohttp import CookieJar
from aiohttp.test_utils import TestClient, TestServer


def _user(user_id: int, *, username: str | None = None, name: str = "测试用户"):
    return SimpleNamespace(
        id=user_id,
        username=username,
        first_name=name,
        last_name=None,
        is_bot=False,
    )


def _package(*, name: str = "基础护理", category: str = "护理"):
    return {
        "name": name,
        "category": category,
        "description": "标准线下服务套餐",
        "tags": ["线下", "护理"],
        "service_modes": ["at_store", "onsite"],
        "price_mode": "fixed",
        "currency": "CNY",
        "price_unit": "per_service",
        "min_price": "100",
        "max_price": "100",
        "duration_minutes": 60,
        "included_items": "基础服务",
        "excluded_items": "额外耗材",
        "addons": [
            {
                "name": "延长服务",
                "unit": "次",
                "price": "20",
                "extra_minutes": 15,
            }
        ],
        "travel_fee": {"mode": "quoted"},
        "service_radius_km": 20,
        "out_of_range_policy": "超区请重新询价",
        "preparation": "请提前沟通需求",
    }


async def _technician(app_modules, service_db, *, user_id: int = 200, category: str = "护理"):
    directory_db = app_modules.load("wbb.utils.dbdirectory")
    await directory_db.upsert_directory_identity(
        user_id=user_id,
        username=f"technician{user_id}",
        first_name="认证",
        last_name="技师",
    )
    await directory_db.save_directory_location(
        user_id=user_id,
        longitude=116.4074,
        latitude=39.9042,
        source="test",
        actor_id=user_id,
        actor_name="认证技师",
    )
    await directory_db.upsert_directory_membership(
        bot_id="primary",
        chat_id=-100200,
        user_id=user_id,
        status="member",
        active=True,
        verified=True,
    )
    await directory_db.submit_teacher_application(
        user_id=user_id,
        source="test",
        bot_id="primary",
        chat_id=-100200,
    )
    await directory_db.decide_teacher_application(
        user_id=user_id,
        action="approve",
        actor_id="admin",
        actor_name="admin",
        source="test",
    )
    await directory_db.set_teacher_state(
        user_id=user_id,
        action="list",
        actor_id=user_id,
        actor_name="认证技师",
        source="test",
    )
    profile = await service_db.publish_technician_profile(
        user_id,
        {
            "headline": "专注线下护理",
            "bio": "提供到店和上门服务",
            "tags": ["护理"],
            "contact_hours": "10:00-20:00",
            "public_area_text": "北京朝阳区附近",
            "venue": {"name": "护理工作室", "address_hint": "测试路 100 号"},
            "onsite_policy": {"description": "20 公里内可上门"},
            "packages": [_package(category=category)],
        },
        actor_id=user_id,
    )
    return profile


async def _accepted_customer(service_db, *, user_id: int = 300):
    customer = _user(user_id, username=None, name=f"顾客{user_id}")
    await service_db.observe_customer(customer, accepted_terms=True)
    return customer


async def _order(service_db, customer, profile, *, service_mode: str = "onsite"):
    package = profile["service_profile"]["packages"][0]
    kwargs = {
        "customer": customer,
        "technician_id": profile["user_id"],
        "package_id": package["package_id"],
        "service_mode": service_mode,
        "requirements": "希望完成一次标准护理服务",
    }
    if service_mode == "onsite":
        kwargs.update(
            {
                "longitude": 116.4174,
                "latitude": 39.9042,
                "address_text": "测试街道 1 号 101 室",
            }
        )
    else:
        kwargs.update({"longitude": 116.4174, "latitude": 39.9042})
    return await service_db.create_service_order(**kwargs)


async def test_public_customer_order_quote_address_and_completion_qr(app_modules):
    service_db = app_modules.load("wbb.utils.dbservice")
    profile = await _technician(app_modules, service_db)
    customer = await _accepted_customer(service_db)

    order = await _order(service_db, customer, profile)
    assert order["status"] == "requested"
    assert order["distance_band"] == "1 公里内"
    stored_address = await app_modules.wbb.control_db.service_order_addresses.find_one(
        {"order_id": order["order_id"]}
    )
    assert stored_address["redact_after"] is None
    assert "测试街道" not in stored_address["encrypted_payload"]

    hidden = await service_db.get_order_for_actor(
        order["order_id"], customer.id, reveal_address=True
    )
    assert "exact_address" not in hidden
    quoted_order, quote = await service_db.submit_service_quote(
        order["order_id"],
        profile["user_id"],
        {
            "base_amount": "100",
            "travel_fee": "10",
            "addons": [{"name": "延长服务", "quantity": 2, "unit_price": "20"}],
            "discount": "5",
            "currency": "CNY",
            "total_amount": "145",
            "note": "最终报价",
        },
    )
    assert quoted_order["status"] == "quoted"
    assert quote["total_amount"] == "145.00"

    confirmed = await service_db.confirm_service_quote(order["order_id"], customer.id)
    assert confirmed["status"] == "confirmed"
    revealed = await service_db.get_order_for_actor(
        order["order_id"], profile["user_id"], reveal_address=True
    )
    assert revealed["exact_address"]["address_text"] == "测试街道 1 号 101 室"

    started = await service_db.start_service_order(order["order_id"], profile["user_id"])
    assert started["status"] == "in_progress"
    qr, token = await service_db.issue_review_qr(order["order_id"], profile["user_id"])
    assert qr["status"] == "issued"
    assert len(token) >= 22
    assert token not in str(qr)
    with pytest.raises(service_db.ServiceDataError, match="不属于当前顾客"):
        await service_db.preview_review_qr(token, customer.id + 1)

    claim_results = await asyncio.gather(
        service_db.claim_review_qr(token, customer.id),
        service_db.claim_review_qr(token, customer.id),
        return_exceptions=True,
    )
    successful_claims = [item for item in claim_results if isinstance(item, tuple)]
    failed_claims = [item for item in claim_results if isinstance(item, Exception)]
    assert len(successful_claims) == 1
    assert len(failed_claims) == 1
    claimed, completed = successful_claims[0]
    assert claimed["status"] == "claimed"
    assert completed["status"] == "completed"
    assert "已使用" in str(failed_claims[0])
    resumable = await service_db.preview_review_qr(token, customer.id)
    assert resumable["status"] == "claimed"

    verified_review = await service_db.submit_review(
        customer=customer,
        technician_id=profile["user_id"],
        source="qr_verified",
        answers={
            "overall_experience": "excellent",
            "service_highlights": ["professional", "communication"],
            "comment": "已完成服务单评价",
        },
        anonymous=False,
        qr_id=claimed["qr_id"],
    )
    assert verified_review["source"] == "qr_verified"
    assert verified_review["order_id"] == order["order_id"]
    with pytest.raises(service_db.ServiceDataError, match="无效"):
        await service_db.preview_review_qr(token, customer.id)
    with pytest.raises(service_db.ServiceDataError, match="已经提交评价"):
        await service_db.submit_review(
            customer=customer,
            technician_id=profile["user_id"],
            source="qr_verified",
            answers={
                "overall_experience": "excellent",
                "service_highlights": ["professional"],
            },
            anonymous=True,
            qr_id=claimed["qr_id"],
        )

    address_after = await app_modules.wbb.control_db.service_order_addresses.find_one(
        {"order_id": order["order_id"]}
    )
    assert address_after["redact_after"] - completed["closed_at"] == timedelta(days=7)
    await app_modules.wbb.control_db.service_order_addresses.update_one(
        {"order_id": order["order_id"]},
        {"$set": {"redact_after": service_db.utc_now() - timedelta(seconds=1)}},
    )
    assert await service_db.redact_expired_addresses() == 1
    redacted_order = await app_modules.wbb.control_db.service_orders.find_one(
        {"order_id": order["order_id"]}
    )
    redacted_address = await app_modules.wbb.control_db.service_order_addresses.find_one(
        {"order_id": order["order_id"]}
    )
    assert redacted_order["address_redacted"] is True
    assert "distance_meters" not in redacted_order
    assert "encrypted_payload" not in redacted_address


async def test_state_machine_cancellation_and_template_snapshot(app_modules):
    service_db = app_modules.load("wbb.utils.dbservice")
    profile = await _technician(app_modules, service_db)
    customer = await _accepted_customer(service_db)

    template = await service_db.create_package_template(
        {"template_name": "护理模板", "package": _package(name="模板初版")},
        actor_id="admin",
    )
    selected = await service_db.instantiate_package_template(template["template_id"])
    assert selected["source_template_version"] == 1
    await service_db.update_package_template(
        template["template_id"],
        {"template_name": "护理模板", "package": _package(name="模板新版")},
        actor_id="admin",
    )
    assert selected["name"] == "模板初版"

    order = await _order(service_db, customer, profile, service_mode="at_store")
    assert order["distance_meters"] > 0
    assert order["distance_band"] == "1 公里内"
    with pytest.raises(service_db.ServiceDataError, match="只有已确认"):
        await service_db.start_service_order(order["order_id"], profile["user_id"])
    canceled = await service_db.cancel_service_order(
        order["order_id"], customer.id, "顾客计划调整"
    )
    assert canceled["status"] == "canceled_customer"
    with pytest.raises(service_db.ServiceDataError, match="不能取消"):
        await service_db.cancel_service_order(
            order["order_id"], profile["user_id"], "重复取消"
        )

    pending_order = await _order(service_db, customer, profile)
    await service_db.submit_service_quote(
        pending_order["order_id"],
        profile["user_id"],
        {"base_amount": "100", "travel_fee": "0", "discount": "0"},
    )
    await service_db.confirm_service_quote(pending_order["order_id"], customer.id)
    await service_db.start_service_order(pending_order["order_id"], profile["user_id"])
    _, pending_token = await service_db.issue_review_qr(
        pending_order["order_id"], profile["user_id"]
    )
    canceled_pending = await service_db.cancel_service_order(
        pending_order["order_id"], profile["user_id"], "服务现场无法继续"
    )
    assert canceled_pending["status"] == "canceled_technician"
    with pytest.raises(service_db.ServiceDataError, match="无效"):
        await service_db.preview_review_qr(pending_token, customer.id)

    disputed_order = await _order(service_db, customer, profile)
    disputed = await service_db.dispute_service_order(
        disputed_order["order_id"], customer.id, "需要管理员协调"
    )
    assert disputed["status"] == "disputed"
    resumed = await service_db.resolve_service_dispute(
        disputed_order["order_id"],
        actor_id="admin",
        action="resume",
        reason="双方已达成一致",
    )
    assert resumed["status"] == "requested"


async def test_order_requires_encryption_before_persisting(app_modules):
    service_db = app_modules.load("wbb.utils.dbservice")
    profile = await _technician(app_modules, service_db)
    customer = await _accepted_customer(service_db)
    original_key = app_modules.wbb.SERVICE_ADDRESS_ENCRYPTION_KEY
    app_modules.wbb.SERVICE_ADDRESS_ENCRYPTION_KEY = ""
    try:
        with pytest.raises(service_db.ServiceDataError, match="加密密钥"):
            await _order(service_db, customer, profile)
    finally:
        app_modules.wbb.SERVICE_ADDRESS_ENCRYPTION_KEY = original_key
    assert (
        await app_modules.wbb.control_db.service_orders.count_documents(
            {"customer_id": customer.id}
        )
        == 0
    )


async def test_expired_review_qr_reopens_order_for_reissue(app_modules):
    service_db = app_modules.load("wbb.utils.dbservice")
    profile = await _technician(app_modules, service_db)
    customer = await _accepted_customer(service_db)
    order = await _order(service_db, customer, profile)
    await service_db.submit_service_quote(
        order["order_id"],
        profile["user_id"],
        {"base_amount": "100", "travel_fee": "0", "discount": "0"},
    )
    await service_db.confirm_service_quote(order["order_id"], customer.id)
    await service_db.start_service_order(order["order_id"], profile["user_id"])
    first_qr, first_token = await service_db.issue_review_qr(
        order["order_id"], profile["user_id"]
    )
    await app_modules.wbb.control_db.teacher_review_qr_records.update_one(
        {"qr_id": first_qr["qr_id"]},
        {"$set": {"expires_at": service_db.utc_now() - timedelta(seconds=1)}},
    )

    with pytest.raises(service_db.ServiceDataError, match="已过期"):
        await service_db.preview_review_qr(first_token, customer.id)
    reopened = await app_modules.wbb.control_db.service_orders.find_one(
        {"order_id": order["order_id"]}
    )
    assert reopened["status"] == "in_progress"
    assert "current_review_qr_id" not in reopened

    replacement, replacement_token = await service_db.issue_review_qr(
        order["order_id"], profile["user_id"]
    )
    assert replacement["qr_id"] != first_qr["qr_id"]
    assert replacement_token != first_token


async def test_reviews_same_weight_anonymous_and_bayesian_leaderboard(app_modules):
    service_db = app_modules.load("wbb.utils.dbservice")
    profile = await _technician(app_modules, service_db)
    customer = await _accepted_customer(service_db)
    answers = {
        "overall_experience": "excellent",
        "service_highlights": ["professional", "communication"],
        "comment": "服务很专业",
    }

    template = await service_db.get_active_review_template()
    draft = await service_db.begin_review_draft(
        customer_id=customer.id,
        technician_id=profile["user_id"],
        source="student_initiated",
        template=template,
    )
    await service_db.save_review_draft_progress(
        customer.id,
        answers={"overall_experience": "excellent"},
        question_index=1,
    )
    resumed_draft = await service_db.begin_review_draft(
        customer_id=customer.id,
        technician_id=profile["user_id"],
        source="student_initiated",
        template=template,
    )
    assert resumed_draft["draft_id"] == draft["draft_id"]
    assert resumed_draft["answers"] == {"overall_experience": "excellent"}
    await service_db.delete_review_draft(customer.id)

    review_ids = []
    for _ in range(1):
        review = await service_db.submit_review(
            customer=customer,
            technician_id=profile["user_id"],
            source="student_initiated",
            answers=answers,
            anonymous=True,
        )
        review_ids.append(review["review_id"])
        assert "qr_id" not in review
        assert review["package_snapshot"]["name"] == "技师服务"
        assert review["category"] == "护理"
    assert len(set(review_ids)) == 1
    for review_id in review_ids:
        await service_db.moderate_review(
            review_id, "approve", actor_id="admin"
        )

    public, total = await service_db.list_public_reviews(profile["user_id"])
    assert total == 1
    assert all(item["customer_name"] == "匿名顾客" for item in public)
    assert all(item["source_label"] == "用户主动评价" for item in public)
    summary = await service_db.get_technician_review_summary(profile["user_id"])
    assert summary["review_count"] == 1

    pending, _ = await service_db.list_reviews(
        status="approved", source="student_initiated"
    )
    assert all(item["similar_content_count"] == 1 for item in pending)

    leaderboard, total = await service_db.list_leaderboard(category="护理")
    assert total == 1
    assert leaderboard[0]["review_count"] == 1
    assert leaderboard[0]["eligible"] is True
    assert leaderboard[0]["rank_score"] == pytest.approx(
        leaderboard[0]["average_score"]
    )


async def test_direct_review_skips_package_selection(app_modules, monkeypatch):
    service_db = app_modules.load("wbb.utils.dbservice")
    service_module = app_modules.load("wbb.modules.technician_services")
    profile = await _technician(app_modules, service_db)
    customer = await _accepted_customer(service_db)
    captured = {}

    async def fake_begin_review(target, **values):
        captured.update(values)
        captured["customer_id"] = target.from_user.id

    monkeypatch.setattr(service_module, "_begin_review", fake_begin_review)
    await service_module._choose_review_package(
        SimpleNamespace(from_user=customer),
        profile["user_id"],
    )

    assert captured == {
        "customer_id": customer.id,
        "technician_id": profile["user_id"],
        "source": "student_initiated",
    }


async def test_review_moderation_only_allows_defined_state_transitions(app_modules):
    service_db = app_modules.load("wbb.utils.dbservice")
    profile = await _technician(app_modules, service_db)
    customer = await _accepted_customer(service_db)
    package_id = profile["service_profile"]["packages"][0]["package_id"]
    review = await service_db.submit_review(
        customer=customer,
        technician_id=profile["user_id"],
        source="student_initiated",
        answers={
            "overall_experience": "excellent",
            "service_highlights": ["professional"],
        },
        anonymous=True,
        package_id=package_id,
    )
    approved = await service_db.moderate_review(
        review["review_id"], "approve", actor_id="admin"
    )
    assert approved["status"] == "approved"
    with pytest.raises(service_db.ServiceDataError, match="不能执行"):
        await service_db.moderate_review(
            review["review_id"], "reject", actor_id="admin", reason="重复处理"
        )
    voided = await service_db.moderate_review(
        review["review_id"], "void", actor_id="admin", reason="争议作废"
    )
    assert voided["status"] == "voided"
    with pytest.raises(service_db.ServiceDataError, match="不能执行"):
        await service_db.moderate_review(
            review["review_id"], "approve", actor_id="admin"
        )


async def test_first_approved_review_creates_topic_and_is_published_once(
    app_modules,
    monkeypatch,
):
    service_db = app_modules.load("wbb.utils.dbservice")
    review_service = app_modules.load("wbb.services.technician_reviews")
    profile = await _technician(app_modules, service_db)
    customer = await _accepted_customer(service_db)
    package_id = profile["service_profile"]["packages"][0]["package_id"]
    await service_db.set_service_settings(
        {
            "review_forum_chat_id": "-1001234567890",
            "review_forum_username": "public_reviews",
        }
    )
    calls = []

    async def fake_bot_api(_token, method, payload):
        calls.append((method, payload))
        if method == "createForumTopic":
            return {"message_thread_id": 321}
        return {"message_id": 900 + len(calls)}

    monkeypatch.setattr(review_service, "_bot_api", fake_bot_api)
    review = await service_db.submit_review(
        customer=customer,
        technician_id=profile["user_id"],
        source="student_initiated",
        answers={
            "overall_experience": "excellent",
            "service_highlights": ["professional"],
            "comment": "服务很专业",
        },
        anonymous=True,
        package_id=package_id,
    )
    approved = await service_db.moderate_review(
        review["review_id"],
        "approve",
        actor_id="admin",
    )
    published = await review_service.publish_approved_review(
        "123456:test-token",
        approved,
    )
    assert published["status"] == "published"
    assert published["topic_url"] == "https://t.me/public_reviews/321"
    assert [method for method, _ in calls] == [
        "createForumTopic",
        "sendMessage",
        "sendMessage",
    ]

    repeated = await review_service.publish_approved_review(
        "123456:test-token",
        approved,
    )
    assert repeated["status"] == "published"
    assert len(calls) == 3


async def test_technician_requires_active_managed_group_membership(app_modules):
    service_db = app_modules.load("wbb.utils.dbservice")
    profile = await _technician(app_modules, service_db)
    await app_modules.wbb.control_db.directory_memberships.update_many(
        {"user_id": profile["user_id"]}, {"$set": {"active": False}}
    )
    customer = await _accepted_customer(service_db)
    with pytest.raises(service_db.ServiceDataError, match="当前不满足接单资格"):
        await _order(service_db, customer, profile)


async def test_technician_uses_public_area_without_precise_location(app_modules):
    service_db = app_modules.load("wbb.utils.dbservice")
    profile = await _technician(app_modules, service_db)
    await app_modules.wbb.control_db.directory_locations.delete_one(
        {"user_id": profile["user_id"]}
    )
    republished = await service_db.publish_technician_profile(
        profile["user_id"],
        profile["service_profile"],
        actor_id=profile["user_id"],
    )
    assert republished["service_profile"]["public_area_text"] == "北京朝阳区附近"

    technicians, total = await service_db.list_available_technicians()
    assert total == 1
    assert technicians[0]["public_area_text"] == "北京朝阳区附近"
    assert technicians[0]["distance_band"] == ""


async def test_service_admin_api_settings_templates_and_permissions(app_modules):
    admin_api = app_modules.load("wbb.admin.api")
    app_modules.wbb.BOT_PERMISSIONS = {"teacher_directory.manage"}
    application = admin_api.build_admin_application()
    await application["admin_api"].initialize()
    client = TestClient(TestServer(application), cookie_jar=CookieJar(unsafe=True))
    await client.start_server()
    try:
        login = await client.post(
            "/api/admin/v1/auth/login",
            json={"username": "admin", "password": "qwe0.123456"},
        )
        login_data = (await login.json())["data"]
        changed = await client.put(
            "/api/admin/v1/auth/password",
            headers={"X-CSRF-Token": login_data["csrf_token"]},
            json={
                "current_password": "qwe0.123456",
                "new_password": "changed-pass-123",
            },
        )
        headers = {"X-CSRF-Token": (await changed.json())["data"]["csrf_token"]}

        unconfirmed = await client.put(
            "/api/admin/v1/directory/service-settings",
            headers=headers,
            json={"customer_max_open_orders": 4},
        )
        assert unconfirmed.status == 409
        saved = await client.put(
            "/api/admin/v1/directory/service-settings",
            headers=headers,
            json={
                "customer_max_open_orders": 4,
                "customer_daily_request_limit": 12,
                "review_qr_expiry_hours": 48,
                "confirm": True,
            },
        )
        assert saved.status == 200
        assert (await saved.json())["data"]["review_qr_expiry_hours"] == 48

        created = await client.post(
            "/api/admin/v1/directory/package-templates",
            headers=headers,
            json={
                "template_name": "后台护理模板",
                "package": _package(),
                "confirm": True,
            },
        )
        assert created.status == 201
        created_data = (await created.json())["data"]
        assert created_data["version"] == 1
        duplicated = await client.post(
            f"/api/admin/v1/directory/package-templates/{created_data['template_id']}/actions",
            headers=headers,
            json={"action": "duplicate", "confirm": True},
        )
        assert duplicated.status == 200
        assert (await duplicated.json())["data"]["status"] == "disabled"

        review_template = await client.get(
            "/api/admin/v1/directory/review-template"
        )
        assert review_template.status == 200
        template_data = (await review_template.json())["data"]
        assert {item["type"] for item in template_data["questions"]} == {
            "single_choice",
            "multiple_choice",
            "text",
        }
        assert sum(
            item.get("weight", 0)
            for item in template_data["questions"]
            if item["type"] == "single_choice"
        ) == 100

        metrics = await client.get(
            "/api/admin/v1/directory/fulfillment-metrics"
        )
        assert metrics.status == 200
        assert (await metrics.json())["data"]["scope_note"].startswith(
            "仅统计平台服务单"
        )

        app_modules.wbb.BOT_PERMISSIONS = set()
        denied = await client.get(
            "/api/admin/v1/directory/fulfillment-metrics"
        )
        assert denied.status == 403
    finally:
        await client.close()