from pathlib import Path
import sys
from typing import Any

import pytest
import yaml
from fastapi import HTTPException

ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from apps.cli.run_samples_autofill import main as samples_autofill_main
from apps.cli.run_rnaseq_de import build_de_plan
from apps.web.vue3.main import _write_samples_draft_rows


def _seed_yaml(path: Path) -> None:
    path.write_text(
        yaml.safe_dump(
            {
                "samples": {},
                "groups": {},
                "compare": [],
                "experiment": {},
                "de": {"strategy": "auto"},
            },
            allow_unicode=True,
            sort_keys=False,
        ),
        encoding="utf-8",
    )


def _row(sample: str, condition: str, time: Any, rep: int, role: str) -> dict[str, Any]:
    return {
        "run": sample,
        "sample_id": sample,
        "fastq_1": f"{sample}_R1.fq.gz",
        "fastq_2": f"{sample}_R2.fq.gz",
        "condition": condition,
        "time": time,
        "replicate": str(rep),
        "library_strandness": "reverse",
        "group_role": role,
        "include": True,
    }


def _write_case(tmp_path: Path, name: str, rows: list[dict[str, Any]], design: str) -> dict[str, Any]:
    path = tmp_path / f"{name}.samples.yaml"
    _seed_yaml(path)
    _write_samples_draft_rows(path, rows, design=design)
    return yaml.safe_load(path.read_text(encoding="utf-8"))


def test_time_series_requires_control_and_generates_baseline_comparisons(tmp_path):
    rows = [
        _row("T0_r1", "mock", "0h", 1, "control"),
        _row("T0_r2", "mock", "0h", 2, "control"),
        _row("T1_r1", "mock", "1h", 1, "compare"),
        _row("T1_r2", "mock", "1h", 2, "compare"),
        _row("T3_r1", "mock", "3h", 1, "compare"),
        _row("T3_r2", "mock", "3h", 2, "compare"),
    ]

    data = _write_case(tmp_path, "time_series", rows, "time_series")

    assert data["experiment"]["control_group"] == "0h"
    assert data["compare"] == [["1h", "0h"], ["3h", "0h"]]
    assert data["de"]["pairwise_modes"] == ["baseline"]


def test_time_series_accepts_numeric_zero_time(tmp_path):
    rows = [
        _row("T0_r1", "mock", 0, 1, "control"),
        _row("T0_r2", "mock", 0, 2, "control"),
        _row("T1_r1", "mock", "1h", 1, "compare"),
        _row("T1_r2", "mock", "1h", 2, "compare"),
    ]

    data = _write_case(tmp_path, "time_series_numeric_zero", rows, "time_series")

    assert data["experiment"]["design"] == "time_series"
    assert data["experiment"]["control_group"] == "0"
    assert data["groups"]["T0_r1"] == "0"
    assert data["compare"] == [["1h", "0"]]


def test_manual_design_hint_is_ignored_for_time_series_table(tmp_path):
    rows = [
        _row("T0_r1", "mock", "0h", 1, "control"),
        _row("T0_r2", "mock", "0h", 2, "control"),
        _row("T1_r1", "mock", "1h", 1, "compare"),
        _row("T1_r2", "mock", "1h", 2, "compare"),
        _row("T3_r1", "mock", "3h", 1, "compare"),
        _row("T3_r2", "mock", "3h", 2, "compare"),
    ]

    data = _write_case(tmp_path, "manual_hint_time_series", rows, "manual")

    assert data["experiment"]["design"] == "time_series"
    assert data["compare"] == [["1h", "0h"], ["3h", "0h"]]


def test_two_group_without_control_is_saved_with_de_warning(tmp_path):
    rows = [
        _row("CK_r1", "CK", "", 1, "compare"),
        _row("CK_r2", "CK", "", 2, "compare"),
        _row("salt_r1", "salt", "", 1, "compare"),
        _row("salt_r2", "salt", "", 2, "compare"),
    ]

    data = _write_case(tmp_path, "two_group_no_control", rows, "two_group")

    assert data["samples"]["CK_r1"]["group_role"] == "compare"
    assert data["groups"] == {}
    assert data["compare"] == []
    assert data["de"] == {}
    assert any("DE design pending user input" in item for item in data["_autofill"]["warnings"])


def test_samples_with_blank_group_role_are_saved_with_de_warning(tmp_path):
    rows = [
        _row("CK_r1", "CK", "", 1, ""),
        _row("salt_r1", "salt", "", 1, ""),
    ]

    data = _write_case(tmp_path, "blank_group_roles", rows, "manual")

    assert data["samples"]["CK_r1"]["group_role"] == ""
    assert data["samples"]["salt_r1"]["group_role"] == ""
    assert data["groups"] == {}
    assert data["compare"] == []
    assert data["de"] == {}
    assert any("DE design pending user input" in item for item in data["_autofill"]["warnings"])


