from __future__ import annotations

import importlib.util
import json
import subprocess
import sys
from pathlib import Path

import pytest

from deerflow.skills.parser import parse_skill_file
from deerflow.skills.storage.local_skill_storage import LocalSkillStorage
from deerflow.skills.types import SkillCategory
from deerflow.skills.validation import _validate_skill_frontmatter

BACKEND_ROOT = Path(__file__).resolve().parents[1]
SKILL_ROOT = BACKEND_ROOT / "skills" / "public" / "cnn-page-generator"
SCRIPT_PATH = SKILL_ROOT / "scripts" / "render_cnn_page.py"
FAST_SCRIPT_PATH = SKILL_ROOT / "scripts" / "fast_page.py"
LEAD_PROMPT_PATH = BACKEND_ROOT / "packages" / "harness" / "deerflow" / "agents" / "lead_agent" / "prompt.py"


def _load_renderer():
    spec = importlib.util.spec_from_file_location(
        "cnn_page_renderer",
        SCRIPT_PATH,
    )
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def test_skill_is_discoverable_and_valid() -> None:
    valid, error, _metadata = _validate_skill_frontmatter(SKILL_ROOT)
    assert valid, error
    skill = parse_skill_file(SKILL_ROOT / "SKILL.md", SkillCategory.PUBLIC)
    assert skill is not None
    assert skill.name == "cnn-page-generator"
    assert "CNN" in skill.description

    discovered = LocalSkillStorage(host_path=str(SKILL_ROOT.parent.parent)).load_skills(enabled_only=True)
    discovered_by_name = {item.name: item for item in discovered}
    assert discovered_by_name["cnn-page-generator"].category == SkillCategory.PUBLIC
    assert discovered_by_name["cnn-page-generator"].enabled is True

    lead_prompt_source = LEAD_PROMPT_PATH.read_text(encoding="utf-8")
    assert "/mnt/skills/public/cnn-page-generator/SKILL.md" in lead_prompt_source
    assert "专用 CNN 路由" in lead_prompt_source


def test_default_json_renders_editable_cnn_html(tmp_path: Path) -> None:
    renderer = _load_renderer()
    data = json.loads((SKILL_ROOT / "assets" / "default-data.json").read_text(encoding="utf-8"))
    summary = renderer.validate_document(data)
    template = (SKILL_ROOT / "assets" / "cnn-template.html").read_text(encoding="utf-8")
    runtime = (SKILL_ROOT / "assets" / "cnn-runtime.js").read_text(encoding="utf-8")
    html = renderer.render_document(data, template, runtime)

    assert summary["modules"] == list(renderer.MODULES)
    assert summary["counts"]["latest"] >= 1
    assert renderer.DATA_MARKER not in html
    assert renderer.RUNTIME_MARKER not in html
    assert 'id="cnnPageData"' in html
    assert 'id="leadGrid"' in html
    assert 'id="videoGrid"' in html
    assert 'id="liveModal"' in html
    assert 'id="openSettings"' in html
    assert 'id="moduleTabs"' in html
    assert 'id="applyJson"' in html
    assert "uploadJsonMedia" in html
    assert "installInlineMediaReplace" in html
    assert "dfInlineMediaInput" in html
    assert 'input.accept = "image/*,video/*"' in html
    assert 'button.title = "上传图片或视频并替换"' in html
    assert "active.expected" not in html
    assert "previous.posterUrl" not in html
    assert "decorateBuiltInVisuals" in html
    assert "replaceBuiltInVisuals" not in html
    assert "readAsDataURL" in html
    assert '"brand": "CNN"' in html
    assert "<title>CNN — 全球新闻现场</title>" in html
    assert '"cnn-page-data-v1:" + pageFingerprint' in html
    assert "const readStorage = (key)" in html
    assert "window.localStorage.getItem" in html
    assert "const cachedData = localStorage.getItem" not in html
    assert '<img src="http' not in html
    assert '<video src="http' not in html

    data["site"]["pageTitle"] = "CNN · 智能体新闻中心"
    titled_html = renderer.render_document(data, template, runtime)
    assert "<title>CNN · 智能体新闻中心</title>" in titled_html
    assert "<title>CNN — 全球新闻现场</title>" not in titled_html

    data["lead"]["primary"]["title"] = "</script><script>alert('x')</script>"
    escaped_html = renderer.render_document(data, template, runtime)
    assert "</script><script>alert" not in escaped_html
    assert "<\\/script><script>alert" in escaped_html

    output = tmp_path / "cnn-page.html"
    renderer._write_new(output, html)
    assert output.is_file()
    with pytest.raises(
        renderer.DataValidationError,
        match="Refusing to overwrite",
    ):
        renderer._write_new(output, html)


