"""データセットの定義の読み込み・評価と，版の作成（SPEC_03「データセットの定義」「版の作成」）.

定義（`datasets/<名前>.toml`）は個人情報を含まないためリポジトリで管理する（PR で変更する）．
実行時は，このパッケージの位置からリポジトリのルートを辿って `datasets/` を見つける
（`importers.reviews` の要確認リストと同じ．版の作成はリポジトリの clone から実行する）．

版とその構成は凍結される（UPDATE・DELETE しない）．構成（ファイルと採用した所見の ID）と
定義の内容が直前の版と同じなら新しい版を作らない．
"""

import json
import sqlite3
import tomllib
from collections import defaultdict
from collections.abc import Iterable, Mapping, Sequence
from contextlib import closing
from datetime import datetime
from pathlib import Path

from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator

from tiasdata import catalog, kinds, repo
from tiasdata.catalog import versions
from tiasdata.catalog.records import utc_timestamp
from tiasdata.catalog.versions import (
    AnnotationRow,
    MemberAnnotationRow,
    MemberAssetRow,
    VersionRecord,
    VersionSummary,
)
from tiasdata.config import Config
from tiasdata.manifest import (
    DATASET_NAME_PATTERN,
    ManifestEntry,
    build_manifest,
    manifest_sha256,
    validation_error_details,
)

DATASETS_DIRNAME = "datasets"
DEFINITION_SUFFIX = ".toml"
# エラーメッセージに出す例の最大件数
_MAX_EXAMPLES = 5
# エラーメッセージでファイルを示す SHA-256 の桁数（位置の特定には先頭で足りる）
_SHORT_SHA256 = 12


# ---------------------------------------------------------------------------
# 定義
# ---------------------------------------------------------------------------
class _StrictModel(BaseModel):
    # 未知のキーは拒否する（綴りの誤りで条件・列が黙って無視されるのを防ぐ）
    model_config = ConfigDict(extra="forbid", frozen=True)


def _unique(values: list[str], label: str) -> list[str]:
    duplicated = sorted({v for v in values if values.count(v) > 1})
    if duplicated:
        raise ValueError(f"{label}に同じ名前が 2 回以上ある: {', '.join(duplicated)}")
    return values


class SelectSpec(_StrictModel):
    """`[select]`: 集めるファイルの条件．

    Attributes:
        asset_kind: 集めるファイルの種類．
        require: この所見（撮影に付いたものを含む）を持つファイルだけを集める．
    """

    asset_kind: str = Field(min_length=1)
    require: list[str] = []

    @field_validator("require")
    @classmethod
    def _unique_require(cls, value: list[str]) -> list[str]:
        return _unique(value, "require ")


class DatasetDefinition(_StrictModel):
    """データセットの定義（`datasets/<名前>.toml`）．

    Attributes:
        name: データセット名（定義ファイル名と同じ）．
        description: 説明．
        select: 集めるファイルの条件．
        columns: manifest に出す所見の列（所見の種類 → 項目．この順に並べる）．
    """

    name: str = Field(pattern=DATASET_NAME_PATTERN.pattern)
    description: str = Field(min_length=1)
    select: SelectSpec
    columns: dict[str, list[str]] = {}

    @field_validator("columns")
    @classmethod
    def _non_empty_fields(cls, value: dict[str, list[str]]) -> dict[str, list[str]]:
        for kind, fields in value.items():
            if not fields:
                raise ValueError(f"columns.{kind} の項目が空")
            _unique(fields, f"columns.{kind} ")
        return value

    @property
    def annotation_kinds(self) -> list[str]:
        """評価に使う所見の種類（`require` のあとに `columns` の種類．重複なし）．"""
        return list(dict.fromkeys([*self.select.require, *self.columns]))


class LoadedDefinition(BaseModel):
    """読み込んだ定義と，版に記録する定義ファイルの内容．

    Attributes:
        definition: 検証済みの定義．
        text: 定義ファイルの内容（そのまま．版の `definition` に保存する）．
    """

    model_config = ConfigDict(frozen=True)

    definition: DatasetDefinition
    text: str


def _check_name(name: str) -> None:
    # 名前をファイル名に使うため，パスの区切り等を含む名前は受け付けない
    if not DATASET_NAME_PATTERN.match(name):
        raise ValueError(
            f"データセット名 '{name}' は英小文字で始まり，英小文字・数字・_ だけにする"
        )


