205 lines
7.3 KiB
Python
205 lines
7.3 KiB
Python
"""Reading and writing the settings document.
|
|
|
|
Three jobs:
|
|
|
|
* seed the document from environment variables the first time the app boots, so
|
|
a filled-in `.env` (the same variables jira-helper uses) is a working setup;
|
|
* hand the API a copy with secrets masked, and merge an edited copy back without
|
|
the UI ever having to know a real token;
|
|
* notify the rest of the process when something changed, so the scheduler can
|
|
re-plan its jobs.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
from typing import Any, Callable
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.config import env_settings
|
|
from app.models import SettingsRow
|
|
from app.services import pomodoro
|
|
from app.settings_schema import SECRET_FIELDS, SECRET_MASK, AppSettings
|
|
|
|
_SETTINGS_ID = 1
|
|
_listeners: list[Callable[[AppSettings], None]] = []
|
|
_lock = threading.Lock()
|
|
|
|
|
|
def on_change(listener: Callable[[AppSettings], None]) -> None:
|
|
"""Register a callback fired after every successful settings save."""
|
|
_listeners.append(listener)
|
|
|
|
|
|
def _csv(value: str) -> list[str]:
|
|
return [part.strip() for part in value.split(",") if part.strip()]
|
|
|
|
|
|
def settings_from_env() -> AppSettings:
|
|
"""Build the initial document out of the environment.
|
|
|
|
The variable names are deliberately the ones from `~/code/jira-helper`, so
|
|
an existing `.env` can be copied across untouched.
|
|
"""
|
|
env = env_settings()
|
|
settings = AppSettings(timezone=env.timezone)
|
|
|
|
jira = settings.sources.jira
|
|
if env.jira_base_url and env.jira_email and env.jira_api_token:
|
|
jira.enabled = True
|
|
jira.base_url = env.jira_base_url.rstrip("/")
|
|
jira.email = env.jira_email
|
|
jira.api_token = env.jira_api_token
|
|
jira.extra_jql = env.jira_extra_jql
|
|
jira.start_date_field = env.jira_start_date_field or jira.start_date_field
|
|
|
|
google = settings.sources.google_tasks
|
|
if env.google_client_id and env.google_client_secret and env.google_refresh_token:
|
|
google.enabled = True
|
|
google.client_id = env.google_client_id
|
|
google.client_secret = env.google_client_secret
|
|
google.refresh_token = env.google_refresh_token
|
|
google.lists = _csv(env.google_tasks_lists)
|
|
|
|
todoist = settings.sources.todoist
|
|
if env.todoist_api_token:
|
|
todoist.enabled = True
|
|
todoist.api_token = env.todoist_api_token
|
|
todoist.projects = _csv(env.todoist_projects)
|
|
|
|
laundry = settings.sources.laundry
|
|
if env.laundry_session:
|
|
laundry.enabled = True
|
|
laundry.session_cookie = env.laundry_session
|
|
laundry.base_url = env.laundry_base_url.rstrip("/")
|
|
|
|
telegram = settings.notifications.telegram
|
|
if env.telegram_bot_token and env.telegram_chat_id:
|
|
telegram.enabled = True
|
|
telegram.bot_token = env.telegram_bot_token
|
|
telegram.chat_id = env.telegram_chat_id
|
|
|
|
if env.slack_webhook_url:
|
|
settings.notifications.slack.enabled = True
|
|
settings.notifications.slack.webhook_url = env.slack_webhook_url
|
|
|
|
auth = settings.auth
|
|
if env.auth_password:
|
|
from app.services import auth_service
|
|
|
|
auth.username = env.auth_username or auth.username
|
|
auth.password_hash = auth_service.hash_password(env.auth_password)
|
|
if env.telegram_owner_id:
|
|
auth.telegram_owner_id = env.telegram_owner_id
|
|
auth.telegram_login_enabled = True
|
|
if env.telegram_bot_username:
|
|
auth.telegram_bot_username = env.telegram_bot_username
|
|
|
|
return settings
|
|
|
|
|
|
def load_settings(session: Session) -> AppSettings:
|
|
"""The current settings, creating the row from the environment if absent."""
|
|
row = session.get(SettingsRow, _SETTINGS_ID)
|
|
if row is None:
|
|
settings = settings_from_env()
|
|
row = SettingsRow(id=_SETTINGS_ID, data=settings.model_dump(mode="json"))
|
|
session.add(row)
|
|
session.commit()
|
|
return settings
|
|
try:
|
|
return AppSettings.model_validate(row.data)
|
|
except Exception:
|
|
# A settings document written by an older build can be missing keys or
|
|
# carry ones that no longer exist; defaults fill the gaps rather than
|
|
# taking the whole app down.
|
|
return AppSettings.model_validate(_merge(AppSettings().model_dump(mode="json"), row.data))
|
|
|
|
|
|
def save_settings(session: Session, settings: AppSettings) -> AppSettings:
|
|
row = session.get(SettingsRow, _SETTINGS_ID)
|
|
data = settings.model_dump(mode="json")
|
|
if row is None:
|
|
row = SettingsRow(id=_SETTINGS_ID, data=data)
|
|
session.add(row)
|
|
else:
|
|
row.data = data
|
|
session.commit()
|
|
|
|
with _lock:
|
|
listeners = list(_listeners)
|
|
for listener in listeners:
|
|
try:
|
|
listener(settings)
|
|
except Exception: # pragma: no cover - a bad listener must not block saving
|
|
import logging
|
|
|
|
logging.getLogger(__name__).exception("settings listener failed")
|
|
return settings
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Secret masking
|
|
# --------------------------------------------------------------------------
|
|
def _walk(node: Any, fn: Callable[[str, Any], Any]) -> Any:
|
|
if isinstance(node, dict):
|
|
return {key: fn(key, _walk(value, fn)) for key, value in node.items()}
|
|
if isinstance(node, list):
|
|
return [_walk(item, fn) for item in node]
|
|
return node
|
|
|
|
|
|
def mask_secrets(data: dict[str, Any]) -> dict[str, Any]:
|
|
"""Replace every configured secret that has a value with the mask."""
|
|
|
|
def mask(key: str, value: Any) -> Any:
|
|
if key in SECRET_FIELDS and isinstance(value, str) and value:
|
|
return SECRET_MASK
|
|
return value
|
|
|
|
return _walk(data, mask)
|
|
|
|
|
|
def _merge(base: Any, incoming: Any) -> Any:
|
|
"""Deep-merge `incoming` onto `base`, keeping `base` where a key is missing."""
|
|
if isinstance(base, dict) and isinstance(incoming, dict):
|
|
merged = dict(base)
|
|
for key, value in incoming.items():
|
|
merged[key] = _merge(base.get(key), value) if key in base else value
|
|
return merged
|
|
return incoming
|
|
|
|
|
|
def unmask_secrets(stored: dict[str, Any], incoming: dict[str, Any]) -> dict[str, Any]:
|
|
"""Put the stored secret back wherever the client sent the mask untouched."""
|
|
|
|
def restore(node_stored: Any, node_in: Any) -> Any:
|
|
if isinstance(node_in, dict):
|
|
out: dict[str, Any] = {}
|
|
for key, value in node_in.items():
|
|
old = node_stored.get(key) if isinstance(node_stored, dict) else None
|
|
if key in SECRET_FIELDS and value == SECRET_MASK:
|
|
out[key] = old or ""
|
|
else:
|
|
out[key] = restore(old, value)
|
|
return out
|
|
return node_in
|
|
|
|
return restore(stored, incoming)
|
|
|
|
|
|
def apply_patch(session: Session, patch: dict[str, Any]) -> AppSettings:
|
|
"""Merge a partial settings document (as the UI sends it) and store it."""
|
|
current = load_settings(session).model_dump(mode="json")
|
|
patch = unmask_secrets(current, patch)
|
|
merged = _merge(current, patch)
|
|
updated = save_settings(session, AppSettings.model_validate(merged))
|
|
|
|
# A changed Focus/Short/Long minute (directly, or via a preset) resizes
|
|
# whatever interval is already running or paused — see
|
|
# `pomodoro.rescale_current`. Cheap and a no-op when nothing moved, so it
|
|
# is simplest to always run it rather than detect which keys changed.
|
|
pomodoro.rescale_current(session, updated)
|
|
return updated
|