from __future__ import annotations

import os

import pytest
import pytest_asyncio
from sqlalchemy import JSON, select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

os.environ["DATABASE_URL"] = "postgresql+asyncpg://pmflow:pmflow_pass@127.0.0.1:5435/pmflow"

from app.models.base import Base
from app.models.project import Project
from app.models.project_signal_link import ProjectSignalLink
from app.models.signal import Signal
from app.models.signal_source_registry import SignalSourceRegistry
from app.models.truth_node import TruthNode
from app.models.user import User
from app.api.sstc_api.signals_list import list_signals
from app.services.signal.signal_source_orchestration import SignalSourceOrchestrationService


def _use_sqlite_json_types() -> None:
    for table in (User.__table__, SignalSourceRegistry.__table__, TruthNode.__table__):
        for column in table.columns:
            if column.name in {"assigned_domains", "config", "raw_context", "content"}:
                column.type = JSON()


@pytest_asyncio.fixture
async def db_session():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    _use_sqlite_json_types()
    async with engine.begin() as conn:
        await conn.run_sync(
            lambda sync_conn: Base.metadata.create_all(
                sync_conn,
                tables=[
                    User.__table__,
                    Project.__table__,
                    Signal.__table__,
                    ProjectSignalLink.__table__,
                    SignalSourceRegistry.__table__,
                    TruthNode.__table__,
                ],
            )
        )
    maker = async_sessionmaker(engine, expire_on_commit=False)
    async with maker() as session:
        yield session
    await engine.dispose()


@pytest.mark.asyncio
async def test_materialize_requirement_sources_dedupes_duplicate_items(db_session) -> None:
    user = User(
        tenant_id=1,
        username="owner",
        email="owner@example.com",
        password_hash="hash",
        name="负责人",
        role="ADMIN",
        status="active",
        is_active=True,
    )
    project = Project(
        tenant_id=1,
        name="demo-project",
        title="Demo Project",
        status="active",
        assigned_role="backend",
    )
    signal = Signal(
        tenant_id=1,
        source="AI-Steward",
        content="main signal",
        priority="high",
        processed=True,
    )
    db_session.add_all([user, project, signal])
    await db_session.commit()

    service = SignalSourceOrchestrationService(db_session)
    results = await service.materialize_requirement_sources(
        tenant_id=1,
        project_id=int(project.id),
        signal_id=int(signal.id),
        operator_user_id=int(user.id),
        requested_via="ai_steward",
        origin="project_analysis",
        origin_text="我们要做一个 AI 客服工具，包含内容本地化和社媒运营。",
        default_source_kind="custom",
        requirement_items=[
            {
                "section": "核心痛点",
                "title": "内容本地化",
                "description": "翻译公司贵、慢、不地道",
            },
            {
                "section": "核心痛点",
                "title": "内容本地化",
                "description": "翻译公司贵、慢、不地道",
            },
            {
                "section": "MVP 功能",
                "title": "社媒运营",
                "description": "自动生成社媒文案并排期发布",
            },
        ],
    )

    assert len(results) == 2
    assert results[0]["section"] == "核心痛点"
    assert results[0]["source_kind"] == "custom"
    assert results[0]["origin"] == "project_analysis"

    rows = (await db_session.execute(select(SignalSourceRegistry).order_by(SignalSourceRegistry.id))).scalars().all()
    assert len(rows) == 2
    assert all(row.signal_id is not None for row in rows)
    assert len({int(row.signal_id) for row in rows if row.signal_id is not None}) == 2
    assert rows[0].raw_context["origin"] == "project_analysis"
    assert rows[0].raw_context["requirement_item"]["title"] == "内容本地化"
    assert rows[1].raw_context["requirement_item"]["section"] == "MVP 功能"

    generated_signals = (
        await db_session.execute(
            select(Signal).where(Signal.id != int(signal.id)).order_by(Signal.id.asc())
        )
    ).scalars().all()
    assert len(generated_signals) == 2
    assert generated_signals[0].title == "核心痛点 - 内容本地化"
    assert generated_signals[1].title == "MVP 功能 - 社媒运营"

    links = (
        await db_session.execute(
            select(ProjectSignalLink).where(ProjectSignalLink.project_id == int(project.id)).order_by(ProjectSignalLink.id.asc())
        )
    ).scalars().all()
    assert len(links) == 2
    assert {int(link.signal_id) for link in links} == {int(item.id) for item in generated_signals}


@pytest.mark.asyncio
async def test_list_signals_returns_linked_signal_when_charter_exists(db_session) -> None:
    owner = User(
        tenant_id=1,
        username="sstc-owner",
        email="sstc-owner@example.com",
        password_hash="hash",
        name="负责人",
        role="ADMIN",
        status="active",
        is_active=True,
    )
    project = Project(
        tenant_id=1,
        name="sstc-project",
        title="SSTC Project",
        status="active",
        assigned_role="backend",
        owner_id=1,
    )
    signal = Signal(
        tenant_id=1,
        source="bank_statement_upload",
        content="银行流水导入模块开发",
        impact_score=0.8,
        priority="high",
        processed=False,
        strategic_analysis="{}",
        tasks_generated="[]",
        matched_project_id=int(1),
        title="📊 银行流水导入功能需求",
        signal_type="新需求",
        summary="银行流水导入模块开发",
        reasoning="[]",
        confidence=0.9,
        recommended_action="enter_only",
        status="待处理",
    )
    db_session.add_all([owner, project, signal])
    await db_session.flush()

    project.owner_id = int(owner.id)
    signal.matched_project_id = int(project.id)

    db_session.add(
        ProjectSignalLink(
            tenant_id=1,
            project_id=int(project.id),
            signal_id=int(signal.id),
            link_type="related",
            relevance_score=1.0,
        )
    )
    db_session.add(
        TruthNode(
            tenant_id=1,
            node_type="charter",
            project_id=int(project.id),
            signal_id=int(signal.id),
            created_by=int(owner.id),
            title="银行流水导入宪章",
            content={"signal_ids": [int(signal.id)]},
            version=1,
            status="draft",
        )
    )
    await db_session.commit()

    response = await list_signals(
        project_id=str(project.id),
        status="linked",
        keyword=None,
        source=None,
        priority=None,
        created_from=None,
        created_to=None,
        limit=10,
        offset=0,
        db=db_session,
        current_user=owner,
    )

    payload = response["data"]
    assert payload["total"] == 1
    assert len(payload["items"]) == 1
    assert payload["items"][0]["id"] == str(signal.id)
    assert payload["items"][0]["status_code"] == "linked"
    assert payload["items"][0]["status"] == "已关联宪章"
    assert payload["items"][0]["can_create_charter"] is False
    assert payload["items"][0]["charter_name"] == "银行流水导入宪章"