def test_multi_group_without_control_is_saved_with_de_warning(tmp_path):
    rows = [
        _row("root_r1", "root", "", 1, "compare"),
        _row("root_r2", "root", "", 2, "compare"),
        _row("stem_r1", "stem", "", 1, "compare"),
        _row("stem_r2", "stem", "", 2, "compare"),
        _row("leaf_r1", "leaf", "", 1, "compare"),
        _row("leaf_r2", "leaf", "", 2, "compare"),
    ]

    data = _write_case(tmp_path, "multi_group_no_control", rows, "multi_group")

    assert data["samples"]["leaf_r1"]["group_role"] == "compare"
    assert data["groups"] == {}
    assert data["compare"] == []
    assert data["de"] == {}
    assert any("DE design pending user input" in item for item in data["_autofill"]["warnings"])


def test_multi_group_with_control_generates_each_group_vs_control(tmp_path):
    rows = [
        _row("root_r1", "root", "", 1, "control"),
        _row("root_r2", "root", "", 2, "control"),
        _row("stem_r1", "stem", "", 1, "compare"),
        _row("stem_r2", "stem", "", 2, "compare"),
        _row("leaf_r1", "leaf", "", 1, "compare"),
        _row("leaf_r2", "leaf", "", 2, "compare"),
    ]

    data = _write_case(tmp_path, "multi_group_control", rows, "multi_group")

    assert data["compare"] == [["leaf", "root"], ["stem", "root"]]
    assert data["de"]["pairwise_modes"] == ["control"]


def test_manual_design_hint_is_ignored_for_group_role_table(tmp_path):
    rows = [
        _row("root_r1", "root", "", 1, "control"),
        _row("stem_r1", "stem", "", 1, "compare"),
    ]

    data = _write_case(tmp_path, "manual_hint", rows, "manual")

    assert data["experiment"]["design"] == "two_group"
    assert data["compare"] == [["stem", "root"]]


def test_confirmed_samples_draft_rows_drop_resolved_group_role_warning(tmp_path):
    path = tmp_path / "confirmed_warning_cleanup.samples.yaml"
    _seed_yaml(path)
    data = yaml.safe_load(path.read_text(encoding="utf-8"))
    data["_autofill"] = {
        "warnings": [
            "DE design pending user input: samples.yaml 缺少 group_role；请在样本表中为这些样本标记 control/compare/ignore: root_r1",
            "Sample info table contains 1 run ID not found in current FASTQ path",
        ]
    }
    path.write_text(yaml.safe_dump(data, allow_unicode=True, sort_keys=False), encoding="utf-8")
    rows = [
        _row("root_r1", "root", "", 1, "control"),
        _row("stem_r1", "stem", "", 1, "compare"),
    ]

    _write_samples_draft_rows(path, rows, design="manual")
    updated = yaml.safe_load(path.read_text(encoding="utf-8"))

    assert updated["compare"] == [["stem", "root"]]
    assert updated["_autofill"]["warnings"] == [
        "Sample info table contains 1 run ID not found in current FASTQ path"
    ]


def test_samples_autofill_confirm_cleans_resolved_warning_but_keeps_metadata_warning(tmp_path, monkeypatch):
    input_dir = tmp_path / "input"
    workspace = tmp_path / "workspace"
    final_dir = workspace / "final"
    final_dir.mkdir(parents=True)
    input_dir.mkdir()
    draft_path = final_dir / "samples.autofill.draft.yaml"
    rows = [
        _row("root_r1", "root", "", 1, "control"),
        _row("stem_r1", "stem", "", 1, "compare"),
    ]
    payload = {
        "samples": {
            row["sample_id"]: {
                "run": row["run"],
                "r1": row["fastq_1"],
                "r2": row["fastq_2"],
                "condition": row["condition"],
                "time": row["time"],
                "replicate": row["replicate"],
                "library_strandness": row["library_strandness"],
                "group_role": row["group_role"],
                "include": True,
            }
            for row in rows
        },
        "groups": {},
        "experiment": {},
        "compare": [],
        "de": {},
        "_autofill": {
            "rows": rows,
            "warnings": [
                "DE design pending user input: samples.yaml 缺少 group_role；请在样本表中为这些样本标记 control/compare/ignore: root_r1",
                "Sample info table contains 1 run ID not found in current FASTQ path",
            ],
        },
    }
    draft_path.write_text(yaml.safe_dump(payload, allow_unicode=True, sort_keys=False), encoding="utf-8")
    monkeypatch.setattr(
        sys,
        "argv",
        [
            "run_samples_autofill.py",
            "--input-dir",
            str(input_dir),
            "--workspace",
            str(workspace),
            "--confirm",
            "--draft",
            str(draft_path),
        ],
    )

    assert samples_autofill_main() == 0
    confirmed = yaml.safe_load((input_dir / "samples.yaml").read_text(encoding="utf-8"))

    assert confirmed["compare"] == [["stem", "root"]]
    assert confirmed["_autofill"]["warnings"] == [
        "Sample info table contains 1 run ID not found in current FASTQ path"
    ]


