"""深度诊断（tier=full）链路端到端：派生 → full_report 日配额 → full 流水线 → 拒绝路径。

与 test_report_flow 同一套基建：Fake 阶段经 ``REPORT_PIPELINE_STAGES_JSON`` /
``REPORT_FULL_PIPELINE_STAGES_JSON`` 动态装配（worker 按 report.tier 分流两
条阶段清单，两份 env 要同时给）；不启真 celery，worker 内芯直调。
"""

from __future__ import annotations

import json
import os
import uuid
from collections.abc import AsyncIterator, Iterator
from contextlib import contextmanager
from typing import Any

import pytest


def _infra_available() -> bool:
    return bool(os.getenv("MEIDA_RUN_INTEGRATION"))


pytestmark = [
    pytest.mark.integration,
    pytest.mark.skipif(
        not _infra_available(),
        reason="set MEIDA_RUN_INTEGRATION=1 with PostgreSQL and Redis running",
    ),
]

_MODULE = "tests.integration.test_full_report_flow"
_BASIC_STAGES = json.dumps([f"{_MODULE}:FakeVisionStage", f"{_MODULE}:FakeDiagnosisStage"])
_FULL_STAGES = json.dumps([f"{_MODULE}:FakeFullDiagnosisStage", f"{_MODULE}:FakeFullCopyStage"])


class FakeVisionStage:
    """与 test_report_flow 同一契约：derived_features 非空即跳过。"""

    name = "vision"

    async def run(self, ctx: Any) -> None:
        features = ctx.report_input.derived_features
        if not features:
            features["vision_runs"] = 1
        ctx.report.result["vision"] = {"features": sorted(features)}


class FakeDiagnosisStage:
    name = "diagnosis"

    async def run(self, ctx: Any) -> None:
        assert "vision" in ctx.report.result
        ctx.report.result["diagnosis"] = {"dimensions": [], "coverage": {"total": 32}}


class FakeFullDiagnosisStage:
    """full 流水线的诊断阶段替身：前面没有 vision 阶段（full 不含视觉）。"""

    name = "diagnosis"

    async def run(self, ctx: Any) -> None:
        assert "vision" not in ctx.report.result
        ctx.report.result["diagnosis"] = {"dimensions": [], "coverage": {"total": 32}}


class FakeFullCopyStage:
    """full 文案阶段替身：copy 与 full 双写（结构与真实 FullCopyStage 一致）。"""

    name = "copy"

    async def run(self, ctx: Any) -> None:
        assert "diagnosis" in ctx.report.result
        ctx.report.result["copy"] = {
            "summary": "深度总评",
            "sections": [{"title": "风格", "paragraphs": ["段落"]}],
            "model": "fake",
            "prompt_version": "v1",
            "dropped_keys": [],
        }
        ctx.report.result["full"] = {
            "season": {
                "code": "summer",
                "detailed": "柔夏",
                "name_cn": "柔夏型",
                "analysis": "解析文案",
            },
            "palette": {
                "best": [{"name": "雾霾蓝", "hex": "#8FA8BF"}],
                "neutral": [{"name": "浅灰", "hex": "#D9D9D9"}],
                "avoid": [{"name": "明黄", "hex": "#FFD93B"}],
            },
            "scenes": [
                {
                    "key": "commute",
                    "title": "日常通勤",
                    "outfits": [{"name": "通勤 A", "points": ["要点一"]}],
                }
            ],
            "plan30": [{"week": 1, "title": "第 1 周", "tasks": ["任务一"]}],
        }


# --- fixtures ---------------------------------------------------------------


