"""tias-seg eval の結合テスト.

仕様（実装サマリーを仕様として扱う．出力の規則は SPEC_01「実験の出力」「回収ホワイトリスト」）:

- 出力: `metrics.json`・`log.txt`・`config.yaml`・`figures/seg_iou_hist_<評価セット>.png`
  （回収対象）と `predictions.csv`（stem を含む．回収対象外）
- 最後に実験 ID を標準出力へ出す
- 回収対象・標準出力・ログにファイル名（stem）を出さない
- 環境変数 `TIAS_SEG_DATA_ROOT`・`TIAS_SEG_MODEL_ROOT` で `data.root`・`models.root` を上書きする
- 入力の問題（正解マスクに前景が無い等）は終了コード 1（ログに評価セット名と何件目かを出し，
  ファイル名は出さない）
"""

from __future__ import annotations

import contextlib
import csv
import io
import json
import logging
import re
import shutil
from pathlib import Path

import numpy as np
import pytest
import yaml
from PIL import Image

import tias_judge_model.cli.seg as cli
from tias_judge_model.remote.whitelist import is_pull_allowed

SECRET = "SECRETPT"
GIT = "abc1234"
PER_SET = 3
NAME = "seg_eval_tiny"
SETS = ["train", "test_cropped", "test_extra_cropped", "test_fullsize"]
RECOVERABLE = {"metrics.json", "log.txt", "config.yaml"} | {
    f"figures/seg_iou_hist_{s}.png" for s in SETS
}


def _invoke(argv: list[str]) -> tuple[int, str, str]:
    root_logger = logging.getLogger()
    old_level = root_logger.level
    root_logger.setLevel(logging.INFO)
    out, err = io.StringIO(), io.StringIO()
    handler = logging.StreamHandler(err)
    root_logger.addHandler(handler)
    try:
        with contextlib.redirect_stdout(out):
            rc = cli.main(argv)
    finally:
        root_logger.removeHandler(handler)
        root_logger.setLevel(old_level)
    return rc, out.getvalue(), err.getvalue()


def _config_doc() -> dict:
    # data.root・models.root は環境変数で上書きする（設定には存在しない場所を書く）
    return {
        "data": {
            "root": "does/not/exist",
            "sets": [
                {"name": "train", "dir": "train", "kind": "cropped"},
                {"name": "test_cropped", "dir": "test/cropped", "kind": "cropped"},
                {"name": "test_extra_cropped", "dir": "test_extra/cropped", "kind": "cropped"},
                {"name": "test_fullsize", "dir": "test/fullsize", "kind": "fullsize"},
            ],
        },
        "models": {"root": "does/not/exist", "detection": "Detection_Normal.onnx"},
    }


def _rename_stems(root: Path) -> None:
    n = 0
    for label_dir in sorted(root.rglob("label")):
        sample_dir = label_dir.parent / "sample"
        for label in sorted(label_dir.iterdir()):
            sample = sample_dir / label.name
            new = f"{SECRET}{n:04d}{label.suffix}"
            label.rename(label_dir / new)
            sample.rename(sample_dir / new)
            n += 1


@pytest.fixture(scope="module")
def run(tmp_path_factory: pytest.TempPathFactory, dummy_seg_models: dict, script_loader) -> dict:
    tmp = tmp_path_factory.mktemp("seg_cli")
    script_loader("make_synthetic_segmentation").generate(tmp / "syn", PER_SET, seed=1)
    data_root = tmp / "syn" / "input"
    _rename_stems(data_root)
    config = tmp / f"{NAME}.yaml"
    config.write_text(yaml.safe_dump(_config_doc()), encoding="utf-8")
    out_root = tmp / "runs"
    mp = pytest.MonkeyPatch()
    mp.setenv("TIAS_SEG_DATA_ROOT", str(data_root))
    mp.setenv("TIAS_SEG_MODEL_ROOT", str(dummy_seg_models["root"]))
    mp.setenv("TIAS_GIT_COMMIT", GIT)
    try:
        rc, stdout, stderr = _invoke(["eval", "--config", str(config), "--out-root", str(out_root)])
    finally:
        mp.undo()
    run_dirs = [p for p in out_root.iterdir() if p.is_dir()] if out_root.exists() else []
    return {
        "tmp": tmp,
        "data_root": data_root,
        "models": dummy_seg_models,
        "rc": rc,
        "stdout": stdout,
        "stderr": stderr,
        "run_dirs": run_dirs,
        "run_dir": run_dirs[0] if run_dirs else None,
    }


def test_exit_code_and_single_run_dir(run: dict) -> None:
    assert run["rc"] == 0, run["stderr"][-2000:]
    assert len(run["run_dirs"]) == 1


