"""researcher_node：检索类型 `knowledge_base`（知识库检索）= 直连内网 ES 接口。

弱模型识别不了技能时用这条路：不编排任何技能、不跑通用 provider，
直接调 ai_writing.intranet_search_url 内网知识库 ES 接口检索。
"""
from __future__ import annotations

from io import BytesIO
from types import SimpleNamespace

import pytest
from fastapi import UploadFile

from app.gateway.routers.ai_writing import _extract_sample_text
from deerflow.agents.ai_writing.nodes import researcher as R
from deerflow.agents.ai_writing.nodes import sample_analyzer as S
from deerflow.agents.ai_writing.search import get_search_provider
from deerflow.agents.ai_writing.search.base import SearchResult
from deerflow.agents.ai_writing.search.intranet_stub import IntranetSearchProvider


class _FakeIntranet:
    instances: list = []

    def __init__(self, **kw):
        self.errors = []
        self.raw_responses = []
        self.calls = []
        self.kw = kw
        self.verify_ssl = kw.get("verify_ssl", False)
        _FakeIntranet.instances.append(self)

    async def search(self, query, max_results=5, **_):
        self.calls.append({"query": query, "status": "ok", "ok": True, "count": 1, "elapsed_ms": 12})
        return [SearchResult(
            title=f"内网-{query}", content="内网正文", source="ES",
            published_at="", url=f"http://es/{query}", source_type="intranet",
        )]


class _EmptyIntranet(_FakeIntranet):
    async def search(self, query, max_results=5, **_):
        self.calls.append({"query": query, "status": "empty", "ok": True, "count": 0, "elapsed_ms": 9})
        return []


class _FailingIntranet(_FakeIntranet):
    async def search(self, query, max_results=5, **_):
        msg = f"ConnectError: cannot connect for {query}"
        self.errors.append(msg)
        self.raw_responses.append("{\"success\": false, \"error\": \"cannot connect\"}")
        self.calls.append({"query": query, "status": "error", "ok": False, "count": 0, "elapsed_ms": 3, "error": msg})
        return []


class _SkillProviderShouldNotRun:
    """knowledge_base 模式不应触碰通用 provider —— 调到就让测试爆。"""
    errors: list = []
    raw_responses: list = []

    async def search(self, query, max_results=5, **_):  # pragma: no cover - 不应被调用
        raise AssertionError("knowledge_base 模式不应调用通用 provider / 技能检索")


class _FakeResponse:
    status_code = 200
    text = '[{"title":"内网标题","content":"内网正文","url":"https://intranet/doc"}]'

    def raise_for_status(self) -> None:
        return None

    def json(self):
        return [{"title": "内网标题", "content": "内网正文", "url": "https://intranet/doc"}]


def _patch(monkeypatch, intranet_url="http://es/api", intranet_cls=_FakeIntranet, intranet_verify_ssl=False):
    _FakeIntranet.instances = []
    cfg = {
        "max_materials": 20,
        "intranet_search_url": intranet_url,
        "intranet_knowledge_base": "samples",
        "intranet_verify_ssl": intranet_verify_ssl,
    }
    monkeypatch.setattr(R, "get_ai_writing_config", lambda: cfg)
    monkeypatch.setattr(R, "use_builtin_agent", lambda *_: False)
    monkeypatch.setattr(R, "safe_get_stream_writer", lambda: (lambda *a, **k: None))
    monkeypatch.setattr(R, "use_skill_search_agent", lambda: False)

    async def fake_stream_llm_json(*a, **k):
        return {"keywords": ["特朗普竞选"]}

    monkeypatch.setattr(R, "_stream_llm_json", fake_stream_llm_json)
    # knowledge_base 模式不该加载/编排技能，但仍打桩以防意外路径
    monkeypatch.setattr(R, "_load_configured_skills",
                        lambda: [SimpleNamespace(name="knowledge-search-v2", description="内网检索")])
    monkeypatch.setattr(R, "get_search_provider", lambda _cfg: _SkillProviderShouldNotRun())
    monkeypatch.setattr(
        "deerflow.agents.ai_writing.search.intranet_stub.IntranetSearchProvider",
        intranet_cls,
    )


