"""バンドルの書き出し（SPEC_03「バンドル」「CLI」の export・「計算サーバへの受け渡し」手順 1）.

必須対象 5（バンドル）の書き出し側: 書き出したバンドルを `open_bundle` で読めること，
同じ版から同じ manifest.csv（バイト列）が得られること，既にあるバンドルを上書きせず照合すること．

目録はダミーの行を直接登録して作り，ストアには小さいダミーのファイルを置く（取り込みを繰り返さず
速くするため．取り込みからの流れは `test_cli.py` の統合テストで確かめる）．臨床データは使わない．
"""

import hashlib
import json
import os
import shutil
import sqlite3
import stat
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path

import pytest

from tiasdata import kinds, store
from tiasdata.bundle import open_bundle
from tiasdata.datasets import freeze_version, load_definition, rebuild_manifest
from tiasdata.export import export_bundle
from tiasdata.repo import RepoState

NAME = "tongue_color"
COMMIT = "0123456789abcdef0123456789abcdef01234567"
NOW = datetime(2026, 10, 6, 9, 0, 0, tzinfo=UTC)
# 元のファイル名・フォルダ名に仕込む架空の患者 ID（バンドル・メッセージに出ないことを確かめる）
LEAK_KEY = "990077"

TONGUE_COLOR_TOML = """\
name = "tongue_color"
description = "舌色 5 分類＋程度（テスト）"

[select]
asset_kind = "tongue_crop_masked"
require = ["tongue_color"]

[columns]
tongue_color = ["label", "severity"]
"""

DATASET_JSON_KEYS = {
    "format",
    "name",
    "version",
    "created_at",
    "hub_commit",
    "files",
    "manifest_sha256",
}


def _tc(label: str, severity: str) -> str:
    return kinds.dump_annotation_value(
        kinds.validate_annotation_value("tongue_color", {"label": label, "severity": severity})
    )


@dataclass
class Env:
    conn: sqlite3.Connection
    root: Path
    store_dir: Path
    bundles_dir: Path
    datasets_dir: Path
    contents: dict[str, bytes]  # SHA-256 → 中身

    def add(self, n: int, subject_id: str, label: str = "red", severity: str = "mild") -> int:
        """撮影 1 つ・ファイル 1 つ（ストアに実体あり）・撮影に付いた舌色の所見 1 つを足す．"""
        capture_id = f"C{n:06d}"
        source = self.root / "source" / f"患者{LEAK_KEY}" / f"{n}_original.png"
        source.parent.mkdir(parents=True, exist_ok=True)
        content = f"dummy-image-{n}".encode() * (n + 1)
        source.write_bytes(content)
        stored = store.add_file(self.store_dir, source)
        self.contents[stored.sha256] = content
        self.conn.execute(
            "INSERT OR IGNORE INTO subjects (subject_id, created_at) VALUES (?, ?)",
            (subject_id, "2026-10-01T00:00:00Z"),
        )
        self.conn.execute(
            "INSERT INTO captures (capture_id, subject_id) VALUES (?, ?)", (capture_id, subject_id)
        )
        self.conn.execute(
            "INSERT INTO assets (sha256, kind, capture_id, ext, size_bytes, source, source_name,"
            " registered_at) VALUES (?, 'tongue_crop_masked', ?, 'png', ?, 'dummy_src', ?,"
            " '2026-10-01T00:00:00Z')",
            (stored.sha256, capture_id, len(content), f"患者{LEAK_KEY}/{n}_original.png"),
        )
        cursor = self.conn.execute(
            "INSERT INTO annotations (kind, capture_id, value, source, created_at)"
            " VALUES ('tongue_color', ?, ?, 'dummy_src', '2026-10-01T00:00:00Z')",
            (capture_id, _tc(label, severity)),
        )
        self.conn.commit()
        return int(cursor.lastrowid)

    def freeze(self, now: datetime = NOW):
        loaded = load_definition(self.datasets_dir, NAME)
        return freeze_version(
            self.conn, loaded, repo_state=RepoState(commit=COMMIT, uncommitted=0), now=now
        )

    def export(self, version: int = 1, *, bundles_dir: Path | None = None, name: str = NAME):
        return export_bundle(
            self.conn,
            bundles_dir=self.bundles_dir if bundles_dir is None else bundles_dir,
            store_dir=self.store_dir,
            name=name,
            version=version,
        )

    def target(self, version: int = 1) -> Path:
        return self.bundles_dir / NAME / f"v{version}"


