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" / "facebook-page-generator"
SCRIPT_PATH = SKILL_ROOT / "scripts" / "render_facebook_page.py"
FAST_SCRIPT_PATH = SKILL_ROOT / "scripts" / "fast_single_post.py"
LEAD_PROMPT_PATH = BACKEND_ROOT / "packages" / "harness" / "deerflow" / "agents" / "lead_agent" / "prompt.py"


def _load_renderer():
    spec = importlib.util.spec_from_file_location("facebook_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 == "facebook-page-generator"
    assert "Facebook" 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["facebook-page-generator"].category == SkillCategory.PUBLIC
    assert discovered_by_name["facebook-page-generator"].enabled is True

    lead_prompt_source = LEAD_PROMPT_PATH.read_text(encoding="utf-8")
    assert "/mnt/skills/public/facebook-page-generator/SKILL.md" in lead_prompt_source
    assert "专用 Facebook 路由" in lead_prompt_source
    assert "fast_single_post.py --spec" in lead_prompt_source
    assert "动态与评论正文必须使用简体中文" in lead_prompt_source


def test_default_json_renders_editable_facebook_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" / "facebook-template.html").read_text(encoding="utf-8")
    html = renderer.render_document(data, template)

    assert summary["modules"] == list(renderer.MODULES)
    assert summary["counts"]["posts"] >= 1
    assert renderer.DATA_MARKER not in html
    assert 'id="circleData"' in html
    assert 'id="openSettings"' in html
    assert 'id="settingsModal"' 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 "文件不能超过 20MB" not in html
    assert '"brand": "Facebook"' in html
    assert "<title>Facebook · 社交动态</title>" in html
    data["site"]["pageTitle"] = "Facebook · 智能体开发圈"
    titled_html = renderer.render_document(data, template)
    assert "<title>Facebook · 智能体开发圈</title>" in titled_html
    assert "<title>Facebook · 社交动态</title>" not in titled_html
    assert "facebook-page-data-v2:${dataFingerprint}" in html
    assert "const readStorage = (key)" in html
    assert "const savedTheme = readStorage('circle-theme')" in html
    assert "writeStorage('circle-theme', next)" in html
    assert "const savedTheme = localStorage.getItem('circle-theme')" not in html
    assert "const storageKey = 'circle-page-data-v1'" not in html

    data["posts"]["items"][0]["text"] = "</script><script>alert('x')</script>"
    escaped_html = renderer.render_document(data, template)
    assert "</script><script>alert" not in escaped_html
    assert "<\\/script><script>alert" in escaped_html

    output = tmp_path / "facebook.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_module_and_duplicate_posts() -> None:
    renderer = _load_renderer()
    data = json.loads((SKILL_ROOT / "assets" / "default-data.json").read_text(encoding="utf-8"))
    data.pop("contacts")
    data["posts"]["items"][1]["id"] = data["posts"]["items"][0]["id"]

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

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


@pytest.mark.parametrize(
    "media",
    [
        {
            "type": "image",
            "mediaUrl": "https://cdn.example.com/photo.jpg",
            "alt": "演示图片",
            "pageUrl": "https://example.com/article",
        },
        {
            "type": "video",
            "mediaUrl": "https://cdn.example.com/video.mp4",
            "posterUrl": "https://cdn.example.com/poster.jpg",
            "alt": "演示视频",
        },
        {
            "type": "webpage",
            "pageUrl": "https://example.com/article",
            "title": "演示网页",
            "description": "可在新窗口打开，也可选择内嵌预览。",
            "embed": True,
        },
    ],
)
def test_remote_media_contract_renders_safe_editable_markup(media: dict[str, object]) -> None:
    renderer = _load_renderer()
    data = json.loads((SKILL_ROOT / "assets" / "default-data.json").read_text(encoding="utf-8"))
    data["posts"]["items"][0]["media"] = media

    renderer.validate_document(data)
    html = renderer.render_document(
        data,
        (SKILL_ROOT / "assets" / "facebook-template.html").read_text(encoding="utf-8"),
    )

    assert "media.type === 'image'" in html
    assert "media.type === 'video'" in html
    assert "media.type === 'webpage'" in html
    assert 'rel="noopener noreferrer"' in html
    assert "safeRemoteUrl" in html


def test_remote_media_contract_rejects_unsafe_urls() -> None:
    renderer = _load_renderer()
    data = json.loads((SKILL_ROOT / "assets" / "default-data.json").read_text(encoding="utf-8"))
    data["posts"]["items"][0]["media"] = {
        "type": "image",
        "mediaUrl": "javascript:alert(1)",
        "alt": "不安全图片",
    }

    with pytest.raises(renderer.DataValidationError, match=r"mediaUrl must use an http\(s\) URL"):
        renderer.validate_document(data)


def test_uploaded_data_url_is_accepted_by_media_contract() -> None:
    renderer = _load_renderer()
    data = json.loads((SKILL_ROOT / "assets" / "default-data.json").read_text(encoding="utf-8"))
    data["posts"]["items"][0]["media"] = {
        "type": "image",
        "mediaUrl": "data:image/png;base64,iVBORw0KGgo=",
        "alt": "用户上传的图片",
    }

    renderer.validate_document(data)


def test_single_post_supports_fifty_preloaded_hot_comments() -> None:
    renderer = _load_renderer()
    data = json.loads((SKILL_ROOT / "assets" / "default-data.json").read_text(encoding="utf-8"))
    post = data["posts"]["items"][0]
    data["posts"]["items"] = [post]
    post["metrics"]["comments"] = 50
    post["commentsExpanded"] = True
    post["disclosure"] = "情景模拟页面 · 帖文与热评为演示内容"
    post["sourceUrl"] = "https://example.com/source-post"
    post["comments"] = [
        {
            "id": f"comment-{index:02d}",
            "author": f"演示用户{index:02d}",
            "handle": f"@demo{index:02d}",
            "initial": str(index % 10),
            "colors": ["#4d6aff", "#8f4bd2"],
            "verified": index % 11 == 0,
            "time": f"{index + 1} 分钟前",
            "text": f"这是第 {index + 1} 条演示热评。",
            "likes": 500 - index,
            "replyCount": index % 7,
        }
        for index in range(50)
    ]

    summary = renderer.validate_document(data)
    html = renderer.render_document(
        data,
        (SKILL_ROOT / "assets" / "facebook-template.html").read_text(encoding="utf-8"),
    )

    assert summary["counts"]["posts"] == 1
    assert summary["counts"]["comments"] == 50
    assert '"id": "comment-49"' in html
    assert "comment-pagination" in html
    assert 'data-sort="hot"' in html
    assert "showAllLabel" in html
    assert "post.disclosure" not in html
    assert "post-source" in html
    assert "查看原帖" in html


def test_comment_limit_rejects_more_than_one_hundred_items() -> None:
    renderer = _load_renderer()
    data = json.loads((SKILL_ROOT / "assets" / "default-data.json").read_text(encoding="utf-8"))
    post = data["posts"]["items"][0]
    post["metrics"]["comments"] = 101
    post["comments"] = [{"author": f"用户{index}", "text": "演示评论"} for index in range(101)]

    with pytest.raises(renderer.DataValidationError, match="at most 100"):
        renderer.validate_document(data)


def test_fast_single_post_builds_json_and_html_in_one_call(tmp_path: Path) -> None:
    command = [
        sys.executable,
        str(FAST_SCRIPT_PATH),
        "--output-dir",
        str(tmp_path),
        "--author",
        "测试账号",
        "--text",
        "这是一条用于验证快速生成路径的动态。",
        "--topic",
        "快速生成",
        "--comment-count",
        "50",
        "--media-type",
        "image",
        "--media-url",
        "https://example.com/image.jpg",
        "--tag",
        "快速生成",
    ]

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

    assert completed.returncode == 0, completed.stderr
    result = json.loads(completed.stdout)
    assert result["status"] == "ok"
    assert result["action"] == "fast-single-post"
    assert result["counts"]["posts"] == 1
    assert result["counts"]["comments"] == 50

    data_path = Path(result["data"])
    html_path = Path(result["html"])
    assert data_path.exists()
    assert html_path.exists()
    data = json.loads(data_path.read_text(encoding="utf-8"))
    post = data["posts"]["items"][0]
    assert len(post["comments"]) == 50
    assert post["comments"][-1]["id"] == "comment-050"
    assert post["media"]["mediaUrl"] == "https://example.com/image.jpg"
    assert "disclosure" not in post
    html = html_path.read_text(encoding="utf-8")
    assert '"id": "comment-050"' in html
    assert 'id="openSettings"' in html

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


def test_fast_single_post_spec_separates_total_and_displayed_comments(
    tmp_path: Path,
) -> None:
    spec_path = tmp_path / "facebook-spec.json"
    spec_path.write_text(
        json.dumps(
            {
                "author": "示例账号",
                "verified": True,
                "text": "今天分享一次重要会晤，双方将继续推进对话与合作。",
                "topic": "中美元首会晤",
                "tags": ["中美关系"],
                "displayCommentCount": 20,
                "totalCommentCount": 10000,
                "commentTopics": ["中美经济影响", "两岸关系"],
                "comments": ["关税调整将直接影响企业成本和供应链安排。"],
                "likes": 214000,
                "shares": 38600,
            },
            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 == 0, completed.stderr
    result = json.loads(completed.stdout)
    data = json.loads(Path(result["data"]).read_text(encoding="utf-8"))
    post = data["posts"]["items"][0]
    assert post["metrics"]["comments"] == 10000
    assert len(post["comments"]) == 20
    assert post["comments"][0]["text"] == "关税调整将直接影响企业成本和供应链安排。"
    assert any(topic in post["comments"][1]["text"] for topic in ("中美经济影响", "两岸关系"))
    assert post["media"]["type"] == "river"
    assert result["counts"]["comments"] == 20


def test_fast_single_post_rejects_english_only_post(tmp_path: Path) -> None:
    spec_path = tmp_path / "facebook-english-spec.json"
    spec_path.write_text(
        json.dumps(
            {
                "author": "Example",
                "text": "This post is written only in English.",
            }
        ),
        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 "Simplified Chinese" in result["error"]

    spec_path.write_text(
        json.dumps(
            {
                "author": "Example",
                "text": "这是一条中文动态。",
                "comments": ["This comment is written only in English."],
            },
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )
    comment_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 comment_completed.returncode == 2
    assert "Simplified Chinese" in json.loads(comment_completed.stdout)["error"]