def _state(material_source="knowledge_base"):
    return {
        "user_intent": "分析特朗普近期动向",
        "article_type": "政治分析",
        "target_audience": "普通读者",
        "keyword_count": 1,
        "material_source": material_source,
    }


@pytest.mark.asyncio
async def test_knowledge_base_hits_intranet_directly(monkeypatch):
    _patch(monkeypatch)
    out = await R.researcher_node(_state(), {"configurable": {}})
    mats = out["material_package"]["materials"]
    assert mats, "知识库检索应返回内网素材"
    assert all(m["source_type"] == "intranet" for m in mats)
    assert mats[0]["title"] == "内网-特朗普竞选"
    assert _FakeIntranet.instances, "应实例化 IntranetSearchProvider 直连内网接口"
    # 走素材确认暂停点
    assert out["status"] == "awaiting_material_confirm"
    dbg = out["material_package"].get("intranet_debug")
    assert dbg["mode"] == "knowledge_base"
    assert dbg["status"] == "ok"
    assert dbg["endpoint"] == "http://es/api"
    assert dbg["result_count"] == 1
    assert dbg["queries"][0]["query"] == "特朗普竞选"
    assert dbg["queries"][0]["count"] == 1
    assert dbg["calls"][0]["status"] == "ok"


@pytest.mark.asyncio
async def test_knowledge_base_no_url_falls_back_to_default_endpoint(monkeypatch):
    """未配置接口地址：仍实例化 IntranetSearchProvider（base_url 传空串，由 provider 内部
    回退默认内网端点），内网直连依然可用——恢复旧版 search_provider=intranet 的行为。"""
    _patch(monkeypatch, intranet_url="")
    out = await R.researcher_node(_state(), {"configurable": {}})
    mats = out["material_package"]["materials"]
    assert mats, "无地址时应回退默认内网端点，仍返回素材"
    assert all(m["source_type"] == "intranet" for m in mats)
    assert _FakeIntranet.instances, "应实例化 IntranetSearchProvider 直连内网接口"
    assert _FakeIntranet.instances[0].kw.get("base_url") == "", "空地址应原样下传，由 provider 回退默认端点"


@pytest.mark.asyncio
async def test_knowledge_base_passes_verify_ssl_config(monkeypatch):
    _patch(monkeypatch, intranet_verify_ssl=True)
    out = await R.researcher_node(_state(), {"configurable": {}})

    assert _FakeIntranet.instances[0].kw.get("verify_ssl") is True
    assert out["material_package"]["intranet_debug"]["verify_ssl"] is True


@pytest.mark.asyncio
async def test_knowledge_base_does_not_orchestrate_skills(monkeypatch):
    """knowledge_base 不编排技能：通用 provider 一旦被调用就会 raise。"""
    _patch(monkeypatch)
    # 不抛异常即证明没走 _SkillProviderShouldNotRun
    out = await R.researcher_node(_state(), {"configurable": {}})
    assert out["material_package"]["materials"]


@pytest.mark.asyncio
async def test_sample_analyzer_skips_normal_mode():
    out = await S.sample_analyzer_node({"writing_mode_type": "normal"}, {"configurable": {}})

    assert out == {}


@pytest.mark.asyncio
async def test_sample_analyzer_builds_style_profile(monkeypatch):
    async def fake_stream_sample_analysis(*_args, **_kwargs):
        return {
            "summary": "短段落、评论式表达",
            "tone": "克制、坚定",
            "structure": ["先提出问题", "再分层论证"],
            "language_features": ["多用判断句"],
            "sentence_style": "短句为主",
            "opening_style": "开门见山",
            "ending_style": "总结升华",
            "imitation_guidance": ["保持评论式节奏"],
            "avoid_copying": ["不要照抄原句"],
            "sample_outline": ["问题提出", "原因分析"],
        }

    monkeypatch.setattr(S, "_stream_sample_analysis", fake_stream_sample_analysis)

    out = await S.sample_analyzer_node({
        "writing_mode_type": "imitate",
        "sample_text": "这是样文第一段。\n\n这是样文第二段。",
        "sample_filename": "demo.md",
        "user_intent": "写一篇新文章",
    }, {"configurable": {"model_name": "m1"}})

    assert out["status"] == "researching"
    assert out["sample_style_profile"]["summary"] == "短段落、评论式表达"
    assert out["sample_style_profile"]["tone"] == "克制、坚定"
    assert any(e["type"] == "sample_profile_ready" for e in out["progress_events"])