def parse_definition(text: str, expected_name: str) -> DatasetDefinition:
    """定義ファイルの内容を読んで形を検証する（目録・種類の照合はしない）．

    Args:
        text: 定義ファイルの内容（TOML）．
        expected_name: 定義ファイル名から得たデータセット名（`name` と一致すること）．

    Returns:
        検証済みの定義．

    Raises:
        ValueError: TOML として読めない，形が誤っている（未知のキー・必須キーの欠落・
            項目の重複等），または `name` が定義ファイル名と一致しない．
    """
    filename = f"{DATASETS_DIRNAME}/{expected_name}{DEFINITION_SUFFIX}"
    try:
        data = tomllib.loads(text)
    except tomllib.TOMLDecodeError as e:
        raise ValueError(f"データセットの定義 {filename} を TOML として読めない: {e}") from e
    try:
        definition = DatasetDefinition.model_validate(data)
    except ValidationError as e:
        raise ValueError(
            f"データセットの定義 {filename} の形が誤っている: {validation_error_details(e)}"
        ) from e
    if definition.name != expected_name:
        raise ValueError(
            f"データセットの定義 {filename} の name '{definition.name}' がファイル名と一致しない"
        )
    return definition


def check_definition_kinds(definition: DatasetDefinition, categories: Mapping[str, str]) -> None:
    """定義の種類が目録の種類と所見の値の型に合うかを確かめる．

    Args:
        definition: 定義．
        categories: 目録の種類名 → 分類（`asset`・`annotation`）．

    Raises:
        ValueError: `asset_kind` がファイルの種類でない，`require`・`columns` の種類が所見の
            種類でない（または値の型が無い），`columns` の項目が値の型に無い．
    """
    prefix = f"データセットの定義 {DATASETS_DIRNAME}/{definition.name}{DEFINITION_SUFFIX}"
    asset_kind = definition.select.asset_kind
    if categories.get(asset_kind) != "asset":
        raise ValueError(f"{prefix}: select.asset_kind '{asset_kind}' はファイルの種類にない")
    for kind in definition.annotation_kinds:
        if categories.get(kind) != "annotation":
            raise ValueError(f"{prefix}: '{kind}' は所見の種類にない")
        value_type = kinds.annotation_value_type(kind)
        unknown = [f for f in definition.columns.get(kind, []) if f not in value_type.model_fields]
        if unknown:
            raise ValueError(
                f"{prefix}: columns.{kind} の項目 {', '.join(unknown)} は所見の値に無い"
                f"（あるもの: {', '.join(value_type.model_fields)}）"
            )


def default_datasets_dir() -> Path:
    """リポジトリの `datasets/` フォルダを返す．

    Raises:
        ValueError: リポジトリのルートが見つからない，またはルートに `datasets/` が無い．
    """
    directory = repo.find_repo_root() / DATASETS_DIRNAME
    if not directory.is_dir():
        raise ValueError(
            f"リポジトリに {DATASETS_DIRNAME}/ が無い（データセットの定義を置くフォルダ）"
        )
    return directory


def list_definition_names(datasets_dir: Path) -> list[str]:
    """定義ファイルのあるデータセット名を名前の順に返す（名前に使えないファイル名は除く）．"""
    return sorted(
        path.stem
        for path in datasets_dir.glob(f"*{DEFINITION_SUFFIX}")
        if path.is_file() and DATASET_NAME_PATTERN.match(path.stem)
    )


def load_definition(datasets_dir: Path, name: str) -> LoadedDefinition:
    """定義ファイルを読んで形を検証する．

    Args:
        datasets_dir: 定義のフォルダ．
        name: データセット名．

    Returns:
        検証済みの定義と，定義ファイルの内容．

    Raises:
        ValueError: 名前が使えない，定義ファイルが無い・読めない，または形が誤っている．
    """
    _check_name(name)
    filename = f"{DATASETS_DIRNAME}/{name}{DEFINITION_SUFFIX}"
    path = datasets_dir / f"{name}{DEFINITION_SUFFIX}"
    if not path.is_file():
        raise ValueError(f"データセットの定義 {filename} が無い")
    try:
        # 改行を変換せずにそのまま読む（版に定義の内容をそのまま記録するため）
        text = path.read_bytes().decode("utf-8")
    except UnicodeDecodeError as e:
        raise ValueError(f"データセットの定義 {filename} が UTF-8 でない") from e
    return LoadedDefinition(definition=parse_definition(text, name), text=text)


