from __future__ import annotations

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

import pytest

BACKEND_ROOT = Path(__file__).resolve().parents[1]
SEED_SKILLS_ROOT = (
    BACKEND_ROOT / "app" / "gateway" / "routers" / "_agentfx_seed_assets" / "skills"
)
LIB_PATH = SEED_SKILLS_ROOT / "task-report-lib" / "convert_lib.py"


def _load_convert_lib():
    spec = importlib.util.spec_from_file_location("task_report_convert_lib", LIB_PATH)
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


@pytest.fixture(scope="module")
def convert_lib():
    return _load_convert_lib()


def _write_hits(path: Path, *, marker: str = "甲") -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        json.dumps(
            {
                "query": f"{marker} 地区 舆情",
                "results": [
                    {
                        "title": f"{marker}事件进展",
                        "snippet": f"{marker}地区发布最新通报。",
                        "url": f"https://example.test/{marker}",
                        "published_at": "2026-09-24",
                    },
                    {
                        "name": f"{marker}背景资料",
                        "content": f"有关{marker}地区的人文与信息环境资料。",
                        "link": f"https://example.test/{marker}/background",
                    },
                ],
            },
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )


def _records(path: Path) -> list[dict[str, object]]:
    payload = json.loads(path.read_text(encoding="utf-8"))
    assert isinstance(payload, list)
    return payload


def _content_by_category(path: Path) -> dict[str, object]:
    return {
        str(record["categoryType"]): json.loads(str(record["contentJson"]))
        for record in _records(path)
    }


def _assert_dashboard_shapes(contents: dict[str, object], evidence_marker: str) -> None:
    targets = contents["敌情目标 JSON"]
    assert isinstance(targets, dict)
    assert set(targets) >= {"organization", "human"}
    assert isinstance(targets["organization"], list) and targets["organization"]
    assert set(targets["organization"][0]) >= {"name", "text", "etype", "reason"}

    events = contents["敌情关键事件 JSON"]
    assert isinstance(events, list) and events
    assert set(events[0]) >= {"topic", "count", "events"}
    assert isinstance(events[0]["events"], list) and events[0]["events"]
    assert set(events[0]["events"][0]) >= {
        "recUuid",
        "date",
        "publish_time",
        "link",
        "title",
        "content",
    }

    enemy_narratives = contents["敌情对手叙事 JSON"]
    assert isinstance(enemy_narratives, list) and enemy_narratives
    assert set(enemy_narratives[0]) >= {
        "topic",
        "narrator",
        "intention",
        "topicType",
        "summary",
    }

    speeches = contents["我情口径 JSON"]
    assert isinstance(speeches, dict)
    assert set(speeches) >= {"selfSpeech", "selfConsistents"}
    assert isinstance(speeches["selfSpeech"], list) and speeches["selfSpeech"]
    assert set(speeches["selfSpeech"][0]) >= {
        "trendSpeechTitle",
        "trendSpeechLike",
        "trendSpeechContent",
        "narrativeDirection",
        "trendSpeechTime",
    }

    ours_narratives = contents["我情叙事 JSON"]
    assert isinstance(ours_narratives, list) and ours_narratives
    assert set(ours_narratives[0]) >= {"标题", "意图", "标签", "内容"}
    assert isinstance(contents["我情上级要求 JSON"], str)

    for category in ("信息环境 JSON", "人文环境 JSON"):
        section_map = contents[category]
        assert isinstance(section_map, dict)
        sections = [value for key, value in section_map.items() if key != "type"]
        assert sections and all(isinstance(value, list) and value for value in sections)

    strengths = contents["优劣势分析 JSON"]
    assert isinstance(strengths, dict)
    assert set(strengths) >= {"优势", "劣势"}
    assert all(set(items[0]) >= {"title", "content"} for items in strengths.values())

    constraints = contents["条件与限制 JSON"]
    assert isinstance(constraints, dict) and constraints
    assert all(isinstance(value, list) and value for value in constraints.values())

    risks = contents["风险及应对 JSON"]
    assert isinstance(risks, dict)
    assert set(risks) >= {"风险分析", "应对方案"}
    assert all(isinstance(value, list) and value for value in risks.values())

    decision = contents["决心建议 JSON"]
    assert isinstance(decision, dict)
    assert set(decision) >= {"决心建议", "实施要点", "决心建议总结"}
    decision_core = decision["决心建议"]
    assert isinstance(decision_core, dict)
    assert set(decision_core) >= {"任务标题", "任务简介", "作战目的", "打击目标"}
    assert decision_core["打击目标"]
    assert set(decision_core["打击目标"][0]) >= {
        "目标名称",
        "目标描述",
        "选择理由",
        "潜在影响",
        "推荐依据",
    }

    disposal = contents["处置建议 JSON"]
    assert isinstance(disposal, dict)
    assert set(disposal) >= {"目标", "处置建议总结", "处置建议"}
    assert disposal["处置建议"]
    assert set(disposal["处置建议"][0]) >= {
        "建议内容",
        "处置方式与步骤",
        "口径基调",
        "预期效果",
        "处置主体",
        "立场与原则",
    }
    assert set(disposal["处置建议"][0]["处置主体"]) >= {"主责单位", "配合单位"}

    plans = contents["行动方案 JSON"]
    assert isinstance(plans, list) and plans
    assert set(plans[0]) >= {
        "trendPlanDesc",
        "createTime",
        "esId",
        "trendPlanName",
        "updateTime",
        "id",
        "trendPlanGoal",
        "trendPlanAction",
        "taskId",
    }

    for category, content in contents.items():
        serialized = json.dumps(content, ensure_ascii=False)
        assert evidence_marker in serialized, category


