ppfeiffer
4b9dd45f05
- config.yaml, nina.yaml, scheduler.yaml aus Root → config/ verschoben - conf/ (env.example, nina.yaml, scheduler.yaml, config.yaml) gelöscht - meshbot/config.py, nina.py, scheduler.py: Pfade auf config/ aktualisiert - .gitignore: config/.env und .forgejo_token ergänzt Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
85 lines
2 KiB
Python
85 lines
2 KiB
Python
import asyncio
|
|
import logging
|
|
import os
|
|
|
|
import yaml
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
CONFIG_PATH = os.environ.get("CONFIG_PATH", os.path.join(os.path.dirname(os.path.dirname(__file__)), "config", "config.yaml"))
|
|
ENV_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)), "config", ".env")
|
|
|
|
_config = {}
|
|
_mtime = 0.0
|
|
_callbacks = []
|
|
|
|
|
|
def _load_env():
|
|
if not os.path.exists(ENV_PATH):
|
|
return
|
|
with open(ENV_PATH, "r") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, _, value = line.partition("=")
|
|
os.environ.setdefault(key.strip(), value.strip())
|
|
logger.info("Environment loaded from %s", ENV_PATH)
|
|
|
|
|
|
def _load():
|
|
global _config, _mtime
|
|
with open(CONFIG_PATH, "r") as f:
|
|
_config = yaml.safe_load(f)
|
|
_mtime = os.path.getmtime(CONFIG_PATH)
|
|
logger.info("Config loaded from %s", CONFIG_PATH)
|
|
|
|
|
|
def _reload_if_changed():
|
|
global _mtime
|
|
try:
|
|
current_mtime = os.path.getmtime(CONFIG_PATH)
|
|
if current_mtime != _mtime:
|
|
old_config = _config.copy()
|
|
_load()
|
|
logger.info("Config reloaded (changed)")
|
|
for cb in _callbacks:
|
|
try:
|
|
cb(old_config, _config)
|
|
except Exception:
|
|
logger.exception("Error in config reload callback")
|
|
except Exception:
|
|
logger.exception("Error checking config file")
|
|
|
|
|
|
def on_reload(callback):
|
|
_callbacks.append(callback)
|
|
|
|
|
|
async def watch(interval: float = 2.0):
|
|
while True:
|
|
await asyncio.sleep(interval)
|
|
_reload_if_changed()
|
|
|
|
|
|
def get(key: str, default=None):
|
|
keys = key.split(".")
|
|
val = _config
|
|
for k in keys:
|
|
if isinstance(val, dict):
|
|
val = val.get(k)
|
|
else:
|
|
return default
|
|
if val is None:
|
|
return default
|
|
return val
|
|
|
|
|
|
def env(key: str, default: str = "") -> str:
|
|
return os.environ.get(key, default)
|
|
|
|
|
|
# Load on import
|
|
_load_env()
|
|
_load()
|