yaris-obd/modules/app_settings.py

"""Validated configuration exposed by the local Settings page."""
import os
import re
from urllib.parse import urlparse

from modules import app_alerts, app_sql, app_upload, app_video


_APN_PATTERN = re.compile(r"^[A-Za-z0-9.-]*$")
_NUMBER_FIELDS = {
    "temperature_c": ("alert_temperature_c", float, 40, 100),
    "disk_percent": ("alert_disk_percent", float, 50, 100),
    "tracking_backlog": ("alert_tracking_backlog", int, 1, 1_000_000),
    "video_backlog": ("alert_video_backlog", int, 1, 100_000),
    "gps_loss_seconds": ("alert_gps_loss_seconds", int, 10, 3_600),
    "repeat_seconds": ("alert_repeat_seconds", int, 60, 604_800),
}


def _env_bool(name):
    return os.getenv(name, "") == "1"


def _setting_bool(key, env_name=None, default=False):
    fallback = "1" if (_env_bool(env_name) if env_name else default) else "0"
    return app_sql.get_setting(key, fallback) == "1"


def _number_value(setting_key, converter):
    default = app_alerts.SETTING_DEFAULTS[setting_key]
    try:
        return converter(app_sql.get_setting(setting_key, default))
    except (TypeError, ValueError):
        return converter(default)


def get_settings():
    return {
        "dashcam_upload_enabled": app_upload.is_dashcam_upload_enabled(),
        "trips_upload_enabled": app_upload.is_trips_upload_enabled(),
        "local_video_retention_days": app_video.retention_days(),
        "video_convert_enabled": app_video.is_convert_enabled(),
        "cellular_enabled": _setting_bool(
            "cellular_enabled", "YARIS_CELLULAR_ENABLED"
        ),
        "cellular_apn": app_sql.get_setting(
            "cellular_apn", os.getenv("YARIS_CELLULAR_APN", "")
        ),
        "ntfy_enabled": _setting_bool("alerts_enabled"),
        "ntfy_url_configured": bool(
            app_sql.get_setting("alert_ntfy_url", "")
        ),
        "ntfy_token_configured": bool(
            app_sql.get_setting("alert_ntfy_token", "")
        ),
        "alert_thresholds": {
            field: _number_value(setting_key, converter)
            for field, (
                setting_key,
                converter,
                _minimum,
                _maximum,
            ) in _NUMBER_FIELDS.items()
        },
    }


def _boolean(body, field, errors):
    value = body.get(field)
    if not isinstance(value, bool):
        errors[field] = "Must be true or false."
        return None
    return value


def _bounded_number(body, field, converter, minimum, maximum, errors):
    value = body.get(field)
    if (
        isinstance(value, bool)
        or (converter is int and isinstance(value, float) and not value.is_integer())
    ):
        errors[field] = f"Must be between {minimum} and {maximum}."
        return None
    try:
        converted = converter(value)
    except (TypeError, ValueError):
        errors[field] = f"Must be between {minimum} and {maximum}."
        return None
    if converted < minimum or converted > maximum:
        errors[field] = f"Must be between {minimum} and {maximum}."
        return None
    return converted


def update_settings(body):
    if not isinstance(body, dict):
        return {"request": "Expected a JSON object."}

    errors = {}
    values = {}
    for field in (
        "dashcam_upload_enabled",
        "trips_upload_enabled",
        "video_convert_enabled",
        "cellular_enabled",
        "ntfy_enabled",
        "clear_ntfy_url",
        "clear_ntfy_token",
    ):
        values[field] = _boolean(body, field, errors)

    values["local_video_retention_days"] = _bounded_number(
        body,
        "local_video_retention_days",
        int,
        0,
        3_650,
        errors,
    )

    apn = body.get("cellular_apn")
    if not isinstance(apn, str):
        errors["cellular_apn"] = "Must be text."
        apn = ""
    else:
        apn = apn.strip()
        if len(apn) > 100 or not _APN_PATTERN.fullmatch(apn):
            errors["cellular_apn"] = (
                "Use at most 100 letters, numbers, dots, or hyphens."
            )
    values["cellular_apn"] = apn
    if values["cellular_enabled"] and not apn:
        errors["cellular_apn"] = "An APN is required when cellular is enabled."

    ntfy_url = body.get("ntfy_url")
    ntfy_token = body.get("ntfy_token")
    if not isinstance(ntfy_url, str):
        errors["ntfy_url"] = "Must be text."
    if not isinstance(ntfy_token, str):
        errors["ntfy_token"] = "Must be text."
    if isinstance(ntfy_url, str):
        ntfy_url = ntfy_url.strip()
        if len(ntfy_url) > 2_048:
            errors["ntfy_url"] = "Must be at most 2048 characters."
        elif ntfy_url:
            parsed = urlparse(ntfy_url)
            if parsed.scheme not in {"http", "https"} or not parsed.netloc:
                errors["ntfy_url"] = "Must be a valid http(s) URL."
    if isinstance(ntfy_token, str) and len(ntfy_token) > 512:
        errors["ntfy_token"] = "Must be at most 512 characters."

    existing_url = app_sql.get_setting("alert_ntfy_url", "")
    effective_url = existing_url
    if isinstance(ntfy_url, str) and ntfy_url:
        effective_url = ntfy_url
    if values["clear_ntfy_url"]:
        effective_url = ""
    if values["ntfy_enabled"] and not effective_url:
        errors["ntfy_url"] = (
            "A ntfy URL is required when notifications are enabled."
        )

    thresholds = body.get("alert_thresholds")
    if not isinstance(thresholds, dict):
        errors["alert_thresholds"] = "Must be an object."
    else:
        for field, (
            setting_key,
            converter,
            minimum,
            maximum,
        ) in _NUMBER_FIELDS.items():
            value = _bounded_number(
                thresholds,
                field,
                converter,
                minimum,
                maximum,
                errors,
            )
            values[setting_key] = value

    if errors:
        return errors

    settings = {
        app_upload.DASHCAM_UPLOAD_SETTING_KEY: (
            "1" if values["dashcam_upload_enabled"] else "0"
        ),
        app_upload.TRIPS_UPLOAD_SETTING_KEY: (
            "1" if values["trips_upload_enabled"] else "0"
        ),
        app_video.RETENTION_SETTING: str(values["local_video_retention_days"]),
        app_video.CONVERT_SETTING: "1" if values["video_convert_enabled"] else "0",
        "cellular_enabled": "1" if values["cellular_enabled"] else "0",
        "cellular_apn": apn,
        "alerts_enabled": "1" if values["ntfy_enabled"] else "0",
    }
    for setting_key, _converter, _minimum, _maximum in _NUMBER_FIELDS.values():
        settings[setting_key] = str(values[setting_key])

    if values["clear_ntfy_url"]:
        settings["alert_ntfy_url"] = ""
    elif ntfy_url:
        settings["alert_ntfy_url"] = ntfy_url
    if values["clear_ntfy_token"]:
        settings["alert_ntfy_token"] = ""
    elif ntfy_token:
        settings["alert_ntfy_token"] = ntfy_token

    app_sql.set_settings(settings)
    return {}