"""要確認リスト(`reviews/<取り込み元の名前>.toml`)の読み込みと検証(SPEC_03「要確認リスト」).
要確認リストに載せた番号は,取り込みで要確認の印(所見の種類 `review`)を付けて登録する
(2026-10-08 のユーザー判断.それまでは「保留」として取り込みで飛ばしていた).
要確認リストは個人情報を含まないためリポジトリで管理する(PR で変更する).
実行時は,リポジトリのルート(`repo.find_repo_root()`)の `reviews/` を見つける.
パッケージとして別の場所に入れた場合は見つからないので,取り込みはリポジトリの 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
from tiasdata import repo
REVIEWS_DIRNAME = "reviews"
# 取り込み元の名前をファイル名に使うため,パスの区切り等を含む名前は受け付けない
_SOURCE_NAME_PATTERN = re.compile(r"^[A-Za-z0-9_-]+$")
class ReviewEntry(BaseModel):
"""要確認リストの 1 件(`[[review]]`).
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 ReviewFile(BaseModel):
"""要確認リストのファイル全体."""
model_config = ConfigDict(extra="forbid", frozen=True)
review: list[ReviewEntry] = []
class Reviews(NamedTuple):
"""取り込み元の要確認の番号.
Attributes:
entries: 番号 → 要確認リストの 1 件.
filename: 要確認リストのファイル名(リポジトリの `reviews/` からの相対.表示用).
"""
entries: dict[int, ReviewEntry]
filename: str
@property
def numbers(self) -> frozenset[int]:
"""要確認の番号."""
return frozenset(self.entries)
def default_reviews_dir() -> Path:
"""リポジトリの `reviews/` フォルダを返す.
リポジトリのルートは `repo.find_repo_root()` で見つける.
Returns:
`reviews/` のパス.
Raises:
ValueError: リポジトリのルートが見つからない(パッケージとして別の場所に入れて
実行した等),またはルートに `reviews/` が無い.
"""
try:
root = repo.find_repo_root()
except ValueError as e:
# 取り込みの文脈で対処が分かるように,要確認リストの場所として言い直す
raise ValueError(
"要確認リスト(reviews/)の場所が分からない."
"取り込みは TIASDataHub のリポジトリの clone から実行する(uv run tiasdata import …)"
) from e
reviews_dir = root / REVIEWS_DIRNAME
if not reviews_dir.is_dir():
raise ValueError(f"リポジトリに {REVIEWS_DIRNAME}/ が無い(要確認リストを置くフォルダ)")
return reviews_dir
def load_reviews(reviews_dir: Path, source_name: str) -> Reviews:
"""取り込み元の要確認リストを読んで検証する.
ファイルが無ければ停止する(設定の取り込み元の名前を誤ると,要確認の印が付かないまま
取り込まれるため).要確認の無い取り込み元には空の要確認リスト(`[[review]]` が 0 件,
または空のファイル)を置く.
Args:
reviews_dir: 要確認リストのフォルダ.
source_name: 取り込み元の名前(設定ファイルの `sources.<名前>`).
Returns:
要確認の番号.
Raises:
ValueError: 取り込み元の名前がファイル名に使えない,要確認リストのファイルが無い,
TOML として読めない,
形が誤っている(未知のキー・必須キーの欠落・番号が 1 未満等),
または番号が重複している.
"""
if not _SOURCE_NAME_PATTERN.match(source_name):
raise ValueError(
f"取り込み元の名前 '{source_name}' は英数字・_・- だけにする"
"(要確認リストのファイル名に使う)"
)
filename = f"{REVIEWS_DIRNAME}/{source_name}.toml"
path = reviews_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:
review_file = ReviewFile.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 review_file.review]
duplicated = sorted({n for n in numbers if numbers.count(n) > 1})
if duplicated:
# 要確認リストの番号はファイル名の連番で個人をたどれないため,リストの誤りの特定に表示する
raise ValueError(f"要確認リスト {filename} に同じ番号が 2 回以上ある: {duplicated}")
return Reviews(entries={entry.number: entry for entry in review_file.review}, filename=filename)