# ---------------------------------------------------------------------------
# 目録への当てはめ
# ---------------------------------------------------------------------------
# 構成の比較用の値（ファイルの SHA-256 の組, (ファイルの SHA-256, 所見の ID) の組）
type CompositionKey = tuple[tuple[str, ...], tuple[tuple[str, int], ...]]


class Composition(BaseModel):
    """定義を目録に当てはめた結果（版の構成）．

    Attributes:
        candidates: 種類が一致したファイルの数．
        members: 版に入れるファイル（SHA-256 の順）．
        member_annotations: 採用した所見（ファイルの SHA-256・所見の ID の順）．
        unlinked: 条件に合うが，撮影にひも付かないため版に入れなかったファイルの数．
    """

    model_config = ConfigDict(frozen=True)

    candidates: int
    members: list[MemberAssetRow]
    member_annotations: list[MemberAnnotationRow]
    unlinked: int = 0

    @property
    def excluded(self) -> int:
        """必須の所見が無いため版に入れなかったファイルの数．"""
        return self.candidates - len(self.members) - self.unlinked

    def key(self) -> CompositionKey:
        """構成の比較に使う値（ファイルの SHA-256 と，採用した所見の ID）．"""
        return composition_key(
            [m.sha256 for m in self.members], _annotation_ids(self.member_annotations)
        )


def composition_key(
    members: Iterable[str], member_annotations: Iterable[tuple[str, int]]
) -> CompositionKey:
    """構成を順序によらない比較用の値にする．"""
    return tuple(sorted(members)), tuple(sorted(member_annotations))


def _annotation_ids(member_annotations: Iterable[MemberAnnotationRow]) -> list[tuple[str, int]]:
    return [(a.member_sha256, a.annotation.annotation_id) for a in member_annotations]


def _examples(items: Sequence[str]) -> str:
    shown = ", ".join(items[:_MAX_EXAMPLES])
    return shown + (" …" if len(items) > _MAX_EXAMPLES else "")


def _asset_label(asset: MemberAssetRow) -> str:
    # 位置は研究用の撮影 ID か SHA-256 の先頭で示す（元のファイル名は出さない）
    return asset.capture_id if asset.capture_id is not None else asset.sha256[:_SHORT_SHA256]


def evaluate_definition(
    definition: DatasetDefinition,
    assets: Sequence[MemberAssetRow],
    annotations: Mapping[str, Sequence[AnnotationRow]],
) -> Composition:
    """定義を目録の行に当てはめて，版の構成を決める（純関数）．

    ファイルごとに，その撮影に付いた所見とファイルに付いた所見のうち有効なもの（訂正されて
    いないもの）を種類ごとに集める．`require` の種類がすべてあるファイルを版に入れ，
    `require`・`columns` の種類の所見を採用する．条件に合うファイルのうち撮影にひも付かない
    もの（manifest の `subject_id`・`capture_id` を決められないもの）は版に入れず，数を返す．

    Args:
        definition: 定義．
        assets: 種類が `select.asset_kind` のファイル．
        annotations: 所見の種類 → 有効な所見（`definition.annotation_kinds` の種類）．

    Returns:
        版の構成．

    Raises:
        ValueError: 版に入れるファイルに同じ種類の有効な所見が 2 つ以上ある（どちらを使うかを
            黙って決めない．SPEC_03）．
    """
    by_capture: dict[tuple[str, str], list[AnnotationRow]] = defaultdict(list)
    by_file: dict[tuple[str, str], list[AnnotationRow]] = defaultdict(list)
    for kind in definition.annotation_kinds:
        for row in annotations.get(kind, []):
            if row.capture_id is not None:
                by_capture[(kind, row.capture_id)].append(row)
            if row.sha256 is not None:
                by_file[(kind, row.sha256)].append(row)

    members: list[MemberAssetRow] = []
    member_annotations: list[MemberAnnotationRow] = []
    conflicts: list[str] = []
    unlinked = 0
    for asset in sorted(assets, key=lambda a: a.sha256):
        found = {
            kind: [
                *(by_capture.get((kind, asset.capture_id), []) if asset.capture_id else []),
                *by_file.get((kind, asset.sha256), []),
            ]
            for kind in definition.annotation_kinds
        }
        if any(not found[kind] for kind in definition.select.require):
            continue
        if asset.capture_id is None or asset.subject_id is None:
            # 版に入れないので，所見の重複の確認より先に除く（2026-10-08 のユーザー判断．
            # 同じ中身で患者の食い違うファイル等．除いた数は表示する）
            unlinked += 1
            continue
        duplicated = [kind for kind, rows in found.items() if len(rows) > 1]
        if duplicated:
            conflicts.extend(f"{_asset_label(asset)} の {kind}" for kind in duplicated)
            continue
        members.append(asset)
        member_annotations.extend(
            MemberAnnotationRow(member_sha256=asset.sha256, annotation=rows[0])
            for rows in found.values()
            if rows
        )

    if conflicts:
        raise ValueError(
            f"データセット '{definition.name}': 同じ種類の有効な所見が 2 つ以上あるファイルが "
            f"{len(conflicts)} 件ある（例: {_examples(conflicts)}）．"
            "所見を訂正して 1 つにしてから版を作る"
        )
    member_annotations.sort(key=lambda a: (a.member_sha256, a.annotation.annotation_id))
    return Composition(
        candidates=len(assets),
        members=members,
        member_annotations=member_annotations,
        unlinked=unlinked,
    )