@pytest.fixture
def env(tmp_path: Path, catalog_conn: sqlite3.Connection) -> Env:
    """版 1（症例 2・撮影 3・ファイル 3）を凍結した目録と，実体のあるストア．"""
    catalog_conn.execute("PRAGMA synchronous = OFF")
    # パスに架空の患者 ID を含め，メッセージにパスが出ればすぐ分かるようにする
    root = tmp_path / f"患者{LEAK_KEY}"
    store_dir = root / "store"
    bundles_dir = root / "bundles"
    datasets_dir = root / "datasets"
    for d in (store_dir, bundles_dir, datasets_dir):
        d.mkdir(parents=True)
    (datasets_dir / f"{NAME}.toml").write_bytes(TONGUE_COLOR_TOML.encode("utf-8"))
    e = Env(catalog_conn, root, store_dir, bundles_dir, datasets_dir, {})
    e.add(3, "S000001", "red", "mild")
    e.add(4, "S000001", "light_red", "normal")
    e.add(1, "S000002", "pale", "strong")
    result = e.freeze()
    assert result.version == 1
    return e


def _record(conn: sqlite3.Connection, version: int = 1) -> tuple[str, str, str]:
    return conn.execute(
        "SELECT manifest_sha256, created_at, hub_commit FROM dataset_versions"
        " WHERE name = ? AND version = ?",
        (NAME, version),
    ).fetchone()


def _tree(path: Path) -> dict[str, bytes]:
    return {
        p.relative_to(path).as_posix(): p.read_bytes()
        for p in sorted(path.rglob("*"))
        if p.is_file()
    }


def _temp_entries(bundles_dir: Path) -> list[Path]:
    return [p for p in bundles_dir.rglob("*") if p.name.startswith(".tmp-")]


def _assert_safe_message(message: str, env: Env) -> None:
    # 例外メッセージにバンドル・ストアのパス・患者 ID・元のファイル名を出さない
    for form in (str(env.root), env.root.as_posix(), str(env.root.resolve())):
        assert form not in message, message
    assert LEAK_KEY not in message, message
    assert "_original" not in message, message


def _export_error(env: Env, **kwargs) -> str:
    with pytest.raises(ValueError) as exc:
        env.export(**kwargs)
    message = str(exc.value)
    _assert_safe_message(message, env)
    return message


def _make_writable(path: Path) -> None:
    os.chmod(path, stat.S_IWRITE | stat.S_IREAD)


