"""Merchant task parsing, confirmation, grounded fields and immutable scope."""
from __future__ import annotations

import asyncio
import copy
from dataclasses import replace
import json
import sqlite3

import httpx
import pytest
from fastapi import HTTPException
from pydantic import ValidationError

from apps.server.business_briefs import BusinessBriefConfirmInput, BusinessBriefInput, BusinessBriefService, checked_conditions
from apps.server.db import Store, StoreConflict, encoded
from apps.server.model_client import ModelCallError, ModelClient
from apps.server.tests.test_analysis import PRIMARY, FALLBACK, config, wire
from apps.server.tests.test_learning import client, initialize
from apps.server.tests.test_postgres import pg_client

TEXT = "在小红书帮我找杭州有装修需求的业主，我提供装修服务，搜索准备装修，不要服务推销。"


def fact(value, evidence=None):
    return {"value": value, "evidence": evidence or value}


def answer():
    return {"platform": fact("小红书"), "service": fact("装修服务"), "target_customer": fact("有装修需求的业主"),
            "search_terms": [fact("准备装修")], "region": fact("杭州"), "region_scope": None,
            "exclude_service_promotions": fact("不要服务推销")}


def confirmed(**changes):
    return BusinessBriefConfirmInput(**({"platform": "xiaohongshu", "service": "装修服务", "target_customer": "有装修需求的业主",
                                        "search_terms": ["准备装修"], "region": None, "region_scope": "unspecified", "exclude_service_promotions": True} | changes))


@pytest.fixture(params=["client", "pg_client"])
def brief_client(request):
    return request.getfixturevalue(request.param)


def setup(client, handler):
    initialize(client)
    store = client.app_reference.state.store
    owner = store.one("SELECT id FROM operators")["id"]
    service = BusinessBriefService(store, config())
    service.client = ModelClient(config(), http_transport=httpx.MockTransport(handler))
    return service, owner


def test_business_brief_grounded_parse_immutable_confirm_reuse_and_ownership(brief_client):
    requests = []
    def handler(request):
        payload = json.loads(request.content)
        user = json.loads(payload["messages"][1]["content"])
        assert user == {"request_text": TEXT}
        assert "author" not in user and request.headers["authorization"] == "Bearer " + PRIMARY
        requests.append(user)
        return wire(answer())
    service, owner = setup(brief_client, handler)
    draft = asyncio.run(service.create(TEXT, owner))
    assert draft["kind"] == "parse" and draft["status"] == "draft" and draft["parent_id"] is None and draft["route"] is None
    assert draft["conditions"] == confirmed(region="杭州", region_scope="specified").model_dump()
    assert draft["field_evidence"]["region"] == ["杭州"] and draft["condition_sources"]["service"] == "request_text"
    assert draft["usage"]["total_tokens"] == 140 and draft["cost"]["status"] == "unknown"
    service.client.config = replace(service.client.config, keys=(("primary", "rotated-fixture-key"),))
    assert asyncio.run(service.create(TEXT, owner)) == draft and len(requests) == 1
    original = service.get(draft["id"], owner)
    final = service.confirm(draft["id"], confirmed(), owner)
    assert final["status"] == final["kind"] == "confirmed" and final["parent_id"] == draft["id"] and final["parent_digest"] == draft["digest"]
    assert final["conditions"]["region_scope"] == "unspecified" and final["conditions"]["region"] is None and "region" in final["missing_fields"]
    assert set(final["condition_sources"].values()) == {"maintainer_edit"} and not any(final["field_evidence"].values())
    assert final["attempts"] == [] and final["usage"] is None and final["route"]["execution_started"] is False
    assert all(item["execution_available"] is False for item in final["route"]["follow_up"])
    assert service.get(draft["id"], owner) == original and service.get_confirmed(final["id"], owner) == final
    assert service.list(owner) == [final, draft] and len(requests) == 1
    with pytest.raises(HTTPException) as error:
        service.get_confirmed(draft["id"], owner)
    assert error.value.detail["code"] == "BRIEF_NOT_CONFIRMED"
    for record in (draft, final):
        with pytest.raises(HTTPException) as error:
            service.get_confirmed(record["id"], "other-owner")
        assert error.value.status_code == 404
    expected = StoreConflict if service.store.backend == "postgresql" else sqlite3.IntegrityError
    for query in ("UPDATE business_briefs SET data_json='{}' WHERE id=$1", "DELETE FROM business_briefs WHERE id=$1"):
        with pytest.raises(expected):
            service.store.execute(query, (final["id"],))
    if service.store.backend == "postgresql":
        with pytest.raises(StoreConflict):
            service.store.execute("TRUNCATE business_briefs CASCADE")
    else:
        assert BusinessBriefService(Store(service.store.home), config()).get(final["id"], owner) == final
    assert service.store.one("SELECT version FROM schema_versions WHERE version=8")["version"] == 8
    assert PRIMARY not in encoded(final) and FALLBACK not in encoded(draft)