def manifest_entries(
    members: Iterable[MemberAssetRow], member_annotations: Iterable[MemberAnnotationRow]
) -> list[ManifestEntry]:
    """版の構成から manifest の行の元を作る（純関数．版の作成と作り直しで同じものを使う）．

    Args:
        members: 版のファイル．
        member_annotations: 採用した所見．

    Returns:
        manifest の行の元．

    Raises:
        ValueError: ファイルが撮影にひも付いていない，1 つのファイルに同じ種類の所見を
            2 つ以上採用している，または所見の値が JSON のオブジェクトでない．
    """
    values: dict[str, dict[str, dict]] = defaultdict(dict)
    for item in member_annotations:
        annotation = item.annotation
        if annotation.kind in values[item.member_sha256]:
            raise ValueError(
                f"ファイル {item.member_sha256[:_SHORT_SHA256]} に所見 {annotation.kind} が"
                " 2 つ以上ある"
            )
        value = json.loads(annotation.value)
        if not isinstance(value, dict):
            raise ValueError(f"所見 {annotation.annotation_id} の値が JSON のオブジェクトでない")
        values[item.member_sha256][annotation.kind] = value

    entries: list[ManifestEntry] = []
    for member in members:
        if member.capture_id is None or member.subject_id is None:
            raise ValueError(f"ファイル {member.sha256[:_SHORT_SHA256]} が撮影にひも付いていない")
        entries.append(
            ManifestEntry(
                sha256=member.sha256,
                ext=member.ext,
                kind=member.kind,
                subject_id=member.subject_id,
                capture_id=member.capture_id,
                annotations=values.get(member.sha256, {}),
            )
        )
    return entries


# ---------------------------------------------------------------------------
# 版の作成
# ---------------------------------------------------------------------------
class FreezeResult(BaseModel):
    """版の作成の結果．

    Attributes:
        name: データセット名．
        version: 作った版（`created` が False なら，構成が同じだった直前の版）．
        created: 新しい版を作ったか．
        candidates: 種類が一致したファイルの数．
        members: 版のファイルの数．
        member_annotations: 採用した所見の数．
        unlinked: 条件に合うが，撮影にひも付かないため除いたファイルの数．
        manifest_sha256: 版の manifest.csv の SHA-256．
    """

    model_config = ConfigDict(frozen=True)

    name: str
    version: int
    created: bool
    candidates: int
    members: int
    member_annotations: int
    unlinked: int
    manifest_sha256: str


def collect_composition(conn: sqlite3.Connection, definition: DatasetDefinition) -> Composition:
    """目録の現在の状態に定義を当てはめる（読み取りのみ）．

    Args:
        conn: 目録への接続．
        definition: 形を検証済みの定義．

    Returns:
        版の構成．

    Raises:
        ValueError: 定義の種類が目録の種類と合わない，または当てはめで問題が見つかった
            （`evaluate_definition` を参照）．
    """
    check_definition_kinds(definition, versions.kind_categories(conn))
    assets = versions.list_assets_of_kind(conn, definition.select.asset_kind)
    annotations = {
        kind: versions.list_valid_annotations(conn, kind) for kind in definition.annotation_kinds
    }
    return evaluate_definition(definition, assets, annotations)