# ---------------------------------------------------------------------------
# 書き出したバンドルの形
# ---------------------------------------------------------------------------
class TestExportedBundle:
    def test_layout(self, env: Env) -> None:
        result = env.export()
        target = env.target()
        assert target.is_dir()
        assert sorted(p.name for p in target.iterdir()) == [
            "SHA256SUMS",
            "dataset.json",
            "files",
            "manifest.csv",
        ]
        assert sorted(p.name for p in (target / "files").iterdir()) == sorted(
            f"{sha}.png" for sha in env.contents
        )
        assert result.reused is False
        assert result.name == NAME
        assert result.version == 1
        assert result.files == 3
        assert result.size_bytes == sum(len(c) for c in env.contents.values())

    def test_files_match_store_contents(self, env: Env) -> None:
        env.export()
        for sha, content in env.contents.items():
            path = env.target() / "files" / f"{sha}.png"
            assert path.read_bytes() == content
            assert hashlib.sha256(path.read_bytes()).hexdigest() == sha

    def test_manifest_matches_version_record(self, env: Env) -> None:
        result = env.export()
        data = (env.target() / "manifest.csv").read_bytes()
        digest, _, _ = _record(env.conn)
        assert hashlib.sha256(data).hexdigest() == digest
        assert result.manifest_sha256 == digest
        assert data == rebuild_manifest(env.conn, NAME, 1)
        assert b"\r" not in data
        assert not data.startswith(b"\xef\xbb\xbf")

    def test_dataset_json_items(self, env: Env) -> None:
        env.export()
        raw = (env.target() / "dataset.json").read_bytes()
        info = json.loads(raw.decode("utf-8"))
        digest, created_at, hub_commit = _record(env.conn)
        assert set(info) == DATASET_JSON_KEYS
        assert info == {
            "format": 1,
            "name": NAME,
            "version": 1,
            "created_at": created_at,  # 版の作成日時（書き出した日時ではない）
            "hub_commit": hub_commit,
            "files": 3,
            "manifest_sha256": digest,
        }
        assert hub_commit == COMMIT
        assert b"\r" not in raw

    def test_sha256sums_format(self, env: Env) -> None:
        env.export()
        raw = (env.target() / "SHA256SUMS").read_bytes()
        assert b"\r" not in raw
        assert raw.endswith(b"\n")
        lines = raw.decode("utf-8").splitlines()
        assert len(lines) == 3
        entries = set()
        for line in lines:
            sha, sep, path = line[:64], line[64:66], line[66:]
            assert sep == "  "  # sha256sum の形（間は空白 2 つ）
            assert path == f"files/{sha}.png"
            entries.add(sha)
        assert entries == set(env.contents)

    def test_no_personal_info_in_bundle_metadata(self, env: Env) -> None:
        env.export()
        for filename in ("dataset.json", "manifest.csv", "SHA256SUMS"):
            text = (env.target() / filename).read_bytes().decode("utf-8")
            assert LEAK_KEY not in text
            assert "_original" not in text
            assert str(env.root) not in text and env.root.as_posix() not in text

    def test_opens_and_verifies_with_reader(self, env: Env) -> None:
        env.export()
        b = open_bundle(env.target())
        assert b.verify() is None
        digest, _, _ = _record(env.conn)
        assert b.fingerprint == {"name": NAME, "version": 1, "manifest_sha256": digest}
        assert [(r.subject_id, r.capture_id) for r in b.rows] == [
            ("S000001", "C000003"),
            ("S000001", "C000004"),
            ("S000002", "C000001"),
        ]
        assert b.rows[1].annotations == {
            "tongue_color": {"label": "light_red", "severity": "normal"}
        }
        for row in b.rows:
            assert b.file_path(row).read_bytes() == env.contents[row.sha256]

    def test_bundle_files_are_not_read_only(self, env: Env) -> None:
        # ストアは読み取り専用だが，バンドルには持ち込まない（ユーザー判断①）
        env.export()
        for path in env.target().rglob("*"):
            if path.is_file():
                assert os.access(path, os.W_OK), path.name
                assert path.stat().st_mode & stat.S_IWRITE
        # 書き出した後に消せる（受け渡し後に消すため）
        shutil.rmtree(env.target())
        assert not env.target().exists()

    def test_no_temp_folder_left_after_success(self, env: Env) -> None:
        env.export()
        assert _temp_entries(env.bundles_dir) == []
        assert sorted(p.name for p in (env.bundles_dir / NAME).iterdir()) == ["v1"]

    def test_store_is_unchanged(self, env: Env) -> None:
        before = _tree(env.store_dir)
        env.export()
        assert _tree(env.store_dir) == before


