lami/lami-base-v1/backend/tests_python/test_09_module_live.py

"""Swagger 模块 9（KOC）— 真实联调（QiWe 同步联系人 + 真实 roomId 意向识别）"""

from __future__ import annotations

import uuid

import pytest

from load_env import load_backend_env

pytestmark = [pytest.mark.integration, pytest.mark.qiwei_live]


@pytest.fixture(scope="module")
def m09_channel(api_session):
    suffix = uuid.uuid4().hex[:6]
    ch = api_session.assert_success(
        api_session.post(
            "/channels",
            json={
                "name": f"联调渠道_{suffix}",
                "code": f"live-ch-{suffix}",
                "type": "offline",
                "remark": "module9-live",
            },
        ),
        status=201,
    )
    yield ch
    api_session.assert_success(api_session.delete(f"/channels/{ch['id']}"))


@pytest.fixture(scope="module")
def m09_contacts(api_session, live_guid):
    data = api_session.assert_success(
        api_session.post(
            "/contacts/sync",
            json={"guid": live_guid, "currentSeq": 0, "limit": 20},
        )
    )
    return {
        "sync": data,
        "contacts": data.get("contacts") or [],
        "user_id": (data.get("contacts") or [{}])[0].get("userId") if data.get("contacts") else None,
    }


@pytest.fixture(scope="module")
def m09_intent(api_session, live_room_id):
    suffix = uuid.uuid4().hex[:8]
    detect = api_session.assert_success(
        api_session.post(
            "/intent-leads/detect",
            json={
                "userId": f"live-user-{suffix}",
                "roomId": live_room_id,
                "content": "这个全屋定制方案多少钱？能约量房看样板间吗？",
            },
        )
    )
    return {"detect": detect, "room_id": live_room_id}


def test_m09_channels_list(api, m09_channel):
    listed = api.assert_success(api.get("/channels"))
    assert any(c["id"] == m09_channel["id"] for c in listed)


def test_m09_update_channel(api, m09_channel):
    updated = api.assert_success(
        api.put(f"/channels/{m09_channel['id']}", json={"remark": "联调已更新"})
    )
    assert updated["remark"] == "联调已更新"


def test_m09_contacts_sync(m09_contacts):
    assert "contacts" in m09_contacts["sync"]
    assert "hasMore" in m09_contacts["sync"]


def test_m09_contacts_list(api, m09_contacts):
    listed = api.assert_success(api.get("/contacts"))
    assert isinstance(listed, list)
    if m09_contacts["contacts"]:
        assert len(listed) >= len(m09_contacts["contacts"])


def test_m09_contact_detail_if_any(api, m09_contacts):
    uid = m09_contacts.get("user_id")
    if not uid:
        pytest.skip("QiWe 同步未返回外部联系人，跳过详情")
    detail = api.assert_success(api.get(f"/contacts/{uid}"))
    assert detail["userId"] == uid


def test_m09_koc_candidates(api, m09_contacts):
    candidates = api.assert_success(api.get("/koc/candidates"))
    assert isinstance(candidates, list)


def test_m09_koc_label_optional(api, live_guid, m09_contacts):
    candidates = api.assert_success(api.get("/koc/candidates"))
    if not candidates:
        pytest.skip("无 KOC 候选人（需先 sync 联系人且随机规则命中）")
    env = load_backend_env()
    label_id = env.get("QIWEI_KOC_LABEL_ID", "").strip() or "1"
    if env.get("QIWEI_LIVE_KOC_LABEL", "").strip() != "1":
        pytest.skip("未开启真实打标签：.env 设置 QIWEI_LIVE_KOC_LABEL=1 与 QIWEI_KOC_LABEL_ID")

    user_id = candidates[0]["userId"]
    resp = api.post(
        "/koc/label",
        json={"userId": user_id, "guid": live_guid, "labelId": label_id},
    )
    body = api.parse_json(resp)
    assert body.get("success") or body.get("error", {}).get("code") == "QIWEI_API_ERROR"


def test_m09_intent_detect(m09_intent):
    d = m09_intent["detect"]
    assert d.get("matched") is True
    assert d.get("lead") is not None


def test_m09_intent_leads_list(api, m09_intent):
    leads = api.assert_success(api.get("/intent-leads"))
    assert isinstance(leads, list)
    if m09_intent["detect"].get("lead"):
        lid = m09_intent["detect"]["lead"]["id"]
        assert any(l["id"] == lid for l in leads)


def test_m09_intent_lead_update_if_any(api, m09_intent):
    lead = m09_intent["detect"].get("lead")
    if not lead:
        pytest.skip("未识别到意向线索")
    updated = api.assert_success(
        api.put(
            f"/intent-leads/{lead['id']}",
            json={"status": "assigned", "assigneeId": 1},
        )
    )
    assert updated["status"] == "assigned"


def test_m09_intent_tasks(api, m09_intent):
    tasks = api.assert_success(api.get("/intent-tasks"))
    assert isinstance(tasks, list)