"""保留リスト(`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)