utils/json.py

import json
from pathlib import Path
from typing import Any

FILE_CONFIG_PATH = "config/files.json"


def _get_nested(d: dict[str, Any], keys: list[str]) -> Any:
    cur: Any = d
    for k in keys:
        if not isinstance(cur, dict) or k not in cur:
            raise KeyError(k)
        cur = cur[k]
    return cur


def _get_field_from_json(
    path: str | Path, key: str, default: str | None = None, sep: str = "."
) -> Any:
    """
    JSONファイルからフィールドを取得する。
    階層構造がある場合は`sep`で区切って指定する。

    example: `get_file_path_from_config("projects.original_csv")`
    """
    with open(path, "r", encoding="utf-8") as f:
        try:
            config: dict[str, Any] = json.load(f)
        except json.JSONDecodeError as e:
            raise ValueError(f"JSONデコードエラー: {e}")
        except FileNotFoundError:
            raise ValueError(f"ファイルが見つかりません: {path}")

    keys = key.split(sep) if sep in key else [key]

    try:
        value = _get_nested(config, keys)
    except KeyError:
        if default is not None:
            return default
        raise KeyError(f"Key '{key}' not found in {path}")

    return value


def get_file_path_from_config(
    key: str, default: str | None = None, sep: str = "."
) -> str:
    """
    設定ファイルからファイルのパスを取得する。
    階層構造がある場合は`sep`で区切って指定する。

    example: `get_file_path_from_config("projects.original_csv")`
    """
    with open(FILE_CONFIG_PATH, "r", encoding="utf-8") as f:
        config: dict[str, Any] = json.load(f)

    keys = key.split(sep) if sep in key else [key]

    try:
        value = _get_nested(config, keys)
    except KeyError:
        if default is not None:
            return default
        raise KeyError(f"Key '{key}' not found in {FILE_CONFIG_PATH}")

    if isinstance(value, str):
        return value

    if default is not None:
        return default
    raise TypeError(f"Value at '{key}' is not a string: {type(value).__name__}")


def field_getter(path: str | Path):
    def getter(key: str, default: str | None = None, sep: str = ".") -> Any:
        return _get_field_from_json(path, key, default, sep)

    return getter


def _write_atomic(path: Path, data: str) -> None:
    temp = path.with_suffix(path.suffix + ".tmp")
    temp.write_bytes(data)
    temp.replace(path)


def json_dumps(obj: Any, path: str | Path) -> None:
    """
    オブジェクトをJSON形式のファイルに書き出す。
    """
    json_bytes = json.dumps(obj, ensure_ascii=False, indent=2).encode("utf-8")
    _write_atomic(Path(path), json_bytes)