def test_validation_rejects_missing_duplicate_and_remote_visual() -> None:
    renderer = _load_renderer()
    data = json.loads((SKILL_ROOT / "assets" / "default-data.json").read_text(encoding="utf-8"))
    data.pop("footer")
    data["latest"]["items"][1]["id"] = data["latest"]["items"][0]["id"]
    data["latest"]["items"][0]["visual"] = "remote-image"

    with pytest.raises(renderer.DataValidationError) as exc_info:
        renderer.validate_document(data)

    message = str(exc_info.value)
    assert "Missing top-level module: footer" in message
    assert "must be unique" in message
    assert "visual must be one of" in message


def test_remote_media_contract_supports_image_video_and_webpage() -> None:
    renderer = _load_renderer()
    source = json.loads((SKILL_ROOT / "assets" / "default-data.json").read_text(encoding="utf-8"))
    media_values = (
        {"type": "image", "mediaUrl": "https://cdn.example.com/news.jpg", "alt": "新闻图片"},
        {"type": "video", "mediaUrl": "https://cdn.example.com/news.mp4", "posterUrl": "https://cdn.example.com/poster.jpg"},
        {"type": "webpage", "pageUrl": "https://example.com/story", "title": "新闻原文", "embed": False},
    )
    for media in media_values:
        data = json.loads(json.dumps(source, ensure_ascii=False))
        data["lead"]["primary"]["media"] = media
        renderer.validate_document(data)
    runtime = (SKILL_ROOT / "assets" / "cnn-runtime.js").read_text(encoding="utf-8")
    assert 'media.type === "webpage"' in runtime
    assert 'rel="noopener noreferrer"' in runtime


def test_high_level_fast_spec_builds_all_modules_in_one_call(tmp_path: Path) -> None:
    spec_path = tmp_path / "cnn-page-spec.json"
    spec_path.write_text(
        json.dumps({"topic": "茂达蛋 IP"}, ensure_ascii=False),
        encoding="utf-8",
    )
    command = [
        sys.executable,
        str(FAST_SCRIPT_PATH),
        "--spec",
        str(spec_path),
        "--output-dir",
        str(tmp_path),
    ]

    completed = subprocess.run(
        command,
        check=False,
        capture_output=True,
        text=True,
        encoding="utf-8",
    )

    assert completed.returncode == 0, completed.stdout + completed.stderr
    result = json.loads(completed.stdout)
    assert result["status"] == "ok"
    assert result["action"] == "fast-cnn-page"
    data_path = Path(result["data"])
    html_path = Path(result["html"])
    assert data_path.name == "cnn-page-data.json"
    assert html_path.name == "cnn-page.html"

    data = json.loads(data_path.read_text(encoding="utf-8"))
    assert data["site"]["pageTitle"] == "CNN 中文专题 · 茂达蛋 IP"
    assert data["site"]["date"] == "中文示意页面 · 非实时新闻"
    assert data["lead"]["primary"]["title"].startswith("茂达蛋 IP发布最新计划")
    assert data["lead"]["primary"]["visual"] == "summit"
    assert data["latest"]["items"][0]["visual"] == "coast"
    assert data["video"]["items"][0]["visual"] == "night-city"
    assert data["world"]["items"][0]["visual"] == "trade"
    assert data["markets"]["title"] == "趋势快照 · 示意数据"
    assert data["markets"]["items"][0]["change"] == "非实时"
    html = html_path.read_text(encoding="utf-8")
    assert "茂达蛋 IP发布最新计划" in html
    assert "本页面内容均为中文示意" in html

    second = subprocess.run(
        command,
        check=False,
        capture_output=True,
        text=True,
        encoding="utf-8",
    )
    assert second.returncode == 0, second.stdout + second.stderr
    second_result = json.loads(second.stdout)
    assert Path(second_result["data"]).name == "cnn-page-2-data.json"
    assert Path(second_result["html"]).name == "cnn-page-2.html"


def test_high_level_fast_spec_rejects_english_only_headline(tmp_path: Path) -> None:
    spec_path = tmp_path / "invalid-spec.json"
    spec_path.write_text(
        json.dumps(
            {"topic": "测试专题", "headline": "This headline must be Chinese."},
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )

    completed = subprocess.run(
        [
            sys.executable,
            str(FAST_SCRIPT_PATH),
            "--spec",
            str(spec_path),
            "--output-dir",
            str(tmp_path),
        ],
        check=False,
        capture_output=True,
        text=True,
        encoding="utf-8",
    )

    assert completed.returncode == 2
    result = json.loads(completed.stdout)
    assert result["status"] == "error"
    assert "must contain Simplified Chinese" in result["error"]


def test_cnn_fast_workflow_forbids_probe_and_deep_patch_paths() -> None:
    skill_text = (SKILL_ROOT / "SKILL.md").read_text(encoding="utf-8")
    prompt_text = LEAD_PROMPT_PATH.read_text(encoding="utf-8")

    assert '{"topic":"茂达蛋 IP"}' in skill_text
    assert "不得写十模块 JSON、`$set` 或深层字段路径" in skill_text
    assert "禁止探路渲染" in skill_text
    assert "禁止探路和二次渲染" in prompt_text