def test_business_brief_missing_unsupported_and_failed_are_distinct(brief_client):
    mode = {"answer": {field: [] if field == "search_terms" else None for field in answer()}}
    service, owner = setup(brief_client, lambda request: wire(mode["answer"]))
    incomplete = asyncio.run(service.create("帮我找客户", owner))
    assert incomplete["status"] == "needs_clarification" and incomplete["conditions"]["service"] is None
    assert set(incomplete["missing_fields"]) == {"platform", "service", "target_customer", "search_terms", "region", "exclude_service_promotions"}
    assert service.confirm(incomplete["id"], confirmed(), owner)["status"] == "confirmed"
    mode["answer"]["platform"] = fact("抖音")
    mode["answer"]["service"] = fact("美甲服务")
    unsupported = asyncio.run(service.create("抖音寻找美甲服务客户", owner))
    assert unsupported["status"] == "unsupported" and unsupported["conditions"]["service"] == "美甲服务"
    with pytest.raises(HTTPException) as error:
        service.confirm(unsupported["id"], confirmed(), owner)
    assert error.value.detail["code"] == "BRIEF_UNCONFIRMABLE"
    mode["answer"]["service"] = fact("装修服务", "原文没有的依据")
    failed = asyncio.run(service.create("抖音寻找美甲服务客户", owner))
    # Same parsed unsupported input/config is cached; a failed sample needs new input.
    assert failed == unsupported
    failed = asyncio.run(service.create("抖音寻找美甲服务客户。", owner))
    assert failed["status"] == "failed" and failed["error_codes"] == ["MODEL_OUTPUT_INVALID"]
    assert failed["conditions"]["service"] is None and "原文没有" not in encoded(failed)
    with pytest.raises(HTTPException):
        service.confirm(failed["id"], confirmed(), owner)


def test_business_brief_grounding_strict_and_no_invented_defaults():
    valid = answer()
    result = checked_conditions(encoded(valid), TEXT)
    assert result[-1] == "draft"
    variants = [lambda v: v.update(extra="do not persist"), lambda v: v.update(platform=fact("抖音")),
                lambda v: v.update(region=fact("北京", "杭州")), lambda v: v.update(region_scope=fact("全国")),
                lambda v: v.update(search_terms=[fact("伪造搜索词")]), lambda v: v.update(exclude_service_promotions=fact("服务推销")),
                lambda v: v.update(service={"value": "装修服务", "evidence": "装修服务", "extra": True})]
    for mutate in variants:
        changed = copy.deepcopy(valid)
        mutate(changed)
        with pytest.raises(ModelCallError) as error:
            checked_conditions(encoded(changed), TEXT)
        assert error.value.code == "MODEL_OUTPUT_INVALID"
    with pytest.raises(ModelCallError):
        checked_conditions('{"platform":null,"platform":null}', TEXT)
    blank = {field: [] if field == "search_terms" else None for field in valid}
    conditions, _, missing, _, status = checked_conditions(encoded(blank), "装修服务")
    assert status == "needs_clarification" and conditions["platform"] is None and conditions["service"] is None and conditions["region_scope"] == "unspecified"
    assert "region" in missing


def test_business_brief_confirmation_is_closed_explicit_and_region_not_nationwide():
    valid = confirmed().model_dump()
    invalid = [{key: value for key, value in valid.items() if key != field} for field in valid]
    invalid += [valid | update for update in ({"platform": "douyin"}, {"service": "美甲服务"}, {"search_terms": ["装修"] * 4}, {"search_terms": ["装修", "装修"]},
                 {"exclude_service_promotions": 1}, {"region": "杭州"}, {"region_scope": "specified"}, {"region_scope": "unrestricted", "region": "全国"},
                 {"region_scope": "specified", "region": "全国"}, {"search_terms": ["13800138000"]}, {"unexpected": True})]
    for value in invalid:
        with pytest.raises((ValidationError, HTTPException)):
            BusinessBriefConfirmInput(**value)
    assert confirmed(region_scope="unrestricted").region is None
    assert confirmed(region_scope="specified", region="杭州").region == "杭州"


def test_business_brief_sensitive_input_redacted_before_model_and_failed_fallback_retained(client):
    calls = []
    def handler(request):
        calls.append(request)
        assert "13800138000" not in request.content.decode() and "owner@example.test" not in request.content.decode()
        if len(calls) == 1:
            return httpx.Response(401, json={"error": {"message": "provider-secret-sentinel"}})
        return wire({field: [] if field == "search_terms" else None for field in answer()}, usage=False)
    service, owner = setup(client, handler)
    parsed = asyncio.run(service.create("帮我找业主，我的手机13800138000，邮箱owner@example.test", owner))
    assert len(calls) == 2 and [item["slot"] for item in parsed["attempts"]] == ["primary", "fallback"]
    assert parsed["usage"] is None and parsed["cost"]["amount"] is None
    assert "13800138000" not in encoded(parsed) and "owner@example.test" not in encoded(parsed) and "provider-secret" not in encoded(parsed)
    for text in ("key api_key=fixture-not-real", "token authorization:fixture-not-real", "密码123456", "A" * 2001, " "):
        with pytest.raises((HTTPException, ValidationError)):
            asyncio.run(service.create(text, owner))
    assert len(calls) == 2
    service.client.config = replace(service.client.config, keys=())
    with pytest.raises(HTTPException) as error:
        asyncio.run(service.create(TEXT, owner))
    assert error.value.status_code == 503 and len(calls) == 2


def test_business_brief_busy_is_bounded_no_model_call(client):
    calls = []
    service, owner = setup(client, lambda request: calls.append(request) or wire(answer()))
    async def busy():
        await service.lock.acquire()
        try:
            with pytest.raises(HTTPException) as error:
                await service.create(TEXT, owner)
            assert error.value.status_code == 409
        finally:
            service.lock.release()
    asyncio.run(busy())
    assert not calls