Newer
Older
TIASDataHub / src / tiasdata / importers / holds.py
"""保留リスト(`holds/<取り込み元の名前>.toml`)の読み込みと検証(SPEC_03「保留リスト」).

保留リストは個人情報を含まないためリポジトリで管理する(PR で変更する).
実行時は,このパッケージの位置からリポジトリのルート(`name = "tiasdata"` の `pyproject.toml`
のあるフォルダ)を辿って `holds/` を見つける.パッケージとして別の場所に入れた場合は
見つからないので,取り込みはリポジトリの clone から実行する.
"""

import re
import tomllib
from datetime import date
from pathlib import Path
from typing import NamedTuple

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

HOLDS_DIRNAME = "holds"
_PROJECT_NAME = "tiasdata"
# 取り込み元の名前をファイル名に使うため,パスの区切り等を含む名前は受け付けない
_SOURCE_NAME_PATTERN = re.compile(r"^[A-Za-z0-9_-]+$")


class HoldEntry(BaseModel):
    """保留リストの 1 件(`[[hold]]`).

    Attributes:
        number: 取り込み元での番号.
        reason: 保留にした理由.
        resolve: 確かめる方法.
        since: 保留にした日.
    """

    model_config = ConfigDict(extra="forbid", frozen=True)

    number: StrictInt = Field(ge=1)
    reason: str = Field(min_length=1)
    resolve: str = Field(min_length=1)
    since: date


class HoldFile(BaseModel):
    """保留リストのファイル全体."""

    model_config = ConfigDict(extra="forbid", frozen=True)

    hold: list[HoldEntry] = []


class Holds(NamedTuple):
    """取り込み元の保留の番号.

    Attributes:
        numbers: 保留にする番号.
        filename: 保留リストのファイル名(リポジトリの `holds/` からの相対.表示用).
    """

    numbers: frozenset[int]
    filename: str


def default_holds_dir() -> Path:
    """リポジトリの `holds/` フォルダを返す.

    このパッケージの位置から親のフォルダを辿り,`[project] name = "tiasdata"` の
    `pyproject.toml` があるフォルダをリポジトリのルートとみなす.

    Returns:
        `holds/` のパス.

    Raises:
        ValueError: リポジトリのルートが見つからない(パッケージとして別の場所に入れて
            実行した等),またはルートに `holds/` が無い.
    """
    package_dir = Path(__file__).resolve().parents[1]
    for parent in package_dir.parents:
        pyproject = parent / "pyproject.toml"
        if pyproject.is_file() and _is_tiasdata_project(pyproject):
            holds_dir = parent / HOLDS_DIRNAME
            if not holds_dir.is_dir():
                raise ValueError(
                    f"リポジトリに {HOLDS_DIRNAME}/ が無い(保留リストを置くフォルダ)"
                )
            return holds_dir
    raise ValueError(
        "保留リスト(holds/)の場所が分からない."
        "取り込みは TIASDataHub のリポジトリの clone から実行する(uv run tiasdata import …)"
    )


def _is_tiasdata_project(pyproject: Path) -> bool:
    try:
        data = tomllib.loads(pyproject.read_text(encoding="utf-8"))
    except (OSError, UnicodeDecodeError, tomllib.TOMLDecodeError):
        return False
    project = data.get("project")
    return isinstance(project, dict) and project.get("name") == _PROJECT_NAME


def load_holds(holds_dir: Path, source_name: str) -> Holds:
    """取り込み元の保留リストを読んで検証する.

    ファイルが無ければ停止する(設定の取り込み元の名前を誤ると,保留が効かないまま取り込まれる
    ため).保留の無い取り込み元には空の保留リスト(`[[hold]]` が 0 件,または空のファイル)を置く.

    Args:
        holds_dir: 保留リストのフォルダ.
        source_name: 取り込み元の名前(設定ファイルの `sources.<名前>`).

    Returns:
        保留の番号.

    Raises:
        ValueError: 取り込み元の名前がファイル名に使えない,保留リストのファイルが無い,
            TOML として読めない,
            形が誤っている(未知のキー・必須キーの欠落・番号が 1 未満等),
            または番号が重複している.
    """
    if not _SOURCE_NAME_PATTERN.match(source_name):
        raise ValueError(
            f"取り込み元の名前 '{source_name}' は英数字・_・- だけにする"
            "(保留リストのファイル名に使う)"
        )
    filename = f"{HOLDS_DIRNAME}/{source_name}.toml"
    path = holds_dir / f"{source_name}.toml"
    if not path.is_file():
        raise ValueError(
            f"保留リスト {filename} が無い.保留が無い取り込み元は空の保留リストを置く"
        )
    try:
        data = tomllib.loads(path.read_text(encoding="utf-8"))
    except (UnicodeDecodeError, tomllib.TOMLDecodeError) as e:
        raise ValueError(f"保留リスト {filename} を TOML として読めない: {e}") from e
    try:
        hold_file = HoldFile.model_validate(data)
    except ValidationError as e:
        details = "; ".join(
            f"{'.'.join(str(part) for part in error['loc']) or '(全体)'}: {error['msg']}"
            for error in e.errors(include_input=False, include_url=False)
        )
        raise ValueError(f"保留リスト {filename} の形が誤っている: {details}") from e

    numbers = [entry.number for entry in hold_file.hold]
    duplicated = sorted({n for n in numbers if numbers.count(n) > 1})
    if duplicated:
        # 保留リストの番号はファイル名の連番で個人をたどれないため,リストの誤りの特定に表示する
        raise ValueError(f"保留リスト {filename} に同じ番号が 2 回以上ある: {duplicated}")
    return Holds(numbers=frozenset(numbers), filename=filename)