# ---------------------------------------------------------------------------
# 同じ版から同じバンドル（PLAN_02 Step 5 の完了条件）
# ---------------------------------------------------------------------------
class TestReproducible:
    def test_same_version_gives_same_bytes(self, env: Env, tmp_path: Path) -> None:
        env.export()
        other = tmp_path / "other_bundles"
        other.mkdir()
        env.export(bundles_dir=other)
        assert _tree(env.target()) == _tree(other / NAME / "v1")

    def test_old_version_unchanged_after_new_data_and_correction(
        self, env: Env, tmp_path: Path
    ) -> None:
        env.export()
        first = _tree(env.target())
        # 新しいデータの追加と所見の訂正の後に版 2 を作る
        new_id = env.add(9, "S000003", "purple", "strong")
        old_id = env.conn.execute(
            "SELECT annotation_id FROM annotations WHERE capture_id = 'C000003'"
        ).fetchone()[0]
        cursor = env.conn.execute(
            "INSERT INTO annotations (kind, capture_id, value, source, created_at)"
            " VALUES ('tongue_color', 'C000003', ?, 'dummy_src', '2026-10-02T00:00:00Z')",
            (_tc("dark_red", "strong"),),
        )
        env.conn.execute(
            "UPDATE annotations SET superseded_by = ? WHERE annotation_id = ?",
            (cursor.lastrowid, old_id),
        )
        env.conn.commit()
        assert new_id
        assert env.freeze(datetime(2026, 10, 7, tzinfo=UTC)).version == 2

        other = tmp_path / "other_bundles"
        other.mkdir()
        env.export(1, bundles_dir=other)
        assert _tree(other / NAME / "v1") == first
        assert open_bundle(other / NAME / "v1").rows[0].annotations["tongue_color"]["label"] == (
            "red"
        )

        env.export(2)
        v2 = open_bundle(env.target(2))
        v2.verify()
        assert len(v2.rows) == 4
        assert v2.rows[0].annotations["tongue_color"]["label"] == "dark_red"
        assert (
            json.loads((env.target(2) / "dataset.json").read_bytes())["created_at"]
            != (json.loads(first["dataset.json"])["created_at"])
        )


# ---------------------------------------------------------------------------
# 既にあるバンドル（上書きしない・照合して使う）
# ---------------------------------------------------------------------------
class TestExistingBundle:
    def test_reuses_intact_bundle(self, env: Env) -> None:
        env.export()
        before = _tree(env.target())
        mtimes = {p.name: p.stat().st_mtime_ns for p in env.target().rglob("*") if p.is_file()}
        result = env.export()
        assert result.reused is True
        assert result.files == 3
        assert result.size_bytes == sum(len(c) for c in env.contents.values())
        assert _tree(env.target()) == before
        assert {
            p.name: p.stat().st_mtime_ns for p in env.target().rglob("*") if p.is_file()
        } == mtimes
        assert _temp_entries(env.bundles_dir) == []

    @pytest.mark.parametrize(
        "tamper",
        [
            "modify_file",
            "remove_file",
            "extra_file",
            "modify_manifest",
            "remove_manifest",
            "modify_dataset_json",
            "remove_dataset_json",
            "modify_sha256sums",
            "remove_sha256sums",
            "remove_files_folder",
        ],
    )
    def test_mismatching_bundle_stops_without_overwrite(self, env: Env, tamper: str) -> None:
        env.export()
        target = env.target()
        first_file = sorted((target / "files").iterdir())[0]
        if tamper == "modify_file":
            first_file.write_bytes(b"tampered")
        elif tamper == "remove_file":
            first_file.unlink()
        elif tamper == "extra_file":
            extra = hashlib.sha256(b"extra").hexdigest()
            (target / "files" / f"{extra}.png").write_bytes(b"extra")
        elif tamper == "modify_manifest":
            path = target / "manifest.csv"
            path.write_bytes(path.read_bytes().replace(b"pale", b"red"))
        elif tamper == "modify_dataset_json":
            path = target / "dataset.json"
            info = json.loads(path.read_bytes())
            info["hub_commit"] = "f" * 40
            path.write_bytes(json.dumps(info, indent=2).encode() + b"\n")
        elif tamper == "modify_sha256sums":
            path = target / "SHA256SUMS"
            path.write_bytes(path.read_bytes().replace(b"  ", b" *", 1))
        elif tamper == "remove_files_folder":
            shutil.rmtree(target / "files")
        else:
            filename = {
                "remove_manifest": "manifest.csv",
                "remove_dataset_json": "dataset.json",
                "remove_sha256sums": "SHA256SUMS",
            }[tamper]
            (target / filename).unlink()
        before = _tree(target)

        message = _export_error(env)
        assert "上書き" in message
        assert _tree(target) == before  # 上書き・修復しない
        assert _temp_entries(env.bundles_dir) == []

    def test_existing_empty_folder_stops(self, env: Env) -> None:
        env.target().mkdir(parents=True)
        _export_error(env)
        assert list(env.target().iterdir()) == []

    def test_existing_path_is_file_stops(self, env: Env) -> None:
        env.target().parent.mkdir(parents=True)
        env.target().write_bytes(b"not a bundle")
        _export_error(env)
        assert env.target().read_bytes() == b"not a bundle"

    def test_existing_bundle_of_other_version_stops(self, env: Env) -> None:
        # v1 の場所に別の版のバンドルが置かれている
        env.add(9, "S000003")
        assert env.freeze(datetime(2026, 10, 7, tzinfo=UTC)).version == 2
        env.export(2)
        shutil.copytree(env.target(2), env.target(1))
        before = _tree(env.target(1))
        _export_error(env, version=1)
        assert _tree(env.target(1)) == before