def test_render_sample_style_section_contains_copy_guardrails():
    section = S.render_sample_style_section({
        "writing_mode_type": "imitate",
        "imitate_strength": "strong",
        "preserve_sample_structure": True,
        "sample_style_profile": {
            "summary": "评论体",
            "tone": "正式",
            "structure": ["开头提出问题"],
            "language_features": ["短句"],
            "sentence_style": "长短句结合",
            "opening_style": "开门见山",
            "ending_style": "总结",
            "imitation_guidance": ["保留节奏"],
            "avoid_copying": ["不要照抄"],
            "sample_outline": ["引入", "展开"],
        },
    })

    assert "样文仿写要求" in section
    assert "不得连续照抄样文原句" in section
    assert "评论体" in section


@pytest.mark.asyncio
async def test_extract_sample_text_reads_txt_upload():
    upload = UploadFile(file=BytesIO("  样文内容  ".encode("utf-8")), filename="sample.txt")

    name, text = await _extract_sample_text(upload)

    assert name == "sample.txt"
    assert text == "样文内容"


@pytest.mark.asyncio
async def test_knowledge_base_empty_result_is_visible(monkeypatch):
    """接口成功但 0 条也要在返回里能看出来，而不是只显示"无素材"。"""
    _patch(monkeypatch, intranet_cls=_EmptyIntranet)
    out = await R.researcher_node(_state(), {"configurable": {}})
    pkg = out["material_package"]
    assert pkg["materials"] == []
    dbg = pkg.get("intranet_debug")
    assert dbg["status"] == "empty"
    assert dbg["result_count"] == 0
    assert dbg["queries"][0]["count"] == 0
    assert dbg["calls"][0]["status"] == "empty"
    steps = pkg["skill_search_steps"]
    assert any(step["type"] == "skill_result" and "0 条" in step["message"] for step in steps)


@pytest.mark.asyncio
async def test_knowledge_base_failure_is_visible(monkeypatch):
    """provider 吞掉异常返回 [] 时，前端仍应看到报错步骤与 notice。"""
    _patch(monkeypatch, intranet_cls=_FailingIntranet)
    out = await R.researcher_node(_state(), {"configurable": {}})
    pkg = out["material_package"]
    dbg = pkg.get("intranet_debug")
    assert dbg["status"] == "error"
    assert dbg["errors"]
    assert dbg["raw_responses"]
    steps = pkg["skill_search_steps"]
    assert any(step["type"] == "skill_error" for step in steps)
    assert any(event["type"] == "notice" and event.get("raw") for event in out["progress_events"])


@pytest.mark.asyncio
async def test_intranet_search_disables_ssl_verification_for_internal_https(monkeypatch):
    import httpx

    seen: dict = {}

    class _FakeAsyncClient:
        def __init__(self, **kwargs):
            seen.update(kwargs)

        async def __aenter__(self):
            return self

        async def __aexit__(self, exc_type, exc, tb):
            return None

        async def post(self, endpoint, json):
            seen["endpoint"] = endpoint
            seen["payload"] = json
            return _FakeResponse()

    monkeypatch.setattr(httpx, "AsyncClient", _FakeAsyncClient)

    provider = IntranetSearchProvider(base_url="https://intranet.example/search", knowledge_base_name="kb", timeout=300)
    results = await provider.search("测试", max_results=1)

    assert results
    assert seen["verify"] is False
    assert seen["trust_env"] is False
    assert seen["endpoint"] == "https://intranet.example/search"
    assert seen["payload"]["knowledge_base_name"] == "kb"
    assert provider.calls[0]["verify_ssl"] is False


def test_intranet_verify_ssl_can_be_reenabled_from_config():
    provider = get_search_provider({"search_provider": "intranet", "intranet_verify_ssl": "true"})

    assert isinstance(provider, IntranetSearchProvider)
    assert provider.verify_ssl is True