def freeze_version(
    conn: sqlite3.Connection,
    loaded: LoadedDefinition,
    *,
    repo_state: repo.RepoState,
    note: str | None = None,
    now: datetime | None = None,
) -> FreezeResult:
    """定義を目録に当てはめて新しい版を作る（構成と定義が直前の版と同じなら作らない）．

    版の行・構成の行は 1 つのトランザクションで INSERT する（凍結のトリガがあるため，
    manifest の SHA-256 を先に計算してから INSERT する）．

    Args:
        conn: 目録への接続（トランザクションの外）．
        loaded: 読み込んだ定義．
        repo_state: 本リポジトリの git の状態（`repo.read_repo_state()`）．
        note: 版の備考．
        now: 作成日時（省略時は現在）．

    Returns:
        版の作成の結果．

    Raises:
        ValueError: 本リポジトリに未コミットの変更がある，定義の種類が目録と合わない，
            当てはめで問題が見つかった，条件に合うファイルが 0 件，または登録に失敗した．
    """
    definition = loaded.definition
    if repo_state.uncommitted:
        # 定義とコードの版をコミットで記録できなくなるため（SPEC_03「版の作成」）
        raise ValueError(
            f"本リポジトリに未コミットの変更が {repo_state.uncommitted} 件ある．"
            "コミットしてから版を作る"
        )
    with catalog.write_transaction(conn):
        composition = collect_composition(conn, definition)
        if not composition.members:
            raise ValueError(
                f"データセット '{definition.name}': 条件に合うファイルが 0 件"
                f"（種類が一致したファイル {composition.candidates} 件・"
                f"撮影にひも付かないため除いた {composition.unlinked} 件）"
            )
        data = build_manifest(
            manifest_entries(composition.members, composition.member_annotations),
            definition.columns,
        )
        digest = manifest_sha256(data)

        latest = versions.get_latest_version(conn, definition.name)
        if (
            latest is not None
            and latest.definition == loaded.text
            and _stored_composition_key(conn, latest) == composition.key()
        ):
            return _freeze_result(latest, composition, created=False)

        created_at = utc_timestamp(now)
        if versions.get_dataset(conn, definition.name) is None:
            versions.insert_dataset(conn, definition.name, definition.description, created_at)
        record = VersionRecord(
            name=definition.name,
            version=1 if latest is None else latest.version + 1,
            definition=loaded.text,
            hub_commit=repo_state.commit,
            manifest_sha256=digest,
            created_at=created_at,
            note=note or None,
        )
        versions.insert_version(
            conn,
            record,
            [m.sha256 for m in composition.members],
            _annotation_ids(composition.member_annotations),
        )
    return _freeze_result(record, composition, created=True)


def _stored_composition_key(conn: sqlite3.Connection, record: VersionRecord) -> CompositionKey:
    return composition_key(
        [m.sha256 for m in versions.version_members(conn, record.name, record.version)],
        _annotation_ids(versions.version_member_annotations(conn, record.name, record.version)),
    )


def _freeze_result(
    record: VersionRecord, composition: Composition, *, created: bool
) -> FreezeResult:
    return FreezeResult(
        name=record.name,
        version=record.version,
        created=created,
        candidates=composition.candidates,
        members=len(composition.members),
        member_annotations=len(composition.member_annotations),
        unlinked=composition.unlinked,
        manifest_sha256=record.manifest_sha256,
    )


def rebuild_manifest(conn: sqlite3.Connection, name: str, version: int) -> bytes:
    """凍結した版の構成から manifest.csv を作り直す（版を作ったときと同じバイト列になる）．

    版に記録した定義（`[columns]`）と，版の構成で採用した所見（訂正された後も古い行）を使う．
    現在の種類の定義・所見の訂正には左右されない．

    Args:
        conn: 目録への接続．
        name: データセット名．
        version: 版．

    Returns:
        manifest.csv の中身．

    Raises:
        ValueError: 版が無い，記録した定義が読めない，または構成が壊れている．
    """
    record = versions.get_version(conn, name, version)
    if record is None:
        raise ValueError(f"データセット '{name}' の版 {version} が無い")
    definition = parse_definition(record.definition, name)
    entries = manifest_entries(
        versions.version_members(conn, name, version),
        versions.version_member_annotations(conn, name, version),
    )
    return build_manifest(entries, definition.columns)


def count_manifest_mismatches(conn: sqlite3.Connection) -> tuple[int, int]:
    """凍結したすべての版について，構成から作り直した manifest の SHA-256 を記録と照合する．

    点検（`check`）用．作り直せない版（構成が壊れている等）も不一致として数える
    （止めずに全体を数える．error-handling.md の例外）．

    Args:
        conn: 目録への接続．

    Returns:
        （照合した版の数, 一致しなかった版の数）．
    """
    summaries = versions.list_versions(conn)
    mismatched = 0
    for summary in summaries:
        record = summary.record
        try:
            data = rebuild_manifest(conn, record.name, record.version)
        except ValueError:
            mismatched += 1
            continue
        if manifest_sha256(data) != record.manifest_sha256:
            mismatched += 1
    return len(summaries), mismatched


