Add local settings management
Add cellular-first ntfy delivery with WiFi fallback and safe SIM868 HTTP header framing. Co-authored-by: Copilot <[email protected]>
This commit is contained in:
Mathieu Moran
@@ -0,0 +1,205 @@
|
||||
"""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(),
|
||||
"local_video_retention_days": app_video.retention_days(),
|
||||
"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",
|
||||
"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."
|
||||
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_video.RETENTION_SETTING: str(
|
||||
values["local_video_retention_days"]
|
||||
),
|
||||
"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 {}
|
||||
Reference in New Issue
Block a user