# ---------------------------------------------------------------------------
# 書き出せないもの
# ---------------------------------------------------------------------------
class TestExportStops:
    def test_missing_version(self, env: Env) -> None:
        _export_error(env, version=2)
        assert not (env.bundles_dir / NAME / "v2").exists()

    def test_unknown_dataset(self, env: Env) -> None:
        _export_error(env, name="no_such")
        assert not (env.bundles_dir / "no_such").exists()

    @pytest.mark.parametrize("name", ["", "Bad", "../tongue_color", "a/b", "a\\b", "."])
    def test_invalid_name(self, env: Env, name: str) -> None:
        _export_error(env, name=name)
        assert list(env.bundles_dir.iterdir()) == []

    @pytest.mark.parametrize("version", [0, -1, True])
    def test_invalid_version(self, env: Env, version: int) -> None:
        _export_error(env, version=version)
        assert list(env.bundles_dir.iterdir()) == []

    def test_missing_bundles_dir(self, env: Env, tmp_path: Path) -> None:
        _export_error(env, bundles_dir=env.root / "no_bundles")
        assert not (env.root / "no_bundles").exists()

    def test_missing_store_file_leaves_nothing(self, env: Env) -> None:
        victim = next(iter(env.contents))
        path = store.store_path(env.store_dir, victim, "png")
        _make_writable(path)
        path.unlink()
        message = _export_error(env)
        assert victim in message or victim[:12] in message
        assert not env.target().exists()
        assert _temp_entries(env.bundles_dir) == []

    def test_corrupted_store_file_leaves_nothing(self, env: Env) -> None:
        victim = next(iter(env.contents))
        path = store.store_path(env.store_dir, victim, "png")
        _make_writable(path)
        path.write_bytes(b"corrupted")
        _export_error(env)
        assert not env.target().exists()
        assert _temp_entries(env.bundles_dir) == []

    def test_catalog_changed_after_freeze_stops(self, env: Env) -> None:
        # 版の記録と構成から作り直した manifest が食い違う（目録の書き換え）
        env.conn.execute("UPDATE assets SET kind = 'tongue_raw' WHERE capture_id = 'C000003'")
        env.conn.commit()
        message = _export_error(env)
        assert "check" in message
        assert not env.target().exists()
        assert _temp_entries(env.bundles_dir) == []

    def test_failure_keeps_other_versions(self, env: Env) -> None:
        env.export()
        before = _tree(env.target())
        _export_error(env, version=5)
        assert _tree(env.target()) == before

    def test_catalog_not_modified(self, env: Env) -> None:
        def dump() -> list:
            return [
                env.conn.execute(f"SELECT * FROM {t} ORDER BY 1").fetchall()
                for t in ("dataset_versions", "dataset_members", "assets", "annotations")
            ]

        before = dump()
        env.export()
        env.export()
        assert dump() == before