# ---------------------------------------------------------------------------
# CLI から使う入口
# ---------------------------------------------------------------------------
class DatasetOverview(BaseModel):
    """データセットの一覧の 1 件．

    Attributes:
        name: データセット名．
        description: 説明（目録の `datasets` の行．版が無ければ定義ファイルの説明）．
        defined: 定義ファイルがあるか．
        versions: 版（版の順）．
    """

    model_config = ConfigDict(frozen=True)

    name: str
    description: str | None
    defined: bool
    versions: list[VersionSummary]


def _resolve_datasets_dir(datasets_dir: Path | None) -> Path:
    # 既定の場所は呼び出し時に引く（テストで差し替えられるようにモジュールの属性として呼ぶ）
    return datasets_dir if datasets_dir is not None else default_datasets_dir()


def _overview(
    conn: sqlite3.Connection, name: str, datasets_dir: Path, defined: bool
) -> DatasetOverview:
    record = versions.get_dataset(conn, name)
    description = record.description if record is not None else None
    if record is None and defined:
        description = load_definition(datasets_dir, name).definition.description
    return DatasetOverview(
        name=name,
        description=description,
        defined=defined,
        versions=versions.list_versions(conn, name),
    )


def run_list(config: Config, *, datasets_dir: Path | None = None) -> list[DatasetOverview]:
    """目録のデータセットと，定義ファイルだけがある（版の無い）データセットの一覧．

    Args:
        config: 設定．
        datasets_dir: 定義のフォルダ（省略時はリポジトリの `datasets/`）．

    Returns:
        データセットの一覧（名前の順）．

    Raises:
        ValueError: 目録・定義のフォルダが無い，または定義ファイルが誤っている．
    """
    directory = _resolve_datasets_dir(datasets_dir)
    defined = set(list_definition_names(directory))
    with closing(catalog.connect_latest(config.catalog.catalog_path)) as conn:
        names = sorted(defined | {d.name for d in versions.list_datasets(conn)})
        return [_overview(conn, name, directory, name in defined) for name in names]


def run_show(config: Config, name: str, *, datasets_dir: Path | None = None) -> DatasetOverview:
    """1 つのデータセットの版の一覧．

    Args:
        config: 設定．
        name: データセット名．
        datasets_dir: 定義のフォルダ（省略時はリポジトリの `datasets/`）．

    Returns:
        データセットと版．

    Raises:
        ValueError: 名前が使えない，目録にも定義にも無い，または目録・定義のフォルダが無い．
    """
    _check_name(name)
    directory = _resolve_datasets_dir(datasets_dir)
    defined = name in list_definition_names(directory)
    with closing(catalog.connect_latest(config.catalog.catalog_path)) as conn:
        if not defined and versions.get_dataset(conn, name) is None:
            raise ValueError(f"データセット '{name}' は目録にも定義（{DATASETS_DIRNAME}/）にも無い")
        return _overview(conn, name, directory, defined)


def run_freeze(
    config: Config,
    name: str,
    *,
    note: str | None = None,
    datasets_dir: Path | None = None,
    repo_state: repo.RepoState | None = None,
    now: datetime | None = None,
) -> FreezeResult:
    """データセットの定義を読み，目録に当てはめて新しい版を作る．

    Args:
        config: 設定．
        name: データセット名．
        note: 版の備考．
        datasets_dir: 定義のフォルダ（省略時はリポジトリの `datasets/`）．
        repo_state: 本リポジトリの git の状態（省略時は `repo.read_repo_state()` で調べる）．
        now: 作成日時（省略時は現在）．

    Returns:
        版の作成の結果．

    Raises:
        ValueError: 定義が無い・誤っている，git の状態を調べられない，未コミットの変更がある，
            目録が無いか古い，または版を作れない（`freeze_version` を参照）．
    """
    directory = _resolve_datasets_dir(datasets_dir)
    loaded = load_definition(directory, name)
    # git の状態は呼び出し時に引く（テストで差し替えられるようにモジュール経由で呼ぶ）
    state = repo_state if repo_state is not None else repo.read_repo_state()
    with closing(catalog.connect_latest(config.catalog.catalog_path)) as conn:
        return freeze_version(conn, loaded, repo_state=state, note=note, now=now)