File size: 2,711 Bytes
efb4d78
c04ba23
c5e9ffd
efb4d78
 
 
 
c5e9ffd
 
 
 
 
 
 
 
 
20bc01d
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
c5e9ffd
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
c04ba23
 
20bc01d
 
 
 
 
 
 
c04ba23
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
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)