from __future__ import annotations

from types import SimpleNamespace

import pytest
from fastapi import HTTPException
from starlette.requests import Request

from app.gateway.routers import llmwiki
from app.gateway.weknora_embed import verify_weknora_signed_payload


def _request(store: object) -> Request:
    return Request(
        {
            "type": "http",
            "method": "GET",
            "path": "/api/public/llmwiki/knowledge-bases/kb-one",
            "headers": [],
            "app": SimpleNamespace(state=SimpleNamespace(llmwiki_store=store)),
        }
    )


class PublicStore:
    def __init__(self, row: dict[str, object] | None) -> None:
        self.row = row
        self.calls: list[tuple[str, str, bool, bool]] = []

    async def get_authorized(
        self,
        mapping_id: str,
        user_id: str,
        *,
        write: bool,
        is_admin: bool,
    ) -> dict[str, object] | None:
        self.calls.append((mapping_id, user_id, write, is_admin))
        return self.row


class PublicListStore:
    def __init__(self, rows: list[dict[str, object]]) -> None:
        self.rows = rows
        self.calls: list[tuple[str, str, bool]] = []

    async def list_visible(
        self,
        user_id: str,
        *,
        scope: str,
        is_admin: bool,
    ) -> list[dict[str, object]]:
        self.calls.append((user_id, scope, is_admin))
        return self.rows


def _published_row() -> dict[str, object]:
    return {
        "id": "kb-one",
        "weknora_id": "remote-one",
        "owner_user_id": "owner-one",
        "name": "公开知识库",
        "description": "公开说明",
        "kb_type": "document",
        "publication_status": "published",
    }


@pytest.mark.asyncio
async def test_public_knowledge_base_list_exposes_live_safe_metadata_and_generated_index(monkeypatch) -> None:
    store = PublicListStore(
        [
            {
                **_published_row(),
                "id": "1063edee-332a-43b4-bbd9-73bf0a708b94",
                "owner_user_id": "secret-owner",
            },
            {**_published_row(), "id": "private-kb", "publication_status": "private"},
            {
                **_published_row(),
                "id": "conversation-deposit",
                "owner_user_id": "system",
                "name": "对话沉淀",
            },
            {
                **_published_row(),
                "id": "stale-kb",
                "weknora_id": "remote-missing",
            },
        ]
    )

    class Client:
        async def list_knowledge_bases(self):
            return [
                {
                    "id": "remote-one",
                    "name": "远端公开知识库",
                    "description": "远端公开说明",
                    "type": "wiki",
                    "knowledge_count": 12,
                    "chunk_count": 34,
                    "processing_count": 1,
                    "share_count": 3,
                    "created_at": "2026-07-01T08:00:00+08:00",
                    "updated_at": "2026-08-01T08:00:00+08:00",
                    "tenant_id": 10001,
                    "creator_id": "secret-creator",
                    "storage_config": {"secret_key": "must-not-leak"},
                }
            ]

        async def get_knowledge_base(self, knowledge_base_id: str):
            assert knowledge_base_id == "remote-one"
            return {
                "id": knowledge_base_id,
                "vector_store_name": "System default",
                "vector_store_source": "env",
                "vector_store_engine_type": "elasticsearch",
                "vector_store_status": "available",
                "wiki_config": {"enabled": True},
            }

        async def list_wiki_pages(self, knowledge_base_id: str, **kwargs):
            assert knowledge_base_id == "remote-one"
            assert kwargs == {"page": 1, "page_size": 500, "query": ""}
            return {
                "pages": [
                    {
                        "id": "page-one",
                        "slug": "concept/start",
                        "title": "开始使用",
                        "page_type": "concept",
                        "status": "completed",
                        "created_at": "2026-07-02T08:00:00+08:00",
                        "updated_at": "2026-08-02T08:00:00+08:00",
                    },
                    {
                        "id": "page-two",
                        "slug": "summary/overview",
                        "title": "知识概览",
                        "page_type": "summary",
                        "status": "completed",
                        "created_at": "2026-07-03T08:00:00+08:00",
                        "updated_at": "2026-08-03T08:00:00+08:00",
                    },
                ],
                "total": 2,
                "page": 1,
                "page_size": 500,
                "total_pages": 1,
            }

    monkeypatch.setattr(llmwiki, "_client_or_503", lambda: Client())

    payload = await llmwiki.list_public_knowledge_bases(_request(store))

    assert store.calls == [("__public_wiki__", "public", False)]
    assert payload["total"] == 1
    item = payload["knowledge_bases"][0]
    assert item["id"] == "1063edee-332a-43b4-bbd9-73bf0a708b94"
    assert item["name"] == "远端公开知识库"
    assert item["description"] == "远端公开说明"
    assert item["type"] == "wiki"
    assert item["knowledge_count"] == 12
    assert item["chunk_count"] == 34
    assert item["processing_count"] == 1
    assert item["share_count"] == 3
    assert item["is_processing"] is True
    assert item["remote_created_at"] == "2026-07-01T08:00:00+08:00"
    assert item["remote_updated_at"] == "2026-08-01T08:00:00+08:00"
    assert item["vector_store_engine_type"] == "elasticsearch"
    assert item["wiki"] == {
        "status": "ready",
        "page_count": 2,
        "loaded_page_count": 2,
        "stats_complete": True,
        "page_type_counts": {"concept": 1, "summary": 1},
        "page_status_counts": {"completed": 2},
        "index_mode": "generated",
        "has_explicit_index": False,
        "index_slug": "__index__",
        "default_slug": "__index__",
        "created_at": "2026-07-02T08:00:00+08:00",
        "updated_at": "2026-08-03T08:00:00+08:00",
        "embed_route": "/embed/knowledge/1063edee-332a-43b4-bbd9-73bf0a708b94?wikiSlug=__index__",
    }
    assert item["local_index"]["state"] == "not_synced"
    assert item["local_index"]["fully_vectorized"] is False
    assert item["owner_user_id"] is None
    assert "weknora_id" not in item
    assert "tenant_id" not in item
    assert "creator_id" not in item
    assert "storage_config" not in item


