mirror of
https://github.com/agessaman/meshcore-bot.git
synced 2026-10-31 18:12:45 +00:00
agessaman
f5f3b6e52a
Fixes from a Codex review of the previous commit, plus the shared settings the rebase onto dev showed were still missing.
Settings that misstated or changed what runs:
- sports never read [Sports_Command] api_timeout (documented in the example config); the new field did nothing. It now sets the ESPN timeout.
- The path-byte requirement help on test, multitest and path said "bytes per hop"; the gate checks the path's total size (at least 2, or exactly 3).
- feed's allow_private_urls showed false while it inherited true from [Feed_Manager]. Fields can now name an inherit_section whose value they show when unset.
- notify_on_prefix_bytes was an int field for a value the service reads as a list (2,3), so the form refused a valid setting.
- The greeter's form defaults differed from the fallbacks it uses when the keys are missing. Both now come from shared constants.
- The MQTT weather limits accepted values the service clamps (payload 256 to 1 MiB, staleness 5 s to 7 days, length 64 to 4000).
Shared settings: the wx card now carries the [Weather] temperature, wind and precipitation units and the three temperature_*_format templates (with previews through format_temperature_high_low), next to wx's own [Wx_Command] unit overrides, which default to "Same as [Weather]". Two fields with one key in different sections are told apart as key@Section in the save payload; a bare key still resolves when only one field has it. An enum's "" option, like a blank number, removes the key. Packet capture brokers gain username, password, topic prefix, JWT timings and include_decoded; blank inheriting values are left out of block saves, since the service reads them with getint/getboolean when present.
Previews:
- They use the bot's saved config: a blank test format falls back to [Keywords] test, multitest lays out paths per condense_paths, and MQTT applies passthrough_max_length.
- A quoted empty template ("") counts as blank, as the commands strip the quotes first.
- A str.format width or precision over 500 isn't rendered, so `{sender:100000000}` can't make the endpoint build a huge string.
- The page invalidates an in-flight preview as soon as the field changes, and shows the server's note about fallbacks instead of guessing it.
The endpoint inventory golden gains the preview route (401 for anonymous requests when a password is set).
668 lines
27 KiB
Python
668 lines
27 KiB
Python
"""Machine-readable plugin settings schema, validation, and view assembly.
|
|
|
|
Plugins (commands and services) may declare an optional ``settings_schema``
|
|
class attribute describing their configurable settings. The web viewer uses
|
|
this to render typed widgets (dropdowns, validated numbers, toggles) and the
|
|
save endpoint uses :func:`validate_field` so server-side validation mirrors the
|
|
schema exactly. Plugins without a schema fall back to a generic enable-toggle
|
|
plus a raw key/value editor.
|
|
|
|
This module performs *class-level* discovery: it imports plugin modules and
|
|
reads class attributes WITHOUT instantiating the plugin. That avoids plugin
|
|
``__init__`` side effects (e.g. opening HTTP sessions) and sidesteps the service
|
|
loader's enable-gate, so disabled services still appear in the UI.
|
|
|
|
It has no Flask or bot-core dependencies; pass a ``configparser`` object in.
|
|
|
|
Schema field format (a list of these dicts on ``settings_schema``)::
|
|
|
|
{
|
|
"key": "poll_interval", # config key within the section
|
|
"label": "Poll interval", # human label
|
|
"type": "int", # bool|int|float|str|enum|list|password
|
|
"options": [{"value": "...", "label": "..."}], # required for enum
|
|
"min": 1000, "max": None, # numeric bounds (int/float)
|
|
"default": 60000,
|
|
"help": "Polling cadence in ms.",
|
|
"required": False,
|
|
"pattern": None, # validation regex (str/list)
|
|
"unit": "ms", # optional display suffix
|
|
"section": None, # read/write a shared section instead
|
|
"inherit_section": None, # show this section's value when unset
|
|
}
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import configparser
|
|
import importlib
|
|
import inspect
|
|
import os
|
|
import re
|
|
from typing import Any, Optional
|
|
|
|
# Legacy [section] enabled aliases (shared with BaseCommand.get_config_value)
|
|
# so a plugin's on/off state displays correctly before the first canonical save.
|
|
from modules.config_schema import LEGACY_ENABLED_ALIASES as _ENABLED_LEGACY_ALIASES
|
|
|
|
VALID_TYPES = {"bool", "int", "float", "str", "enum", "list", "password"}
|
|
|
|
# Truthy/falsey string forms accepted for bool fields (configparser-compatible).
|
|
_TRUE = {"1", "true", "yes", "on"}
|
|
_FALSE = {"0", "false", "no", "off"}
|
|
|
|
# Command name -> section base for camelCase names (mirrors
|
|
# BaseCommand._derive_config_section_name).
|
|
_CAMEL_CASE_SECTION = {"dadjoke": "DadJoke", "webviewer": "WebViewer"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Validation / coercion
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _coerce_bool(raw: Any) -> Optional[bool]:
|
|
if isinstance(raw, bool):
|
|
return raw
|
|
s = str(raw).strip().lower()
|
|
if s in _TRUE:
|
|
return True
|
|
if s in _FALSE:
|
|
return False
|
|
return None
|
|
|
|
|
|
def validate_field(field: dict, raw: Any) -> tuple[bool, Any, Optional[str]]:
|
|
"""Validate and coerce a single raw value against a schema field.
|
|
|
|
Returns ``(ok, coerced_value, error_message)``. On failure ``ok`` is False
|
|
and ``error_message`` explains why; ``coerced_value`` is None.
|
|
"""
|
|
ftype = field.get("type", "str")
|
|
label = field.get("label") or field.get("key", "value")
|
|
required = bool(field.get("required", False))
|
|
|
|
# Empty handling (treat None / "" as empty for str/enum/list).
|
|
is_empty = raw is None or (isinstance(raw, str) and raw.strip() == "")
|
|
if is_empty and ftype not in ("bool",):
|
|
if required:
|
|
return False, None, f"{label} is required"
|
|
# Empty + optional: fall back to default if provided, else empty string.
|
|
return True, field.get("default", ""), None
|
|
|
|
if ftype == "bool":
|
|
coerced = _coerce_bool(raw)
|
|
if coerced is None:
|
|
return False, None, f"{label} must be true or false"
|
|
return True, coerced, None
|
|
|
|
if ftype in ("int", "float"):
|
|
try:
|
|
num = int(raw) if ftype == "int" else float(raw)
|
|
except (ValueError, TypeError):
|
|
return False, None, f"{label} must be a number"
|
|
lo, hi = field.get("min"), field.get("max")
|
|
if lo is not None and num < lo:
|
|
return False, None, f"{label} must be ≥ {lo}"
|
|
if hi is not None and num > hi:
|
|
return False, None, f"{label} must be ≤ {hi}"
|
|
return True, num, None
|
|
|
|
if ftype == "enum":
|
|
allowed = {str(o.get("value")) for o in field.get("options", [])}
|
|
s = str(raw)
|
|
if s not in allowed:
|
|
return False, None, f"{label} must be one of: {', '.join(sorted(allowed))}"
|
|
return True, s, None
|
|
|
|
if ftype == "list":
|
|
items = [item.strip() for item in str(raw).split(",") if item.strip()]
|
|
pattern = field.get("pattern")
|
|
if pattern:
|
|
rx = re.compile(pattern)
|
|
for item in items:
|
|
if not rx.fullmatch(item):
|
|
return False, None, f"{label} contains an invalid value: {item}"
|
|
return True, items, None
|
|
|
|
# str / password (password is a str stored in plaintext in config.ini,
|
|
# masked only in the web UI — same validation as str)
|
|
s = str(raw)
|
|
pattern = field.get("pattern")
|
|
if pattern and not re.fullmatch(pattern, s):
|
|
return False, None, f"{label} has an invalid format"
|
|
return True, s, None
|
|
|
|
|
|
def field_id(field: dict) -> str:
|
|
"""A field's identity within its plugin: ``key``, or ``key@Section`` for a shared section.
|
|
|
|
A plugin may expose the same key in two sections (``wx``'s shared ``[Weather]``
|
|
units and its own ``[Wx_Command]`` overrides), so the key alone can't name the
|
|
field in a save payload.
|
|
"""
|
|
section = field.get("section")
|
|
return f"{field['key']}@{section}" if section else field["key"]
|
|
|
|
|
|
def to_config_string(field: dict, coerced: Any) -> str:
|
|
"""Convert a coerced value into its config.ini string form."""
|
|
ftype = field.get("type", "str")
|
|
if ftype == "bool":
|
|
return "true" if coerced else "false"
|
|
if ftype == "list":
|
|
if isinstance(coerced, (list, tuple)):
|
|
return ", ".join(str(x) for x in coerced)
|
|
return str(coerced)
|
|
return str(coerced)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Section / value resolution
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def command_section_name(name: str) -> str:
|
|
"""Derive a command's config section from its name (mirrors base_command)."""
|
|
base = _CAMEL_CASE_SECTION.get(name, name.title())
|
|
return f"{base}_Command"
|
|
|
|
|
|
def service_section_name(service_class: type) -> str:
|
|
"""Derive a service's config section from its class attr / class name."""
|
|
explicit = getattr(service_class, "config_section", None)
|
|
if explicit:
|
|
return explicit
|
|
cls = service_class.__name__
|
|
return cls[:-7] if cls.endswith("Service") else cls
|
|
|
|
|
|
def _read_typed(config: configparser.ConfigParser, section: str, field: dict) -> Any:
|
|
"""Read a field's current value from config, typed, falling back to default.
|
|
|
|
A field may override ``section`` to read from a shared section (e.g. a
|
|
weather command exposing ``[Weather] default_state``).
|
|
"""
|
|
key = field["key"]
|
|
section = field.get("section") or section
|
|
ftype = field.get("type", "str")
|
|
default = field.get("default")
|
|
if not config.has_section(section) or not config.has_option(section, key):
|
|
# A field the plugin inherits from another section when its own is unset
|
|
# shows the inherited value, so the form doesn't misstate what runs.
|
|
inherit = field.get("inherit_section")
|
|
if inherit and config.has_section(inherit) and config.has_option(inherit, key):
|
|
return _read_typed(config, inherit, {**field, "inherit_section": None, "section": None})
|
|
return default
|
|
# raw=True so values containing '%' (cron/strftime/templates) don't trip
|
|
# configparser's interpolation, which would raise InterpolationError.
|
|
try:
|
|
if ftype == "bool":
|
|
return config.getboolean(section, key, raw=True)
|
|
if ftype == "int":
|
|
return config.getint(section, key, raw=True)
|
|
if ftype == "float":
|
|
return config.getfloat(section, key, raw=True)
|
|
if ftype == "list":
|
|
raw = config.get(section, key, raw=True)
|
|
return [item.strip() for item in raw.split(",") if item.strip()]
|
|
return config.get(section, key, raw=True)
|
|
except (ValueError, TypeError):
|
|
return default
|
|
|
|
|
|
def read_enabled(
|
|
config: configparser.ConfigParser, section: str, default: bool
|
|
) -> bool:
|
|
"""Read a plugin's on/off state: canonical key first, then legacy aliases."""
|
|
if config.has_section(section) and config.has_option(section, "enabled"):
|
|
try:
|
|
return config.getboolean(section, "enabled", raw=True)
|
|
except ValueError:
|
|
pass
|
|
for legacy_section, legacy_key in _ENABLED_LEGACY_ALIASES.get(section, []):
|
|
if config.has_section(legacy_section) and config.has_option(legacy_section, legacy_key):
|
|
try:
|
|
return config.getboolean(legacy_section, legacy_key, raw=True)
|
|
except ValueError:
|
|
continue
|
|
return default
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Class-level discovery
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _discover_classes(base_class: type, package: str, directory: str, logger=None) -> list[type]:
|
|
"""Import every ``*.py`` in ``directory`` and collect ``base_class`` subclasses."""
|
|
found: list[type] = []
|
|
if not os.path.isdir(directory):
|
|
return found
|
|
excluded = {"__init__", "base_command", "base_service"}
|
|
for fname in sorted(os.listdir(directory)):
|
|
if not fname.endswith(".py"):
|
|
continue
|
|
stem = fname[:-3]
|
|
if stem in excluded or stem.endswith("_utils"):
|
|
continue
|
|
module_path = f"{package}.{stem}"
|
|
try:
|
|
module = importlib.import_module(module_path)
|
|
except Exception as exc: # noqa: BLE001 - never let one bad plugin break the list
|
|
if logger:
|
|
logger.warning("Could not import %s for settings discovery: %s", module_path, exc)
|
|
continue
|
|
for _n, obj in inspect.getmembers(module, inspect.isclass):
|
|
if (
|
|
issubclass(obj, base_class)
|
|
and obj is not base_class
|
|
and obj.__module__ == module_path
|
|
):
|
|
found.append(obj)
|
|
break
|
|
return found
|
|
|
|
|
|
def _discover_local_classes(
|
|
base_class: type,
|
|
directory: str,
|
|
*,
|
|
package_name: str,
|
|
package_paths: Optional[list[str]] = None,
|
|
local_base_module: Optional[str] = None,
|
|
logger=None,
|
|
) -> list[type]:
|
|
"""Import local plugin modules and collect ``base_class`` subclasses.
|
|
|
|
``package_name`` mirrors the runtime namespaces (``local_plugins`` for
|
|
commands and ``local_services`` for services), keeping their relative
|
|
imports isolated even when files in both directories share a name.
|
|
"""
|
|
import importlib.util
|
|
import sys
|
|
import types
|
|
from pathlib import Path
|
|
|
|
found: list[type] = []
|
|
if not os.path.isdir(directory):
|
|
return found
|
|
|
|
local_path = Path(directory)
|
|
|
|
# Ensure the synthetic parent package exists for relative imports. Refresh
|
|
# its search path in case local_dir_path changed during a config reload.
|
|
pkg = sys.modules.get(package_name)
|
|
if pkg is None:
|
|
pkg = types.ModuleType(package_name)
|
|
sys.modules[package_name] = pkg
|
|
pkg.__path__ = [str(local_path), *(package_paths or [])]
|
|
|
|
for fpath in sorted(local_path.glob("*.py")):
|
|
if fpath.name == "__init__.py":
|
|
continue
|
|
stem = fpath.stem
|
|
module_name = f"{package_name}.{stem}"
|
|
|
|
try:
|
|
spec = importlib.util.spec_from_file_location(module_name, fpath)
|
|
if spec is None or spec.loader is None:
|
|
if logger:
|
|
logger.warning("Could not create spec for local plugin %s", fpath)
|
|
continue
|
|
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[module_name] = module
|
|
spec.loader.exec_module(module)
|
|
|
|
bases = [base_class]
|
|
if local_base_module:
|
|
local_base = getattr(
|
|
sys.modules.get(f"{package_name}.{local_base_module}"),
|
|
base_class.__name__,
|
|
None,
|
|
)
|
|
if inspect.isclass(local_base):
|
|
bases.append(local_base)
|
|
|
|
for _n, obj in inspect.getmembers(module, inspect.isclass):
|
|
if (
|
|
obj not in bases
|
|
and any(issubclass(obj, candidate) for candidate in bases)
|
|
and obj.__module__ == module_name
|
|
):
|
|
found.append(obj)
|
|
break
|
|
except Exception as exc: # noqa: BLE001 - never let one bad plugin break the list
|
|
if logger:
|
|
logger.warning("Could not import local plugin %s for settings discovery: %s", stem, exc)
|
|
continue
|
|
|
|
return found
|
|
|
|
|
|
def build_plugin_settings_view(
|
|
config: configparser.ConfigParser,
|
|
*,
|
|
logger=None,
|
|
commands_dir: Optional[str] = None,
|
|
services_dir: Optional[str] = None,
|
|
local_commands_dir: Optional[str] = None,
|
|
local_services_dir: Optional[str] = None,
|
|
) -> list[dict]:
|
|
"""Assemble the per-plugin settings view for the web UI.
|
|
|
|
Returns a list of dicts, one per discovered command/service::
|
|
|
|
{name, kind, section, label, description, category,
|
|
enabled, source, has_schema, fields, values}
|
|
|
|
``fields`` is the plugin's ``settings_schema`` with a resolved ``value`` on
|
|
each field. ``values`` is the raw current config section (minus ``enabled``)
|
|
used by the generic editor for plugins without a schema.
|
|
|
|
``local_commands_dir`` and ``local_services_dir`` auto-detect the
|
|
``local/commands`` and ``local/service_plugins`` directories (relative to
|
|
the project root) when not supplied, and are tagged with ``source="local"``.
|
|
"""
|
|
# Import bases lazily so this module stays import-light.
|
|
from modules.commands.base_command import BaseCommand
|
|
from modules.service_plugins.base_service import BaseServicePlugin
|
|
|
|
here = os.path.dirname(__file__)
|
|
commands_dir = commands_dir or os.path.join(here, "commands")
|
|
services_dir = services_dir or os.path.join(here, "service_plugins")
|
|
|
|
view: list[dict] = []
|
|
|
|
# --- Commands (default enabled = True; they're active unless turned off) ---
|
|
for cls in _discover_classes(BaseCommand, "modules.commands", commands_dir, logger):
|
|
name = getattr(cls, "name", "") or cls.__name__.lower().replace("command", "")
|
|
if not name:
|
|
continue
|
|
section = command_section_name(name)
|
|
try:
|
|
view.append(_assemble_entry(
|
|
config, cls, kind="command", name=name, section=section,
|
|
label=name.replace("_", " ").title(),
|
|
description=getattr(cls, "description", "") or "",
|
|
category=getattr(cls, "category", "general") or "general",
|
|
# Commands are active unless turned off, except plugins that
|
|
# declare themselves opt-in (e.g. Announcements, Greeter read
|
|
# 'enabled' with fallback=False in __init__).
|
|
enabled_default=bool(getattr(cls, "settings_enabled_default", True)),
|
|
source="base",
|
|
))
|
|
except Exception as exc: # noqa: BLE001 - one bad plugin must not break the list
|
|
if logger:
|
|
logger.warning("Skipping command %s in settings view: %s", name, exc)
|
|
|
|
# --- Local Commands (from local/commands directory) ---
|
|
if local_commands_dir is None:
|
|
# Auto-detect local/commands directory relative to the project root
|
|
local_path = os.path.join(here, "..", "local", "commands")
|
|
if os.path.isdir(local_path):
|
|
local_commands_dir = local_path
|
|
|
|
if local_commands_dir and os.path.isdir(local_commands_dir):
|
|
for cls in _discover_local_classes(
|
|
BaseCommand,
|
|
local_commands_dir,
|
|
package_name="local_plugins",
|
|
local_base_module="base_command",
|
|
logger=logger,
|
|
):
|
|
name = getattr(cls, "name", "") or cls.__name__.lower().replace("command", "")
|
|
if not name:
|
|
continue
|
|
if any(entry["kind"] == "command" and entry["name"] == name for entry in view):
|
|
if logger:
|
|
logger.warning(
|
|
"Local command %s is already present in the settings view; skipping",
|
|
name,
|
|
)
|
|
continue
|
|
section = command_section_name(name)
|
|
try:
|
|
view.append(_assemble_entry(
|
|
config, cls, kind="command", name=name, section=section,
|
|
label=name.replace("_", " ").title(),
|
|
description=getattr(cls, "description", "") or "",
|
|
category=getattr(cls, "category", "general") or "general",
|
|
enabled_default=bool(getattr(cls, "settings_enabled_default", True)),
|
|
source="local",
|
|
))
|
|
except Exception as exc: # noqa: BLE001 - one bad plugin must not break the list
|
|
if logger:
|
|
logger.warning("Skipping local command %s in settings view: %s", name, exc)
|
|
|
|
# --- Services (default enabled = False; must opt in) ---
|
|
for cls in _discover_classes(BaseServicePlugin, "modules.service_plugins", services_dir, logger):
|
|
section = service_section_name(cls)
|
|
name = getattr(cls, "name", "") or cls.__name__.lower().replace("service", "")
|
|
try:
|
|
view.append(_assemble_entry(
|
|
config, cls, kind="service", name=name, section=section,
|
|
label=section.replace("_", " "),
|
|
description=getattr(cls, "description", "") or "",
|
|
category="service",
|
|
enabled_default=bool(getattr(cls, "settings_enabled_default", False)),
|
|
source="base",
|
|
))
|
|
except Exception as exc: # noqa: BLE001 - one bad plugin must not break the list
|
|
if logger:
|
|
logger.warning("Skipping service %s in settings view: %s", name, exc)
|
|
|
|
# --- Local Services (from local/service_plugins directory) ---
|
|
if local_services_dir is None:
|
|
# Auto-detect local/service_plugins directory relative to the project root
|
|
local_services_path = os.path.join(here, "..", "local", "service_plugins")
|
|
if os.path.isdir(local_services_path):
|
|
local_services_dir = local_services_path
|
|
|
|
if local_services_dir and os.path.isdir(local_services_dir):
|
|
for cls in _discover_local_classes(
|
|
BaseServicePlugin,
|
|
local_services_dir,
|
|
package_name="local_services",
|
|
package_paths=[os.path.join(here, "service_plugins"), here],
|
|
local_base_module="base_service",
|
|
logger=logger,
|
|
):
|
|
section = service_section_name(cls)
|
|
name = getattr(cls, "name", "") or cls.__name__.lower().replace("service", "")
|
|
if any(entry["kind"] == "service" and entry["name"] == name for entry in view):
|
|
if logger:
|
|
logger.warning(
|
|
"Local service %s is already present in the settings view; skipping",
|
|
name,
|
|
)
|
|
continue
|
|
try:
|
|
view.append(_assemble_entry(
|
|
config, cls, kind="service", name=name, section=section,
|
|
label=section.replace("_", " "),
|
|
description=getattr(cls, "description", "") or "",
|
|
category="service",
|
|
enabled_default=bool(getattr(cls, "settings_enabled_default", False)),
|
|
source="local",
|
|
))
|
|
except Exception as exc: # noqa: BLE001 - one bad plugin must not break the list
|
|
if logger:
|
|
logger.warning("Skipping local service %s in settings view: %s", name, exc)
|
|
|
|
view.sort(key=lambda e: (e["kind"], e["label"].lower()))
|
|
return view
|
|
|
|
|
|
def _assemble_entry(
|
|
config: configparser.ConfigParser,
|
|
cls: type,
|
|
*,
|
|
kind: str,
|
|
name: str,
|
|
section: str,
|
|
label: str,
|
|
description: str,
|
|
category: str,
|
|
enabled_default: bool,
|
|
source: str = "base",
|
|
) -> dict:
|
|
schema = list(getattr(cls, "settings_schema", []) or [])
|
|
fields: list[dict] = []
|
|
for field in schema:
|
|
if not isinstance(field, dict) or "key" not in field:
|
|
continue
|
|
resolved = dict(field)
|
|
resolved["id"] = field_id(field)
|
|
resolved["value"] = _read_typed(config, section, field)
|
|
if field.get("type") == "password":
|
|
# Never send the plaintext secret to the browser; the UI shows a
|
|
# placeholder when has_value is true and only submits a new value
|
|
# if the user actually types one.
|
|
resolved["has_value"] = bool(resolved["value"])
|
|
resolved["value"] = ""
|
|
fields.append(resolved)
|
|
|
|
# Every command can restrict which channels it responds in
|
|
# ([Name_Command] channels). Surface it on every command card unless the
|
|
# plugin already declares its own channels field.
|
|
if kind == "command" and not any(f["key"].lower() == "channels" for f in fields):
|
|
channels_field = {
|
|
"key": "channels", "label": "Channels", "type": "list", "default": "",
|
|
"help": ("Comma-separated channels this command responds in. Blank = the "
|
|
"global monitored channels (default). DMs always work."),
|
|
}
|
|
channels_field["value"] = _read_typed(config, section, channels_field)
|
|
channels_field["id"] = "channels"
|
|
fields.append(channels_field)
|
|
|
|
# Keys handled elsewhere shouldn't appear in the raw "Other config values"
|
|
# editor: the enable toggle, its legacy *_enabled aliases in this section,
|
|
# the channels field above, and aliases (managed via keywords).
|
|
skip_keys = {"enabled", "channels", "aliases"}
|
|
for legacy_section, legacy_key in _ENABLED_LEGACY_ALIASES.get(section, []):
|
|
if legacy_section == section:
|
|
skip_keys.add(legacy_key.lower())
|
|
schema_keys = {f["key"].lower() for f in fields}
|
|
|
|
# Raw values for the fallback editor: every other key in the section.
|
|
# raw=True so '%' in values doesn't trip configparser interpolation.
|
|
values: dict[str, str] = {}
|
|
if config.has_section(section):
|
|
for key, raw in config.items(section, raw=True):
|
|
if key.lower() in skip_keys or key.lower() in schema_keys:
|
|
continue
|
|
values[key] = raw
|
|
|
|
return {
|
|
"name": name,
|
|
"kind": kind,
|
|
"section": section,
|
|
"label": label,
|
|
"description": description,
|
|
"category": category,
|
|
"enabled": read_enabled(config, section, enabled_default),
|
|
"source": source,
|
|
"has_schema": bool(fields),
|
|
"fields": fields,
|
|
"values": values,
|
|
"dynamic_sections": _read_dynamic_sections(config, cls),
|
|
"repeating_blocks": _read_repeating_blocks(config, cls, section),
|
|
}
|
|
|
|
|
|
def _read_repeating_blocks(
|
|
config: configparser.ConfigParser, cls: type, section: str
|
|
) -> list[dict]:
|
|
"""Build the repeating structured-block editors a plugin declares.
|
|
|
|
A plugin may declare ``settings_repeating_blocks`` for families of indexed
|
|
keys like ``mqtt1_server``, ``mqtt2_server`` … (PacketCapture). Each block
|
|
is a structured group with its own sub-schema and an enable toggle
|
|
(``mqtt<N>_enabled``). Blocks are returned sorted by index; ``values``
|
|
carries every sub-key (including ones not in the sub-schema) so unknown
|
|
fields survive a save.
|
|
"""
|
|
decls = list(getattr(cls, "settings_repeating_blocks", []) or [])
|
|
result: list[dict] = []
|
|
for d in decls:
|
|
if not isinstance(d, dict) or "id" not in d:
|
|
continue
|
|
bid = d["id"]
|
|
enabled_field = d.get("enabled_field", "enabled")
|
|
rx = re.compile(rf"^{re.escape(bid)}(\d+)_(.+)$")
|
|
groups: dict[int, dict[str, str]] = {}
|
|
if config.has_section(section):
|
|
for key, raw in config.items(section, raw=True):
|
|
m = rx.match(key)
|
|
if m:
|
|
groups.setdefault(int(m.group(1)), {})[m.group(2)] = raw
|
|
blocks = []
|
|
for idx in sorted(groups):
|
|
sub = groups[idx]
|
|
enabled = str(sub.get(enabled_field, "true")).strip().lower() in _TRUE
|
|
values = {k: v for k, v in sub.items() if k != enabled_field}
|
|
blocks.append({"index": idx, "enabled": enabled, "values": values})
|
|
result.append({
|
|
"id": bid,
|
|
"label": d.get("label", bid),
|
|
"item_label": d.get("item_label", "item"),
|
|
"help": d.get("help", ""),
|
|
"enabled_field": enabled_field,
|
|
"fields": list(d.get("fields", [])),
|
|
"blocks": blocks,
|
|
})
|
|
return result
|
|
|
|
|
|
def read_section_items(
|
|
config: configparser.ConfigParser, section: str, prefix: str = ""
|
|
) -> list[dict]:
|
|
"""Return a section's key/value pairs as ``[{"key", "value"}, ...]`` (raw).
|
|
|
|
When ``prefix`` is given, only keys starting with it are returned and the
|
|
prefix is stripped from the displayed key (e.g. ``agency.county1`` ->
|
|
``county1``). This lets a dynamic editor manage just the prefixed keys in a
|
|
section that also holds typed schema fields.
|
|
"""
|
|
items: list[dict] = []
|
|
if not config.has_section(section):
|
|
return items
|
|
pl = prefix.lower()
|
|
# raw=True so '%' in values doesn't trip interpolation.
|
|
for key, raw in config.items(section, raw=True):
|
|
if prefix:
|
|
if not key.lower().startswith(pl):
|
|
continue
|
|
items.append({"key": key[len(prefix):], "value": raw})
|
|
else:
|
|
items.append({"key": key, "value": raw})
|
|
return items
|
|
|
|
|
|
def _read_dynamic_sections(config: configparser.ConfigParser, cls: type) -> list[dict]:
|
|
"""Build the dynamic key/value section editors a plugin declares.
|
|
|
|
A plugin may declare ``settings_dynamic_sections`` — a list of descriptors
|
|
for whole config sections that hold a free-form, user-extendable list of
|
|
``key = value`` entries (e.g. ``[Channels_List]``). The web viewer renders
|
|
these as add/edit/delete tables.
|
|
"""
|
|
declared = list(getattr(cls, "settings_dynamic_sections", []) or [])
|
|
result: list[dict] = []
|
|
for ds in declared:
|
|
if not isinstance(ds, dict) or "section" not in ds:
|
|
continue
|
|
sec = ds["section"]
|
|
prefix = ds.get("key_prefix", "") or ""
|
|
result.append({
|
|
"section": sec,
|
|
"key_prefix": prefix,
|
|
"label": ds.get("label", sec),
|
|
"help": ds.get("help", ""),
|
|
"key_label": ds.get("key_label", "Key"),
|
|
"value_label": ds.get("value_label", "Value"),
|
|
"key_placeholder": ds.get("key_placeholder", ""),
|
|
"value_placeholder": ds.get("value_placeholder", ""),
|
|
"items": read_section_items(config, sec, prefix),
|
|
})
|
|
return result
|