"""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." 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 {}