def test_resolve_hit_files_maps_virtual_user_data_to_workspace(tmp_path: Path, convert_lib) -> None:
    user_data = tmp_path / "thread" / "user-data"
    workspace = user_data / "workspace"
    output = user_data / "outputs"
    workspace.mkdir(parents=True)
    output.mkdir()
    hit_file = workspace / "hits" / "q01.json"
    _write_hits(hit_file)

    files, warnings = convert_lib.resolve_hit_files(
        "/mnt/user-data/workspace/hits",
        cwd=workspace,
    )

    assert files == [hit_file.resolve()]
    assert warnings == []


def test_all_group_writes_fourteen_unicode_records(tmp_path: Path, convert_lib) -> None:
    workspace = tmp_path / "user-data" / "workspace"
    (tmp_path / "user-data" / "outputs").mkdir(parents=True)
    _write_hits(workspace / "hits" / "q01.json", marker="海峡")

    result = convert_lib.convert_reports(
        group="all",
        task_id="T-001",
        task_name="海峡态势研判",
        task_content="分析公开信息与风险信号",
        hits_dir="/mnt/user-data/workspace/hits",
        output="/mnt/user-data/outputs/task-reports.json",
        cwd=workspace,
    )

    assert result["success"] is True
    assert result["hits"] == 2
    output_path = Path(str(result["output"]))
    records = _records(output_path)
    assert [record["categoryType"] for record in records] == convert_lib.CATEGORY_TYPES
    assert len(records) == 14
    assert all(record["taskId"] == "T-001" for record in records)
    assert "海峡" in output_path.read_text(encoding="utf-8")
    assert "\\u6d77" not in output_path.read_text(encoding="utf-8")
    _assert_dashboard_shapes(_content_by_category(output_path), convert_lib.EVIDENCE_MARKER)


def test_page_group_replaces_only_its_categories(tmp_path: Path, convert_lib) -> None:
    workspace = tmp_path / "user-data" / "workspace"
    outputs = tmp_path / "user-data" / "outputs"
    outputs.mkdir(parents=True)
    hit_file = workspace / "hits" / "q01.json"
    _write_hits(hit_file, marker="旧材料")
    output = outputs / "task-reports.json"

    first = convert_lib.convert_reports(
        group="all",
        task_id="T-002",
        task_name="测试任务",
        task_content="测试内容",
        hits_dir=str(hit_file.parent),
        output=str(output),
        cwd=workspace,
    )
    assert first["success"] is True
    before = {record["categoryType"]: record for record in _records(output)}

    _write_hits(hit_file, marker="新敌情")
    second = convert_lib.convert_reports(
        group="enemy",
        task_id="T-002",
        task_name="测试任务",
        task_content="测试内容",
        hits=[str(hit_file)],
        output=str(output),
        cwd=workspace,
    )
    assert second["success"] is True
    after = {record["categoryType"]: record for record in _records(output)}

    assert set(after) == set(convert_lib.CATEGORY_TYPES)
    assert after["信息环境 JSON"] == before["信息环境 JSON"]
    assert "新敌情" in str(after["敌情关键事件 JSON"]["contentJson"])
    assert "旧材料" not in str(after["敌情关键事件 JSON"]["contentJson"])