def test_run_id_is_last_stdout_line(run: dict) -> None:
    run_id = run["run_dir"].name
    assert re.fullmatch(rf"\d{{8}}-\d{{6}}_{NAME}_{GIT}", run_id)
    lines = [ln for ln in run["stdout"].splitlines() if ln.strip()]
    assert lines == [run_id]


def test_outputs_and_pull_whitelist(run: dict) -> None:
    d = run["run_dir"]
    files = {p.relative_to(d).as_posix() for p in d.rglob("*") if p.is_file()}
    assert files == RECOVERABLE | {"predictions.csv"}
    for name in RECOVERABLE:
        assert is_pull_allowed(name, with_checkpoints=False), name
    assert not is_pull_allowed("predictions.csv", with_checkpoints=True)


def test_no_file_names_in_recoverable_outputs_stdout_or_log(run: dict) -> None:
    d = run["run_dir"]
    for name in RECOVERABLE:
        data = (d / name).read_bytes()
        assert SECRET.encode() not in data, name
    assert SECRET not in run["stdout"]
    assert SECRET not in run["stderr"]


def test_predictions_csv_has_per_image_rows(run: dict) -> None:
    with (run["run_dir"] / "predictions.csv").open(encoding="utf-8") as f:
        rows = list(csv.DictReader(f))
    # cropped 3 セット × 1 方式 ＋ fullsize × 3 方式
    assert len(rows) == PER_SET * (3 + 3)
    assert all(r["stem"].startswith(SECRET) for r in rows)


def test_metrics_structure(run: dict) -> None:
    m = json.loads((run["run_dir"] / "metrics.json").read_text(encoding="utf-8"))
    assert m["run_id"] == run["run_dir"].name
    assert m["git_commit"] == GIT
    assert m["detector"]["used"] is True
    assert set(m["sets"]) == set(SETS)
    for name in SETS:
        s = m["sets"][name]
        assert s["n_images"] == PER_SET
        expected = ["whole", "gt_bbox", "detector"] if name == "test_fullsize" else ["whole"]
        assert list(s["methods"]) == expected
        for method in s["methods"].values():
            for key in ("iou", "dice", "iou_no_post", "dice_no_post", "iou_other_channel"):
                summary = method[key]
                assert summary["n"] == PER_SET
                assert {"mean", "median", "p10", "p25", "min", "max", "n_below"} <= set(summary)
                assert summary["below_threshold"] == 0.8
    det = m["sets"]["test_fullsize"]["methods"]["detector"]["detection"]
    assert det["n_detected"] + det["n_not_detected"] == PER_SET
    assert {"total_seconds", "loading_seconds"} <= set(m["timing"])


def test_config_yaml_records_resolved_roots(run: dict) -> None:
    doc = yaml.safe_load((run["run_dir"] / "config.yaml").read_text(encoding="utf-8"))
    assert doc["run"]["command"] == "seg eval"
    assert doc["run"]["git_commit"] == GIT
    assert Path(doc["config"]["data"]["root"]) == run["data_root"]


def test_bad_label_fails_with_position_but_no_file_name(run: dict) -> None:
    tmp = run["tmp"]
    root = tmp / "bad" / "input"
    shutil.copytree(run["data_root"] / "train", root / "train")
    label = sorted((root / "train" / "label").iterdir())[0]
    with Image.open(label) as img:
        w, h = img.size
    Image.fromarray(np.zeros((h, w), np.uint8)).save(label)
    doc = _config_doc()
    doc["data"]["root"] = root.as_posix()
    doc["data"]["sets"] = [{"name": "train", "dir": "train", "kind": "cropped"}]
    doc["models"]["root"] = run["models"]["root"].as_posix()
    config = tmp / "seg_bad.yaml"
    config.write_text(yaml.safe_dump(doc), encoding="utf-8")
    out_root = tmp / "runs_bad"
    mp = pytest.MonkeyPatch()
    mp.delenv("TIAS_SEG_DATA_ROOT", raising=False)
    mp.delenv("TIAS_SEG_MODEL_ROOT", raising=False)
    try:
        rc, stdout, stderr = _invoke(["eval", "--config", str(config), "--out-root", str(out_root)])
    finally:
        mp.undo()
    assert rc == 1
    assert stdout.strip() == ""
    assert "1 件目" in stderr
    assert SECRET not in stderr
    (run_dir,) = [p for p in out_root.iterdir() if p.is_dir()]
    log = (run_dir / "log.txt").read_text(encoding="utf-8")
    assert "1 件目" in log
    assert SECRET not in log


def test_invalid_config_returns_1(run: dict) -> None:
    config = run["tmp"] / "seg_invalid.yaml"
    config.write_text(yaml.safe_dump({"data": {"root": "x", "sets": []}}), encoding="utf-8")
    rc, stdout, _ = _invoke(["eval", "--config", str(config), "--out-root", str(run["tmp"] / "r")])
    assert rc == 1
    assert stdout.strip() == ""