268 lines
7.9 KiB
Python
268 lines
7.9 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
from urllib.parse import urlparse
|
|
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
|
|
|
from dotenv import dotenv_values, load_dotenv
|
|
|
|
|
|
REQUIRED_KEYS = (
|
|
"MASTODON_BASE_URL",
|
|
"MASTODON_ACCESS_TOKEN",
|
|
"MASTODON_VISIBILITY",
|
|
"SITE_URL",
|
|
"HASHTAGS",
|
|
"THROWBACK_PREFIX",
|
|
"MAX_RETRIES",
|
|
"RUN_AT",
|
|
"TZ",
|
|
)
|
|
|
|
OPTIONAL_KEYS: tuple[str, ...] = ("BLOG_REPO_URL", "BLOG_DIR")
|
|
|
|
DEFAULTS = {
|
|
"MAX_RETRIES": "5",
|
|
"BLOG_REPO_URL": "https://git.chaospott.de/Chaospott/site",
|
|
"BLOG_DIR": "blog",
|
|
}
|
|
|
|
|
|
ALLOWED_VISIBILITY = frozenset({"public", "unlisted"})
|
|
|
|
|
|
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
|
|
max_retries: int
|
|
run_at: str
|
|
tz: str
|
|
data_dir: Path = field(default_factory=lambda: Path("/app/data"))
|
|
blog_repo_url: str = "https://git.chaospott.de/Chaospott/site"
|
|
blog_dir: Path = field(default_factory=lambda: Path("/app/data/blog"))
|
|
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]
|
|
|
|
|
|
DEFAULT_DOTENV_PATH = Path("/app/.env")
|
|
|
|
|
|
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.
|
|
|
|
The .env file is loaded via python-dotenv at startup so the bind-mounted
|
|
runtime secrets are visible to the process. Environment variables take
|
|
precedence over the dotenv file so a mounted `.env` can be supplemented
|
|
by Compose-level overrides.
|
|
|
|
Required keys are validated against the **raw** merged map (dotenv +
|
|
process env) before any defaults are applied. Defaults are only used
|
|
for the retry budget and the optional Jekyll blog source so a missing
|
|
Mastodon/secret/URL/etc. fails fast with a clear error.
|
|
"""
|
|
if dotenv_path is None:
|
|
dotenv_path = DEFAULT_DOTENV_PATH if DEFAULT_DOTENV_PATH.exists() else None
|
|
|
|
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()
|
|
|
|
raw: dict[str, str] = {}
|
|
raw.update(file_values)
|
|
raw.update(env_values)
|
|
|
|
# MAX_RETRIES is the only required key that has a permitted default
|
|
# (5 attempts per the Job). Apply it before the required-key check so a
|
|
# missing value uses the documented default rather than failing fast.
|
|
if not raw.get("MAX_RETRIES"):
|
|
raw["MAX_RETRIES"] = DEFAULTS["MAX_RETRIES"]
|
|
|
|
validate_required(raw)
|
|
|
|
merged = dict(raw)
|
|
apply_defaults(merged)
|
|
|
|
validate_config(merged)
|
|
|
|
data_dir = Path(os.environ.get("DATA_DIR", "/app/data")).resolve()
|
|
|
|
blog_dir_raw = Path(merged["BLOG_DIR"]).expanduser()
|
|
blog_dir = (
|
|
blog_dir_raw
|
|
if blog_dir_raw.is_absolute()
|
|
else (data_dir / blog_dir_raw).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"],
|
|
max_retries=_parse_max_retries(merged["MAX_RETRIES"]),
|
|
run_at=merged["RUN_AT"],
|
|
tz=merged["TZ"],
|
|
data_dir=data_dir,
|
|
blog_repo_url=merged["BLOG_REPO_URL"],
|
|
blog_dir=blog_dir,
|
|
extra=extra,
|
|
)
|
|
|
|
|
|
def apply_defaults(values: dict[str, str]) -> None:
|
|
for key, default in DEFAULTS.items():
|
|
values.setdefault(key, default)
|
|
|
|
|
|
def validate_required(values: dict[str, str]) -> None:
|
|
"""Validate that every required key is present and non-empty in ``values``.
|
|
|
|
Called against the raw (no-default) merged map so a missing key always
|
|
surfaces as a startup error rather than being silently substituted.
|
|
"""
|
|
errors: list[str] = []
|
|
for key in REQUIRED_KEYS:
|
|
raw = values.get(key, "")
|
|
if not raw:
|
|
errors.append(f"missing required configuration key: {key}")
|
|
if errors:
|
|
raise ConfigError("; ".join(errors))
|
|
|
|
|
|
def validate_config(values: dict[str, str]) -> None:
|
|
errors: list[str] = []
|
|
|
|
run_at = values.get("RUN_AT", "")
|
|
if run_at and not _is_valid_hhmm(run_at):
|
|
errors.append(f"RUN_AT={run_at!r} must be in HH:MM (24-hour) format")
|
|
|
|
base_url = values.get("MASTODON_BASE_URL", "")
|
|
if base_url and not _is_valid_url(base_url):
|
|
errors.append(f"MASTODON_BASE_URL={base_url!r} must be a valid http(s) URL")
|
|
|
|
site_url = values.get("SITE_URL", "")
|
|
if site_url and not _is_valid_url(site_url):
|
|
errors.append(f"SITE_URL={site_url!r} must be a valid http(s) URL")
|
|
|
|
visibility = values.get("MASTODON_VISIBILITY", "")
|
|
if visibility and not _is_valid_visibility(visibility):
|
|
errors.append(
|
|
f"MASTODON_VISIBILITY={visibility!r} must be one of: {sorted(ALLOWED_VISIBILITY)}"
|
|
)
|
|
|
|
blog_repo_url = values.get("BLOG_REPO_URL", "")
|
|
if blog_repo_url and not _is_valid_url(blog_repo_url):
|
|
errors.append(
|
|
f"BLOG_REPO_URL={blog_repo_url!r} must be a valid http(s) URL"
|
|
)
|
|
|
|
tz = values.get("TZ", "")
|
|
if tz and not _is_valid_tz(tz):
|
|
errors.append(f"TZ={tz!r} must be a valid IANA timezone")
|
|
|
|
max_retries = values.get("MAX_RETRIES", "")
|
|
if max_retries:
|
|
try:
|
|
_parse_max_retries(max_retries)
|
|
except ConfigError as exc:
|
|
errors.append(str(exc))
|
|
|
|
if errors:
|
|
raise ConfigError("; ".join(errors))
|
|
|
|
|
|
def _is_valid_hhmm(value) -> 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 _is_valid_url(value: str) -> bool:
|
|
if not isinstance(value, str) or not value:
|
|
return False
|
|
try:
|
|
parsed = urlparse(value)
|
|
except (TypeError, ValueError):
|
|
return False
|
|
return parsed.scheme in {"http", "https"} and bool(parsed.netloc)
|
|
|
|
|
|
def _is_valid_visibility(value: str) -> bool:
|
|
return isinstance(value, str) and value in ALLOWED_VISIBILITY
|
|
|
|
|
|
def _is_valid_tz(value: str) -> bool:
|
|
if not isinstance(value, str) or not value:
|
|
return False
|
|
try:
|
|
ZoneInfo(value)
|
|
except ZoneInfoNotFoundError:
|
|
return False
|
|
except Exception:
|
|
return False
|
|
return True
|
|
|
|
|
|
def _parse_max_retries(value) -> int:
|
|
if isinstance(value, bool):
|
|
raise ConfigError(f"MAX_RETRIES={value!r} must be a non-negative integer")
|
|
try:
|
|
count = int(value)
|
|
except (TypeError, ValueError):
|
|
raise ConfigError(f"MAX_RETRIES={value!r} must be a non-negative integer")
|
|
if count < 0:
|
|
raise ConfigError(f"MAX_RETRIES={value!r} must be >= 0")
|
|
return count
|