def test_page_group_rejects_existing_records_from_another_task(tmp_path: Path, convert_lib) -> None:
    workspace = tmp_path / "user-data" / "workspace"
    outputs = tmp_path / "user-data" / "outputs"
    outputs.mkdir(parents=True)
    hit_file = workspace / "hits" / "q01.json"
    _write_hits(hit_file, marker="原任务")
    output = outputs / "task-reports.json"
    initial = convert_lib.convert_reports(
        group="all",
        task_id="TASK-A",
        hits=[str(hit_file)],
        output=str(output),
        cwd=workspace,
    )
    assert initial["success"] is True
    original = output.read_bytes()

    _write_hits(hit_file, marker="另一任务")
    result = convert_lib.convert_reports(
        group="enemy",
        task_id="TASK-B",
        hits=[str(hit_file)],
        output=str(output),
        cwd=workspace,
    )

    assert result["success"] is False
    assert any("TASK-A" in warning and "TASK-B" in warning for warning in result["warnings"])
    assert output.read_bytes() == original


def test_zero_hits_fails_without_overwriting_existing_output(tmp_path: Path, convert_lib) -> None:
    workspace = tmp_path / "user-data" / "workspace"
    outputs = tmp_path / "user-data" / "outputs"
    workspace.mkdir(parents=True)
    outputs.mkdir()
    hits_dir = workspace / "hits"
    hits_dir.mkdir()
    (hits_dir / "q01.json").write_text(
        '{"query":"没有结果的检索","results":[]}',
        encoding="utf-8",
    )
    output = outputs / "task-reports.json"
    original = '[{"sentinel": true}]\n'
    output.write_text(original, encoding="utf-8")

    result = convert_lib.convert_reports(
        group="all",
        task_id="T-003",
        task_name="无命中任务",
        task_content="不能编造证据",
        hits_dir="/mnt/user-data/workspace/hits",
        output="/mnt/user-data/outputs/task-reports.json",
        cwd=workspace,
    )

    assert result["success"] is False
    assert result["hits"] == 0
    assert result["warnings"]
    assert output.read_text(encoding="utf-8") == original


@pytest.mark.parametrize("unsafe_kind", ["hits", "output"])
def test_paths_outside_user_data_are_rejected(
    tmp_path: Path,
    convert_lib,
    unsafe_kind: str,
) -> None:
    workspace = tmp_path / "thread" / "user-data" / "workspace"
    outputs = workspace.parent / "outputs"
    outputs.mkdir(parents=True)
    safe_hit = workspace / "hits" / "q01.json"
    _write_hits(safe_hit)
    outside = tmp_path / "outside"
    outside.mkdir()
    outside_hit = outside / "q01.json"
    _write_hits(outside_hit, marker="越界")
    outside_output = outside / "task-reports.json"

    result = convert_lib.convert_reports(
        group="all",
        task_id="T-SAFE",
        hits=[str(outside_hit if unsafe_kind == "hits" else safe_hit)],
        output=str(outside_output if unsafe_kind == "output" else outputs / "task-reports.json"),
        cwd=workspace,
    )

    assert result["success"] is False
    assert any("user-data" in warning for warning in result["warnings"])
    assert not outside_output.exists()


def test_symlink_escape_is_rejected(tmp_path: Path, convert_lib) -> None:
    workspace = tmp_path / "thread" / "user-data" / "workspace"
    (workspace.parent / "outputs").mkdir(parents=True)
    outside = tmp_path / "outside"
    outside.mkdir()
    _write_hits(outside / "q01.json", marker="链接越界")
    link = workspace / "hits"
    workspace.mkdir(exist_ok=True)
    try:
        link.symlink_to(outside, target_is_directory=True)
    except OSError as exc:
        pytest.skip(f"symlink creation is unavailable: {exc}")

    result = convert_lib.convert_reports(
        group="all",
        task_id="T-LINK",
        hits_dir=str(link),
        output="/mnt/user-data/outputs/task-reports.json",
        cwd=workspace,
    )

    assert result["success"] is False
    assert any("user-data" in warning for warning in result["warnings"])


