script-build-and-framework-tmp/script_build_host/tests/test_legacy_active_inputs.py

from __future__ import annotations

import json
from types import SimpleNamespace

import httpx
import numpy as np
import pytest

from script_build_host.adapters.legacy_retrieval import (
    LegacyExternalRetrievalAdapter,
    LegacyLlmKnowledgeRetrievalAdapter,
    LegacyOpenRouterClient,
    LegacyVectorDecodeRetrievalAdapter,
    OpenAICompatibleEmbeddingClient,
)
from script_build_host.adapters.retrieval import SafeHttpClient
from script_build_host.infrastructure.outbound import OutboundPolicy


async def _public_resolver(_: str, __: int) -> tuple[str, ...]:
    return ("93.184.216.34",)


def _decode_fixture(tmp_path):
    index = tmp_path / "index"
    raw = tmp_path / "raw" / "acct"
    index.mkdir()
    raw.mkdir(parents=True)
    (index / "manifest.json").write_text(
        json.dumps({"model": "embed", "dimension": 2}), encoding="utf-8"
    )
    chunks = [
        {
            "chunk_id": "p1|1",
            "post_id": "p1",
            "account": "acct",
            "json_relpath": "script_decode_raw_data/acct/p1.json",
            "column": "主题",
            "sub_field": "维度名",
        },
        {
            "chunk_id": "p2|1",
            "post_id": "p2",
            "account": "other",
            "json_relpath": "script_decode_raw_data/acct/p1.json",
            "column": "形式",
            "sub_field": "维度值",
        },
    ]
    (index / "meta.jsonl").write_text(
        "".join(json.dumps(item, ensure_ascii=False) + "\n" for item in chunks),
        encoding="utf-8",
    )
    np.save(index / "embeddings.npy", np.asarray([[1, 0], [0, 1]], dtype=np.float16))
    (raw / "p1.json").write_text(
        json.dumps(
            {
                "段落列表": [
                    {
                        "id": "段落1",
                        "名称": "开头",
                        "主题": "主题正文",
                        "描述": "完整描述",
                        "主题原子点": [{"维度类型": "主维度", "维度名": "对象", "维度值": "测试"}],
                    }
                ]
            },
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )
    return index, tmp_path / "raw"


@pytest.mark.asyncio
async def test_http_is_permitted_only_for_the_separate_legacy_allowlist() -> None:
    policy = OutboundPolicy(
        allowed_hosts=frozenset({"legacy.example"}),
        allowed_ports=frozenset({80}),
        allowed_http_hosts=frozenset({"legacy.example"}),
        resolver=_public_resolver,
    )
    assert await policy.validate_url("http://legacy.example/input") == (
        "http://legacy.example/input"
    )
    denied = OutboundPolicy(
        allowed_hosts=frozenset({"legacy.example"}),
        allowed_ports=frozenset({80}),
        resolver=_public_resolver,
    )
    with pytest.raises(Exception, match="HTTPS"):
        await denied.validate_url("http://legacy.example/input")


@pytest.mark.asyncio
async def test_legacy_decode_supports_account_only_and_semantic_vector_search(tmp_path) -> None:
    index, raw = _decode_fixture(tmp_path)

    def handler(request: httpx.Request) -> httpx.Response:
        assert request.headers["authorization"] == "Bearer secret"
        return httpx.Response(200, json={"data": [{"index": 0, "embedding": [1, 0]}]})

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
        safe = SafeHttpClient(
            client,
            OutboundPolicy(frozenset({"openrouter.example"}), resolver=_public_resolver),
        )
        embedding_client = OpenAICompatibleEmbeddingClient(
            safe,
            api_key="secret",
            embedding_endpoint="https://openrouter.example/embeddings",
            embedding_model="embed",
            embedding_dimension=2,
        )
        adapter = LegacyVectorDecodeRetrievalAdapter(
            index_root=index, raw_root=raw, embedding_client=embedding_client
        )
        account_result = await adapter.retrieve(
            query={
                "return_field": "主脉络",
                "keyword": "",
                "account_name": "acct",
                "match_fields": {},
                "top_k": 3,
            },
            snapshot=SimpleNamespace(),
        )
        semantic_result = await adapter.retrieve(
            query={
                "return_field": "主题",
                "keyword": "测试",
                "account_name": None,
                "match_fields": {"主题": "维度名"},
                "top_k": 3,
            },
            snapshot=SimpleNamespace(),
        )
    assert account_result.source_refs == ("decode:p1",)
    assert "主维度" in account_result.summary
    assert semantic_result.metadata["scores"] == [1.0]
    assert "主题正文" in semantic_result.summary


@pytest.mark.asyncio
async def test_legacy_decode_rejects_query_model_that_differs_from_index(tmp_path) -> None:
    index, raw = _decode_fixture(tmp_path)

    async with httpx.AsyncClient(
        transport=httpx.MockTransport(lambda _: httpx.Response(500))
    ) as client:
        safe = SafeHttpClient(
            client,
            OutboundPolicy(frozenset({"embedding.example"}), resolver=_public_resolver),
        )
        embedding_client = OpenAICompatibleEmbeddingClient(
            safe,
            api_key="secret",
            embedding_endpoint="https://embedding.example/embeddings",
            embedding_model="different-model",
            embedding_dimension=2,
        )
        adapter = LegacyVectorDecodeRetrievalAdapter(
            index_root=index,
            raw_root=raw,
            embedding_client=embedding_client,
        )
        with pytest.raises(Exception, match="model does not match"):
            await adapter.retrieve(
                query={
                    "return_field": "主题",
                    "keyword": "测试",
                    "account_name": None,
                    "match_fields": {},
                    "top_k": 3,
                },
                snapshot=SimpleNamespace(),
            )


@pytest.mark.asyncio
async def test_legacy_knowledge_uses_full_llm_selection_contract(tmp_path) -> None:
    source = tmp_path / "knowledge.json"
    source.write_text(
        json.dumps(
            [
                {
                    "id": 7,
                    "title": "结构",
                    "purpose": "搭建结构",
                    "steps": [
                        {
                            "intent": "开头",
                            "directive": "先给结果",
                            "outputs": [{"value": "钩子"}],
                        }
                    ],
                }
            ],
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )
    bodies = []

    def handler(request: httpx.Request) -> httpx.Response:
        bodies.append(json.loads(request.content))
        return httpx.Response(
            200,
            json={"choices": [{"message": {"content": '{"ids":[7]}'}}]},
        )

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
        safe = SafeHttpClient(
            client,
            OutboundPolicy(frozenset({"openrouter.example"}), resolver=_public_resolver),
        )
        openrouter = LegacyOpenRouterClient(
            safe,
            api_key="secret",
            chat_endpoint="https://openrouter.example/chat",
        )
        result = await LegacyLlmKnowledgeRetrievalAdapter(
            source, openrouter=openrouter, model="knowledge-model"
        ).retrieve(query={"keyword": "结构", "max_count": 3}, snapshot=SimpleNamespace())
    assert bodies[0]["model"] == "knowledge-model"
    assert "先给结果" in bodies[0]["messages"][1]["content"]
    assert "钩子" in result.summary


@pytest.mark.asyncio
async def test_legacy_external_reproduces_xhs_detail_and_zhihu_shapes() -> None:
    requests: list[tuple[str, dict[str, object]]] = []

    def handler(request: httpx.Request) -> httpx.Response:
        body = json.loads(request.content)
        requests.append((request.url.path, body))
        if request.url.path == "/xhs/search":
            return httpx.Response(
                200,
                json={"code": 0, "data": {"data": [{"id": "note-1"}], "has_more": False}},
            )
        if request.url.path == "/xhs/detail":
            return httpx.Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        "data": {
                            "channel_content_id": "note-1",
                            "title": "标题",
                            "body_text": "正文",
                            "like_count": "12",
                            "publish_timestamp": 100,
                            "image_url_list": [
                                {"image_url": "https://image.example/a.jpg"},
                                {"image_url": "https://image.example/a.jpg"},
                            ],
                        }
                    },
                },
            )
        return httpx.Response(200, json={"code": 0, "data": [{"id": "zhihu-1"}]})

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
        safe = SafeHttpClient(
            client,
            OutboundPolicy(frozenset({"legacy.example"}), resolver=_public_resolver),
        )
        adapter = LegacyExternalRetrievalAdapter(
            xhs_search_endpoint="https://legacy.example/xhs/search",
            xhs_detail_endpoint="https://legacy.example/xhs/detail",
            zhihu_search_endpoint="https://legacy.example/zhihu",
            http=safe,
        )
        xhs = await adapter.retrieve(
            query={"keyword": "测试", "platform_channel": "xhs", "max_count": 5},
            snapshot=SimpleNamespace(),
        )
        zhihu = await adapter.retrieve(
            query={"keyword": "测试", "platform_channel": "zhihu", "max_count": 5},
            snapshot=SimpleNamespace(),
        )
    xhs_rows = json.loads(xhs.summary)
    assert xhs_rows[0]["publish_timestamp"] == 100_000
    assert xhs_rows[0]["images"] == ["https://image.example/a.jpg"]
    assert xhs_rows[0]["images_text_escape"] == [
        {"image_url": "https://image.example/a.jpg", "text": ""}
    ]
    assert zhihu.source_refs == ("external:zhihu-1",)
    assert requests[0][1]["content_type"] == "图文"