from __future__ import annotations

import json
import zipfile
from pathlib import Path
from types import SimpleNamespace

import pytest

from app.gateway.routers import artifacts
from deerflow.html_runtime import offline_bundle
from deerflow.html_runtime.offline_bundle import build_html_offline_bundle
from deerflow.uploads.policy import default_upload_policy, set_runtime_upload_policy


@pytest.fixture(autouse=True)
def _restore_upload_policy():
    set_runtime_upload_policy(default_upload_policy())
    yield
    set_runtime_upload_policy(default_upload_policy())


@pytest.mark.asyncio
async def test_artifact_preview_reads_small_text_without_unbounded_artifact_response(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    source = tmp_path / "notes.txt"
    source.write_text("第一行\n第二行", encoding="utf-8")

    async def resolve(_thread_id: str, _path: str) -> Path:
        return source

    monkeypatch.setattr(artifacts, "aresolve_thread_virtual_path", resolve)
    response = await artifacts.preview_artifact.__wrapped__(
        thread_id="thread-1",
        request=SimpleNamespace(),
        path="mnt/user-data/uploads/notes.txt",
        max_chars=1024,
    )

    assert response.content.splitlines() == ["第一行", "第二行"]
    assert response.truncated is False
    assert response.strategy == "complete-text"
    assert response.total_bytes == source.stat().st_size


@pytest.mark.asyncio
async def test_artifact_preview_samples_large_text_with_full_file_statistics(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    source = tmp_path / "words.txt"
    source.write_text("".join(f"word-{index:04d}\n" for index in range(1, 1001)), encoding="utf-8")
    policy = default_upload_policy()
    policy["large_file"] = {
        "threshold_bytes": 100,
        "max_result_chars": 4096,
        "hard_max_result_chars": 4096,
    }
    set_runtime_upload_policy(policy)

    async def resolve(_thread_id: str, _path: str) -> Path:
        return source

    monkeypatch.setattr(artifacts, "aresolve_thread_virtual_path", resolve)
    response = await artifacts.preview_artifact.__wrapped__(
        thread_id="thread-1",
        request=SimpleNamespace(),
        path="mnt/user-data/uploads/words.txt",
        max_chars=4096,
    )

    assert response.truncated is True
    assert response.stats is not None
    assert response.stats["lines"] == 1000
    assert "流式扫描" in response.content
    assert "word-0001" in response.content
    assert len(response.content) <= 4096
    assert "sampl" in response.strategy


@pytest.mark.asyncio
async def test_artifact_preview_rejects_binary_content(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    source = tmp_path / "payload.bin"
    source.write_bytes(b"abc\x00def")

    async def resolve(_thread_id: str, _path: str) -> Path:
        return source

    monkeypatch.setattr(artifacts, "aresolve_thread_virtual_path", resolve)
    with pytest.raises(artifacts.HTTPException) as raised:
        await artifacts.preview_artifact.__wrapped__(
            thread_id="thread-1",
            request=SimpleNamespace(),
            path="mnt/user-data/uploads/payload.bin",
            max_chars=1024,
        )
    assert raised.value.status_code == 415


@pytest.mark.asyncio
async def test_html_artifact_offline_bundle_uses_zip_export(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    source = tmp_path / "page.html"
    source.write_text("<!doctype html><title>portable</title>", encoding="utf-8")
    calls: list[Path] = []

    async def resolve(_thread_id: str, _path: str) -> Path:
        return source

    async def bundle_response(path: Path):
        calls.append(path)
        return artifacts.Response(content=b"zip-content", media_type="application/zip")

    monkeypatch.setattr(artifacts, "aresolve_thread_virtual_path", resolve)
    monkeypatch.setattr(artifacts, "build_html_offline_bundle_response", bundle_response)
    response = await artifacts.get_artifact.__wrapped__(
        thread_id="thread-1",
        path="mnt/user-data/outputs/page.html",
        request=SimpleNamespace(),
        download=True,
        offline_bundle=True,
    )

    assert response.media_type == "application/zip"
    assert calls == [source]


def _write_offline_fixture(path: Path, content: str | bytes) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    if isinstance(content, bytes):
        path.write_bytes(content)
    else:
        path.write_text(content, encoding="utf-8")


def _offline_runtime_fixture(root: Path) -> Path:
    runtime = root / "runtime"
    _write_offline_fixture(runtime / "echarts.min.js", "window.echarts={};")
    _write_offline_fixture(runtime / "deerflow-runtime.js", "window.__runtimeLoaded=true;")
    _write_offline_fixture(runtime / "brand-logo-dark.png", b"fallback-logo")
    _write_offline_fixture(runtime / "LICENSE.echarts.txt", "ECharts license")
    for name in (
        "china-provinces.geojson",
        "japan-prefectures.geojson",
        "taiwan-counties.geojson",
        "us-states.geojson",
    ):
        _write_offline_fixture(runtime / "maps" / name, '{"type":"FeatureCollection","features":[]}')
    return runtime


def test_html_offline_bundle_rewrites_and_packages_gateway_runtime(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    runtime = _offline_runtime_fixture(tmp_path)
    logo = tmp_path / "branding" / "logo.webp"
    _write_offline_fixture(logo, b"configured-logo")
    uploaded_image = tmp_path / "uploads" / "background.png"
    _write_offline_fixture(uploaded_image, b"uploaded-image")
    monkeypatch.setattr(
        offline_bundle,
        "verify_html_image_token",
        lambda _token: ("thread-1", "uploads", "background.png"),
    )
    monkeypatch.setattr(
        offline_bundle,
        "resolve_html_image_path",
        lambda _thread_id, _kind, _filename: uploaded_image,
    )
    source = tmp_path / "态势报告.html"
    source.write_text(
        """<!doctype html><html><head>
        <script src="http://gateway.local/prefix/api/public/html-runtime/echarts.min.js"></script>
        <script src="http://gateway.local/prefix/api/public/html-runtime/deerflow-runtime.js"></script>
        <style>.brand{background:url('http://gateway.local/prefix/api/public/html-runtime/generated-page-logo?v=2')}</style>
        </head><body><script type="application/json" data-df-geo="map">{"map":"china-provinces"}</script>
        <script type="application/json" data-df-map="tile-map">{"markers":[]}</script>
        <img src="http://gateway.local/prefix/api/public/html-images/signed.token">
        <a href="https://example.com/source">来源</a></body></html>""",
        encoding="utf-8",
    )
    destination = tmp_path / "bundle.zip"

    result = build_html_offline_bundle(
        source,
        destination,
        runtime_base_url="http://gateway.local/prefix/api/public/html-runtime",
        map_tile_url="https://tiles.example/{z}/{x}/{y}.png",
        generated_logo_path=logo,
        runtime_dir=runtime,
    )

    assert destination.is_file()
    assert "assets/html-runtime/echarts.min.js" in result.bundled_assets
    assert "https://example.com/source" in result.external_urls
    assert "https://tiles.example/{z}/{x}/{y}.png" in result.external_urls
    with zipfile.ZipFile(destination) as archive:
        names = set(archive.namelist())
        assert "index.html" in names
        assert "README.txt" in names
        assert "manifest.json" in names
        assert "assets/html-runtime/echarts.min.js" in names
        assert "assets/html-runtime/deerflow-runtime.js" in names
        assert "assets/html-runtime/generated-page-logo.webp" in names
        assert "assets/html-runtime/maps/china-provinces.geojson" in names
        assert any(name.startswith("assets/html-images/") for name in names)
        assert "THIRD_PARTY_LICENSES/LICENSE.echarts.txt" in names

        html = archive.read("index.html").decode("utf-8")
        assert "gateway.local" not in html
        assert 'src="assets/html-runtime/echarts.min.js"' in html
        assert "assets/html-runtime/generated-page-logo.webp" in html
        assert "/api/public/html-images/" not in html

        runtime_script = archive.read("assets/html-runtime/deerflow-runtime.js").decode("utf-8")
        assert "DEERFLOW_HTML_RUNTIME_CONFIG" in runtime_script
        assert "assets/html-runtime/maps/china-provinces.geojson" in runtime_script
        assert "window.__runtimeLoaded=true" in runtime_script

        manifest = json.loads(archive.read("manifest.json"))
        assert manifest["entry"] == "index.html"
        assert manifest["source_name"] == "态势报告.html"
        assert manifest["external_urls"] == [
            "https://example.com/source",
            "https://tiles.example/{z}/{x}/{y}.png",
        ]


def test_html_offline_bundle_copies_cognitive_runtime_tree(tmp_path: Path) -> None:
    runtime = _offline_runtime_fixture(tmp_path)
    _write_offline_fixture(runtime / "cognitive-situation-map" / "app.js", "window.cognitive=true;")
    _write_offline_fixture(runtime / "cognitive-situation-map" / "styles.css", "body{color:white}")
    _write_offline_fixture(runtime / "cognitive-situation-map" / "assets" / "maps" / "taiwan.png", b"map")
    source = tmp_path / "认知态势图.html"
    source.write_text(
        """<script>window.__DF_SKILL_BASE__="http://gw/api/public/html-runtime/cognitive-situation-map";</script>
        <link rel="stylesheet" href="http://gw/api/public/html-runtime/cognitive-situation-map/styles.css">
        <script src="http://gw/api/public/html-runtime/cognitive-situation-map/app.js"></script>""",
        encoding="utf-8",
    )
    destination = tmp_path / "cognitive.zip"

    build_html_offline_bundle(
        source,
        destination,
        runtime_base_url="http://gw/api/public/html-runtime",
        map_tile_url="https://tiles.example/{z}/{x}/{y}.png",
        runtime_dir=runtime,
    )

    with zipfile.ZipFile(destination) as archive:
        html = archive.read("index.html").decode("utf-8")
        assert "http://gw" not in html
        assert "assets/html-runtime/cognitive-situation-map" in html
        assert "assets/html-runtime/cognitive-situation-map/app.js" in archive.namelist()
        assert "assets/html-runtime/cognitive-situation-map/assets/maps/taiwan.png" in archive.namelist()