def test_thin_cli_prints_one_json_line(tmp_path: Path) -> None:
    workspace = tmp_path / "user-data" / "workspace"
    outputs = tmp_path / "user-data" / "outputs"
    outputs.mkdir(parents=True)
    hit_file = workspace / "hits" / "q01.json"
    _write_hits(hit_file, marker="命令行")
    script = SEED_SKILLS_ROOT / "task-report-env" / "scripts" / "convert.py"
    output = outputs / "task-reports.json"

    completed = subprocess.run(
        [
            sys.executable,
            str(script),
            "--task-id",
            "T-CLI",
            "--task-name",
            "命令行测试",
            "--task-content",
            "验证标准输出",
            "--hits",
            str(hit_file),
            "--output",
            str(output),
        ],
        cwd=workspace,
        capture_output=True,
        text=True,
        encoding="utf-8",
        check=False,
    )

    assert completed.returncode == 0, completed.stderr
    lines = completed.stdout.splitlines()
    assert len(lines) == 1
    result = json.loads(lines[0])
    assert result["success"] is True
    assert result["hits"] == 2
    assert [record["categoryType"] for record in _records(output)] == convert_lib_env_categories()


def test_task_context_cli_treats_shell_metacharacters_as_data(tmp_path: Path, convert_lib) -> None:
    workspace = tmp_path / "user-data" / "workspace"
    outputs = tmp_path / "user-data" / "outputs"
    outputs.mkdir(parents=True)
    _write_hits(workspace / "hits" / "q01.json", marker="上下文")
    task_name = '名称"; $(New-Item PWNED); `whoami` $env:PATH'
    task_content = "内容'; rm -rf ignored; && echo not-executed"
    (workspace / "task-context.json").write_text(
        json.dumps(
            {
                "taskId": "T-CONTEXT",
                "taskName": task_name,
                "taskContent": task_content,
            },
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )
    script = SEED_SKILLS_ROOT / "task-report-build" / "scripts" / "convert.py"

    completed = subprocess.run(
        [
            sys.executable,
            str(script),
            "--task-context",
            "/mnt/user-data/workspace/task-context.json",
        ],
        cwd=workspace,
        capture_output=True,
        text=True,
        encoding="utf-8",
        check=False,
    )

    assert completed.returncode == 0, completed.stderr
    assert len(completed.stdout.splitlines()) == 1
    result = json.loads(completed.stdout)
    output = Path(result["output"])
    decision = _content_by_category(output)["决心建议 JSON"]
    assert decision["决心建议"]["任务标题"] == task_name
    assert decision["决心建议"]["任务简介"] == task_content
    assert not (tmp_path / "PWNED").exists()
    _assert_dashboard_shapes(_content_by_category(output), convert_lib.EVIDENCE_MARKER)


def test_task_context_path_is_confined(tmp_path: Path, convert_lib) -> None:
    workspace = tmp_path / "user-data" / "workspace"
    (workspace.parent / "outputs").mkdir(parents=True)
    outside = tmp_path / "task-context.json"
    outside.write_text('{"taskId":"OUTSIDE"}', encoding="utf-8")

    with pytest.raises(convert_lib.PathConfinementError):
        convert_lib.load_task_context(outside, cwd=workspace)


def test_random_workspace_directory_is_not_a_trusted_user_data_root(
    tmp_path: Path,
    convert_lib,
) -> None:
    fake_workspace = tmp_path / "workspace"
    fake_workspace.mkdir()

    assert convert_lib.discover_user_data_root(fake_workspace) is None
    with pytest.raises(convert_lib.PathConfinementError):
        convert_lib.resolve_sandbox_path("/mnt/user-data/outputs/task-reports.json", cwd=fake_workspace)


def convert_lib_env_categories() -> list[str]:
    return ["信息环境 JSON", "人文环境 JSON"]