@pytest.mark.asyncio
async def test_public_wiki_reader_uses_anonymous_non_admin_authorization(monkeypatch) -> None:
    store = PublicStore(_published_row())

    class Client:
        async def get_knowledge_base(self, knowledge_base_id: str):
            assert knowledge_base_id == "remote-one"
            return {"id": knowledge_base_id, "name": "公开知识库", "knowledge_count": 2}

    monkeypatch.setattr(llmwiki, "_client_or_503", lambda: Client())

    payload = await llmwiki.get_public_knowledge_base(_request(store), "kb-one")

    assert store.calls == [("kb-one", "__public_wiki__", False, False)]
    assert payload["publication_status"] == "published"
    assert payload["owner_user_id"] is None
    assert payload["can_write"] is False


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "row",
    [
        {**_published_row(), "publication_status": "private"},
        {
            **_published_row(),
            "owner_user_id": "system",
            "name": "对话沉淀",
        },
        None,
    ],
)
async def test_public_wiki_reader_rejects_non_public_resources(row) -> None:
    with pytest.raises(HTTPException) as exc_info:
        await llmwiki._public_mapping(_request(PublicStore(row)), "kb-one")

    assert exc_info.value.status_code == 404


@pytest.mark.asyncio
async def test_public_wiki_page_list_is_normalized(monkeypatch) -> None:
    store = PublicStore(_published_row())

    class Client:
        async def list_wiki_pages(self, knowledge_base_id: str, **kwargs):
            assert knowledge_base_id == "remote-one"
            assert kwargs == {"page": 1, "page_size": 50, "query": "目录"}
            return {
                "pages": [
                    {
                        "id": "page-one",
                        "slug": "guide/start",
                        "title": "开始使用",
                        "page_type": "article",
                        "status": "completed",
                        "content": "# 开始使用",
                    }
                ],
                "total": 1,
                "page": 1,
                "page_size": 50,
                "total_pages": 1,
            }

    monkeypatch.setattr(llmwiki, "_client_or_503", lambda: Client())

    payload = await llmwiki.list_public_wiki_pages(
        _request(store),
        "kb-one",
        page=1,
        page_size=50,
        query=" 目录 ",
    )

    assert payload["pages"][0]["slug"] == "guide/start"
    assert payload["pages"][0]["aliases"] == []


def test_public_wiki_summary_prefers_an_explicit_index_page() -> None:
    payload = llmwiki._public_wiki_summary(
        "mapping-one",
        [
            {
                "slug": "guide/start",
                "title": "开始使用",
                "page_type": "article",
                "status": "completed",
            },
            {
                "slug": "home",
                "title": "知识首页",
                "page_type": "index",
                "status": "completed",
            },
        ],
        total=2,
        complete=True,
    )

    assert payload["status"] == "ready"
    assert payload["index_mode"] == "explicit"
    assert payload["has_explicit_index"] is True
    assert payload["index_slug"] == "home"
    assert payload["default_slug"] == "home"
    assert payload["embed_route"] == "/embed/knowledge/mapping-one?wikiSlug=home"


@pytest.mark.asyncio
async def test_public_wiki_file_is_authorized_and_streamed(monkeypatch) -> None:
    store = PublicStore(_published_row())

    class Client:
        async def get_knowledge_base_file(self, knowledge_base_id: str, file_path: str):
            assert knowledge_base_id == "remote-one"
            assert file_path == "storage://tenant/resource://wiki/figure.png"
            return b"png-bytes", "image/png"

    monkeypatch.setattr(llmwiki, "_client_or_503", lambda: Client())

    response = await llmwiki.get_public_wiki_file(
        _request(store),
        "kb-one",
        "storage://tenant/resource://wiki/figure.png",
    )

    assert store.calls == [("kb-one", "__public_wiki__", False, False)]
    assert response.body == b"png-bytes"
    assert response.media_type == "image/png"
    assert response.headers["cache-control"] == "public, max-age=300"
    assert response.headers["content-disposition"] == "inline; filename*=UTF-8''figure.png"