def test_samples_autofill_confirm_allows_pending_de_roles(tmp_path, monkeypatch):
    input_dir = tmp_path / "input"
    workspace = tmp_path / "workspace"
    final_dir = workspace / "final"
    final_dir.mkdir(parents=True)
    input_dir.mkdir()
    draft_path = final_dir / "samples.autofill.draft.yaml"
    rows = [
        _row("CK_r1", "CK", "", 1, ""),
        _row("salt_r1", "salt", "", 1, ""),
    ]
    payload = {
        "samples": {
            row["sample_id"]: {
                "run": row["run"],
                "r1": row["fastq_1"],
                "r2": row["fastq_2"],
                "condition": row["condition"],
                "time": row["time"],
                "replicate": row["replicate"],
                "library_strandness": row["library_strandness"],
                "group_role": row["group_role"],
                "include": True,
            }
            for row in rows
        },
        "groups": {},
        "experiment": {},
        "compare": [],
        "de": {},
        "_autofill": {"rows": rows, "warnings": []},
    }
    draft_path.write_text(yaml.safe_dump(payload, allow_unicode=True, sort_keys=False), encoding="utf-8")
    monkeypatch.setattr(
        sys,
        "argv",
        [
            "run_samples_autofill.py",
            "--input-dir",
            str(input_dir),
            "--workspace",
            str(workspace),
            "--confirm",
            "--draft",
            str(draft_path),
        ],
    )

    assert samples_autofill_main() == 0
    confirmed = yaml.safe_load((input_dir / "samples.yaml").read_text(encoding="utf-8"))

    assert confirmed["samples"]["CK_r1"]["group_role"] == ""
    assert confirmed["groups"] == {}
    assert confirmed["compare"] == []
    assert confirmed["de"] == {}
    assert any("DE design pending user input" in item for item in confirmed["_autofill"]["warnings"])


def test_build_de_plan_honors_configured_compare_and_auto_modes():
    matrix_samples = ["root_r1", "root_r2", "stem_r1", "stem_r2", "leaf_r1", "leaf_r2"]
    groups = {
        "root_r1": "root",
        "root_r2": "root",
        "stem_r1": "stem",
        "stem_r2": "stem",
        "leaf_r1": "leaf",
        "leaf_r2": "leaf",
    }
    samples = {
        "root_r1": {"condition": "root", "group_role": "control"},
        "root_r2": {"condition": "root", "group_role": "control"},
        "stem_r1": {"condition": "stem", "group_role": "compare"},
        "stem_r2": {"condition": "stem", "group_role": "compare"},
        "leaf_r1": {"condition": "leaf", "group_role": "compare"},
        "leaf_r2": {"condition": "leaf", "group_role": "compare"},
    }

    configured = build_de_plan(
        {
            "samples": samples,
            "compare": [["leaf", "root"]],
            "de": {"strategy": "auto", "design": "multi_group"},
            "experiment": {"design": "multi_group"},
        },
        matrix_samples,
        groups,
    )
    assert configured["comparisons"] == [["leaf", "root"]]
    assert configured["comparison_mode"] == "configured"

    no_control_samples = {
        sample: {**meta, "group_role": "compare"}
        for sample, meta in samples.items()
    }
    with pytest.raises(ValueError, match="唯一对照组"):
        build_de_plan(
            {"samples": no_control_samples, "de": {"strategy": "auto", "design": "multi_group"}},
            matrix_samples,
            groups,
        )

    control = build_de_plan(
        {
            "samples": samples,
            "de": {"strategy": "auto", "design": "multi_group", "control_group": "root"},
            "experiment": {"design": "multi_group", "control_group": "root"},
        },
        matrix_samples,
        groups,
    )
    assert control["comparisons"] == [["leaf", "root"], ["stem", "root"]]
    assert control["comparison_mode"] == "group_role_control"

    missing_role_samples = {
        sample: {"condition": meta["condition"]}
        for sample, meta in samples.items()
    }
    with pytest.raises(ValueError, match="缺少 group_role"):
        build_de_plan(
            {"samples": missing_role_samples, "de": {"strategy": "auto"}},
            matrix_samples,
            groups,
        )