-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathbot_config.py
75 lines (59 loc) · 2.15 KB
/
bot_config.py
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
# Copyright (c) NiceBots.xyz
# SPDX-License-Identifier: MIT
import contextlib
import os
from collections import defaultdict
from typing import Any
import orjson
import yaml
from dotenv import load_dotenv
load_dotenv()
SPLIT: str = "__"
def load_from_env() -> dict[str, dict[str, Any]]:
_config: dict[str, Any] = {}
values = {k: v for k, v in os.environ.items() if k.startswith("BOTKIT__")}
for key, value in values.items():
parts = key[len("BOTKIT__") :].lower().split("__")
current = _config
for i, part in enumerate(parts):
if i == len(parts) - 1:
current[part] = value
else:
if part not in current:
current[part] = {}
elif not isinstance(current[part], dict):
raise ValueError(f"Key {key} in environment must be a leaf")
current = current[part]
return load_json_recursive(_config)
def load_json_recursive(data: dict[str, Any]) -> dict[str, Any]:
for key, value in data.items():
if isinstance(value, dict):
data[key] = load_json_recursive(value)
elif isinstance(value, str):
if value.lower() == "true":
data[key] = True
elif value.lower() == "false":
data[key] = False
elif value.startswith("0x"):
with contextlib.suppress(ValueError):
data[key] = int(value, 16)
else:
with contextlib.suppress(orjson.JSONDecodeError):
data[key] = orjson.loads(value)
return data
path = None
if os.path.exists("config.yaml"):
path = "config.yaml"
elif os.path.exists("config.yml"):
path = "config.yml"
def merge_dicts(dct: dict[str, Any], merge_dct: dict[str, Any]) -> None:
for k, v in merge_dct.items():
if isinstance(dct.get(k), dict) and isinstance(v, dict):
merge_dicts(dct[k], v)
else:
dct[k] = v
config: dict[str, dict[str, Any]] = defaultdict(dict)
if path:
with open(path, encoding="utf-8") as f:
config.update(yaml.safe_load(f))
merge_dicts(config, load_from_env())