AI Implementation feature(1082): Project Scaffold and Docker Packaging (#1)

src/tenbackward/config.py

@@ -0,0 +1,166 @@
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Optional
+
+from dotenv import dotenv_values, load_dotenv
+
+
+REQUIRED_KEYS = (
+    "MASTODON_BASE_URL",
+    "MASTODON_ACCESS_TOKEN",
+    "SITE_URL",
+    "HASHTAGS",
+    "RUN_AT",
+)
+
+OPTIONAL_KEYS = (
+    "MASTODON_VISIBILITY",
+    "THROWBACK_PREFIX",
+    "RETRY_COUNT",
+    "TZ",
+)
+
+DEFAULTS = {
+    "MASTODON_VISIBILITY": "public",
+    "THROWBACK_PREFIX": "Throwback:",
+    "RETRY_COUNT": "3",
+    "TZ": "Europe/Berlin",
+    "RUN_AT": "09:00",
+}
+
+
+class ConfigError(ValueError):
+    """Raised when required configuration is missing or invalid."""
+
+
+@dataclass(frozen=True)
+class Config:
+    mastodon_base_url: str
+    mastodon_access_token: str
+    mastodon_visibility: str
+    site_url: str
+    hashtags: str
+    throwback_prefix: str
+    retry_count: int
+    run_at: str
+    tz: str
+    data_dir: Path = field(default_factory=lambda: Path("/app/data"))
+    extra: dict = field(default_factory=dict)
+
+    @property
+    def cron_minute(self) -> str:
+        return self.run_at.split(":", 1)[0]
+
+    @property
+    def cron_hour(self) -> str:
+        return self.run_at.split(":", 1)[1]
+
+
+def _read_dotenv(dotenv_path: Optional[Path]) -> dict[str, str]:
+    if dotenv_path is None:
+        return {}
+    if not dotenv_path.exists():
+        return {}
+    values = dotenv_values(dotenv_path=str(dotenv_path))
+    return {k: v for k, v in values.items() if v is not None}
+
+
+def _values_from_env() -> dict[str, str]:
+    values: dict[str, str] = {}
+    for key in REQUIRED_KEYS + OPTIONAL_KEYS:
+        raw = os.environ.get(key)
+        if raw is not None and raw != "":
+            values[key] = raw
+    return values
+
+
+def load_config(dotenv_path: Optional[Path] = None) -> Config:
+    """Load configuration from a dotenv file and the process environment.
+
+    Environment variables take precedence over the dotenv file so a mounted
+    `.env` can be supplemented by Compose-level overrides.
+    """
+    if dotenv_path is not None:
+        load_dotenv(dotenv_path=str(dotenv_path), override=False)
+
+    file_values = _read_dotenv(dotenv_path)
+    env_values = _values_from_env()
+
+    merged: dict[str, str] = {}
+    merged.update(file_values)
+    merged.update(env_values)
+
+    apply_defaults(merged)
+
+    validate_config(merged)
+
+    data_dir = Path(os.environ.get("DATA_DIR", "/app/data")).resolve()
+
+    extra = {k: v for k, v in merged.items() if k not in REQUIRED_KEYS + OPTIONAL_KEYS}
+
+    return Config(
+        mastodon_base_url=merged["MASTODON_BASE_URL"],
+        mastodon_access_token=merged["MASTODON_ACCESS_TOKEN"],
+        mastodon_visibility=merged["MASTODON_VISIBILITY"],
+        site_url=merged["SITE_URL"],
+        hashtags=merged["HASHTAGS"],
+        throwback_prefix=merged["THROWBACK_PREFIX"],
+        retry_count=_parse_retry_count(merged["RETRY_COUNT"]),
+        run_at=merged["RUN_AT"],
+        tz=merged["TZ"],
+        data_dir=data_dir,
+        extra=extra,
+    )
+
+
+def apply_defaults(values: dict[str, str]) -> None:
+    for key, default in DEFAULTS.items():
+        values.setdefault(key, default)
+
+
+def validate_config(values: dict[str, str]) -> None:
+    missing = [k for k in REQUIRED_KEYS if not values.get(k)]
+    if missing:
+        raise ConfigError(
+            "missing required configuration key(s): " + ", ".join(missing)
+        )
+
+    run_at = values.get("RUN_AT", "")
+    if not _is_valid_hhmm(run_at):
+        raise ConfigError(
+            f"RUN_AT={run_at!r} must be in HH:MM (24-hour) format"
+        )
+
+    try:
+        _parse_retry_count(values.get("RETRY_COUNT", ""))
+    except ConfigError as exc:
+        raise ConfigError(str(exc)) from exc
+
+
+def _is_valid_hhmm(value: str) -> bool:
+    if not isinstance(value, str):
+        return False
+    parts = value.split(":")
+    if len(parts) != 2:
+        return False
+    hour, minute = parts
+    if len(hour) != 2 or len(minute) != 2:
+        return False
+    if not hour.isdigit() or not minute.isdigit():
+        return False
+    h = int(hour)
+    m = int(minute)
+    return 0 <= h <= 23 and 0 <= m <= 59
+
+
+def _parse_retry_count(value: str) -> int:
+    try:
+        count = int(value)
+    except (TypeError, ValueError):
+        raise ConfigError(f"RETRY_COUNT={value!r} must be a positive integer")
+    if count < 0:
+        raise ConfigError(f"RETRY_COUNT={value!r} must be >= 0")
+    return count