@pytest.mark.asyncio
async def test_private_wiki_file_uses_signed_http_access_instead_of_blob_url(monkeypatch) -> None:
    async def authorized_mapping(*_args, **_kwargs):
        return _published_row(), "user-one", False

    class Client:
        async def get_knowledge_base_file(self, knowledge_base_id: str, file_path: str):
            assert knowledge_base_id == "remote-one"
            assert file_path == "storage://tenant/resource://wiki/figure.png"
            return b"png-bytes", "image/png"

    monkeypatch.setattr(llmwiki, "_authorized_mapping", authorized_mapping)
    monkeypatch.setattr(llmwiki, "_client_or_503", lambda: Client())

    access = await llmwiki.create_wiki_file_access(
        _request(PublicStore(_published_row())),
        "kb-one",
        "storage://tenant/resource://wiki/figure.png",
    )
    payload = verify_weknora_signed_payload(access["token"])
    assert payload["kind"] == "wiki_file"
    assert payload["mapping_id"] == "kb-one"
    assert payload["weknora_id"] == "remote-one"
    assert payload["file_path"] == "storage://tenant/resource://wiki/figure.png"

    response = await llmwiki.get_signed_wiki_file(access["token"])
    assert response.body == b"png-bytes"
    assert response.headers["cache-control"] == "private, max-age=3600"
    assert response.headers["content-disposition"] == "inline; filename*=UTF-8''figure.png"


def test_wiki_file_proxy_rejects_non_weknora_urls() -> None:
    with pytest.raises(HTTPException) as exc_info:
        llmwiki._validate_protected_wiki_file_path("https://example.test/private.png")

    assert exc_info.value.status_code == 422


@pytest.mark.asyncio
async def test_wiki_file_proxy_adds_image_extension_for_opaque_resource_ids(monkeypatch) -> None:
    class Client:
        async def get_knowledge_base_file(self, knowledge_base_id: str, file_path: str):
            return b"png-bytes", "image/png"

    monkeypatch.setattr(llmwiki, "_client_or_503", lambda: Client())

    response = await llmwiki._wiki_file_response(
        {"weknora_id": "remote-one"},
        "resource://opaque-image-id",
        public=True,
    )

    assert response.headers["content-disposition"] == "inline; filename*=UTF-8''opaque-image-id.png"


@pytest.mark.asyncio
async def test_public_wiki_index_preserves_weknora_group_and_item_order(monkeypatch) -> None:
    store = PublicStore(_published_row())
    calls: list[tuple[tuple[str, ...], str]] = []

    class Client:
        async def get_wiki_index(
            self,
            knowledge_base_id: str,
            *,
            types: list[str],
            limit: int,
            cursor: str,
        ):
            assert knowledge_base_id == "remote-one"
            assert limit == 500
            page_type = types[0]
            calls.append((tuple(types), cursor))
            if page_type == "concept" and not cursor:
                return {
                    "intro": "# Wiki Index\n\n公开目录",
                    "version": 3,
                    "groups": [
                        {
                            "type": "concept",
                            "total": 2,
                            "next_cursor": "concept-next",
                            "items": [
                                {
                                    "slug": "concept/first",
                                    "title": "第一项",
                                    "summary": "第一项摘要",
                                    "category_path": ["流程图", "规范"],
                                }
                            ],
                        }
                    ],
                }
            if page_type == "concept":
                return {
                    "intro": "# Wiki Index\n\n公开目录",
                    "version": 3,
                    "groups": [
                        {
                            "type": "concept",
                            "total": 2,
                            "items": [
                                {
                                    "slug": "concept/second",
                                    "title": "第二项",
                                    "summary": "第二项摘要",
                                    "category_path": ["流程图", "规范"],
                                }
                            ],
                        }
                    ],
                }
            return {
                "intro": "# Wiki Index\n\n公开目录",
                "version": 3,
                "groups": [{"type": page_type, "total": 0, "items": []}],
            }

    monkeypatch.setattr(llmwiki, "_client_or_503", lambda: Client())

    payload = await llmwiki.get_public_wiki_index(_request(store), "kb-one")

    concept = next(group for group in payload["groups"] if group["type"] == "concept")
    assert payload["intro"] == "# Wiki Index\n\n公开目录"
    assert payload["version"] == 3
    assert [item["slug"] for item in concept["items"]] == [
        "concept/first",
        "concept/second",
    ]
    assert concept["total"] == 2
    assert (("concept",), "concept-next") in calls