@pytest.fixture
def no_enqueue(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr("apps.api.routers.reports._enqueue_report", lambda task_id: None)


@contextmanager
def _stage_config(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
    from packages.shared.config import get_settings

    monkeypatch.setenv("REPORT_PIPELINE_STAGES_JSON", _BASIC_STAGES)
    monkeypatch.setenv("REPORT_FULL_PIPELINE_STAGES_JSON", _FULL_STAGES)
    get_settings.cache_clear()
    try:
        yield
    finally:
        monkeypatch.undo()
        get_settings.cache_clear()


@pytest.fixture
def fake_stages(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
    with _stage_config(monkeypatch):
        yield


@pytest.fixture
async def client(no_enqueue: None) -> AsyncIterator[Any]:
    import httpx

    from apps.api.main import create_app
    from packages.shared.config import get_settings

    app = create_app(get_settings())
    transport = httpx.ASGITransport(app=app)
    async with (
        httpx.AsyncClient(transport=transport, base_url="http://test") as http,
        app.router.lifespan_context(app),
    ):
        yield http


# --- helpers ----------------------------------------------------------------


async def _login(client: Any) -> tuple[dict[str, str], int]:
    resp = await client.post(
        "/api/v1/auth/dev-token", json={"device_key": f"full-it-{uuid.uuid4().hex[:8]}"}
    )
    assert resp.json()["code"] == 0
    data = resp.json()["data"]
    return {"Authorization": f"Bearer {data['access_token']}"}, int(data["user_id"])


async def _upload_photo(client: Any, headers: dict[str, str]) -> str:
    resp = await client.post(
        "/api/v1/uploads/direct?scene=report_photo",
        headers={**headers, "Content-Type": "image/jpeg"},
        content=b"\xff\xd8\xff\xe0" + b"\x00" * 64,
    )
    assert resp.status_code == 200
    return str(resp.json()["data"]["object_key"])


async def _create_succeeded_basic(client: Any, headers: dict[str, str]) -> dict[str, Any]:
    from apps.worker.report_tasks import _generate_report

    resp = await client.post(
        "/api/v1/reports",
        headers=headers,
        json={
            "questionnaire": {"style": "通勤", "height_cm": 168},
            "photo_object_key": await _upload_photo(client, headers),
            "request_key": f"rk-{uuid.uuid4().hex}",
        },
    )
    assert resp.status_code == 200
    body = resp.json()["data"]
    await _generate_report(int(body["task_id"]))
    return body


async def _derive_full(
    client: Any, headers: dict[str, str], report_id: str, *, request_key: str | None = None
) -> Any:
    return await client.post(
        f"/api/v1/reports/{report_id}/full",
        headers=headers,
        json={"request_key": request_key or f"rk-full-{uuid.uuid4().hex}"},
    )


async def _fetch_report(report_id: int) -> Any:
    from sqlalchemy import select

    from packages.shared.db import session_scope
    from packages.shared.models import RpReport

    async with session_scope() as session:
        result = await session.execute(select(RpReport).where(RpReport.report_id == report_id))
        return result.scalar_one_or_none()


async def _fetch_input(report_id: int) -> Any:
    from sqlalchemy import select

    from packages.shared.db import session_scope
    from packages.shared.models import RpInput

    async with session_scope() as session:
        result = await session.execute(select(RpInput).where(RpInput.report_id == report_id))
        return result.scalar_one_or_none()


async def _ledger_rows(user_id: int, quota_key: str) -> list[Any]:
    from sqlalchemy import select

    from packages.shared.db import session_scope
    from packages.shared.models import Ledger

    async with session_scope() as session:
        result = await session.execute(
            select(Ledger)
            .where(Ledger.user_id == user_id, Ledger.quota_key == quota_key)
            .order_by(Ledger.created_at.asc(), Ledger.tx_id.asc())
        )
        return list(result.scalars().all())


# --- 场景 1：派生成功 + full 流水线 + 幂等重放 ----------------------------------


async def test_derive_full_success_runs_full_pipeline(fake_stages: None, client: Any) -> None:
    from apps.worker.report_tasks import _generate_report

    headers, user_id = await _login(client)
    basic = await _create_succeeded_basic(client, headers)

    request_key = f"rk-full-{uuid.uuid4().hex}"
    resp = await _derive_full(client, headers, basic["report_id"], request_key=request_key)
    assert resp.status_code == 200
    data = resp.json()["data"]
    full_report_id = int(data["report_id"])
    assert full_report_id != int(basic["report_id"])
    assert data["version"] == 1  # full 是独立谱系，版本从 1 起算

    # full_report 日配额预扣（period_key d:YYYY-MM-DD）。
    rows = await _ledger_rows(user_id, "full_report")
    assert len(rows) == 1
    assert rows[0].status == "reserved"
    assert rows[0].period_key.startswith("d:")

    # 输入三件套复用（derived_features 带着 basic 的视觉中间态，无视觉阶段）。
    full_input = await _fetch_input(full_report_id)
    assert full_input.derived_features == {"vision_runs": 1}

    await _generate_report(int(data["task_id"]))
    report = await _fetch_report(full_report_id)
    assert report.status == "succeeded"
    assert report.tier == "full"
    # full 流水线：无 vision 键，copy + full + diagnosis 齐全。
    assert set(report.result) == {"diagnosis", "copy", "full"}
    full = report.result["full"]
    assert full["season"]["name_cn"] == "柔夏型"
    assert full["palette"]["best"][0]["hex"] == "#8FA8BF"
    assert full["scenes"][0]["outfits"][0]["points"]
    assert full["plan30"][0]["week"] == 1

    rows = await _ledger_rows(user_id, "full_report")
    assert [row.status for row in rows] == ["committed"]

    # 详情视图透出 tier 与 full 结构（App 凭 tier/full 渲染深度区块）。
    detail = (await client.get(f"/api/v1/reports/{full_report_id}", headers=headers)).json()["data"]
    assert detail["tier"] == "full"
    assert detail["result"]["full"]["season"]["detailed"] == "柔夏"

    # 幂等重放：同 request_key 返回同任务同报告，不二次预扣。
    replay = await _derive_full(client, headers, basic["report_id"], request_key=request_key)
    assert replay.status_code == 200
    assert replay.json()["data"]["report_id"] == str(full_report_id)
    assert len(await _ledger_rows(user_id, "full_report")) == 1


# --- 场景 2：非 SUCCEEDED 源报告拒绝 --------------------------------------------


async def test_derive_full_rejects_pending_source(fake_stages: None, client: Any) -> None:
    headers, _user_id = await _login(client)
    resp = await client.post(
        "/api/v1/reports",
        headers=headers,
        json={
            "questionnaire": {"style": "通勤"},
            "photo_object_key": await _upload_photo(client, headers),
            "request_key": f"rk-{uuid.uuid4().hex}",
        },
    )
    assert resp.status_code == 200
    basic = resp.json()["data"]  # pending，不跑 worker

    derived = await _derive_full(client, headers, basic["report_id"])
    assert derived.status_code == 409
    assert derived.json()["code"] == 40900  # STATE_CONFLICT


# --- 场景 3：full 不能再派生 full（防套娃） --------------------------------------


async def test_derive_full_rejects_full_source(fake_stages: None, client: Any) -> None:
    from apps.worker.report_tasks import _generate_report

    headers, _user_id = await _login(client)
    basic = await _create_succeeded_basic(client, headers)
    derived = await _derive_full(client, headers, basic["report_id"])
    assert derived.status_code == 200
    full = derived.json()["data"]
    await _generate_report(int(full["task_id"]))

    nested = await _derive_full(client, headers, full["report_id"])
    assert nested.status_code == 409
    assert nested.json()["code"] == 40900


# --- 场景 4：full_report 配额耗尽 ------------------------------------------------


async def test_derive_full_blocked_when_quota_exhausted(fake_stages: None, client: Any) -> None:
    """开发库 full_report 规则行是测试期放宽值（base_limit=10）；把本用例用户的
    bl_quota 快照 limit_value 收敛到已用量，确定性断言 40301（同 test_report_flow
    场景 4 的做法，只动本用户的快照行，不动全局规则）。"""
    from sqlalchemy import update

    from packages.shared.db import session_scope
    from packages.shared.models import Quota

    headers, user_id = await _login(client)
    basic = await _create_succeeded_basic(client, headers)

    first = await _derive_full(client, headers, basic["report_id"])
    assert first.status_code == 200  # 首次派生占掉 1 次

    async with session_scope() as session:
        await session.execute(
            update(Quota)
            .where(Quota.user_id == user_id, Quota.quota_key == "full_report")
            .values(limit_value=1)
        )

    second = await _derive_full(client, headers, basic["report_id"])
    assert second.status_code == 403
    assert second.json()["code"] == 40301  # QUOTA_EXHAUSTED