sections: init dynamic section

parent 2485c3e6
from .classic import ClassicSection
from .custom import CustomSection
from .dynamic import DynamicSection
class SectionFactory:
def __init__(self):
self.sections = {
'classic': ClassicSection,
'custom': CustomSection,
'dynamic': DynamicSection,
}
def create_section(self, section_data, module):
......
class BaseSection():
def __init__(self, section_data, module):
self.section_data = section_data
self.module = module
self.settings = []
self.name = module.get_translation(section_data['name'])
self.weight = section_data.get('weight', 0)
self.page = section_data.get('page')
......@@ -10,7 +10,6 @@ class ClassicSection(BaseSection):
self.logger = logging.getLogger(f"{self.__class__.__name__}[{self.name}]")
self.settings = [Setting(s, module) for s in section_data.get('settings', [])]
self.settings = sorted(self.settings, key=lambda s: s.weight, reverse=True)
self.module = module
self.module.add_section(self)
......
......@@ -13,7 +13,6 @@ class CustomSection(BaseSection):
self.logger = logging.getLogger(f"{self.__class__.__name__}[{self.name}]")
self.settings = [CustomSetting(s, module, self) for s in section_data.get('settings', [])]
self.settings_dict = {s.orig_name: s for s in self.settings}
self.module = module
self.module.add_section(self)
self._callback_buffer = []
......
import json
import logging
import re
import subprocess
from .base import BaseSection
from .custom import CustomSection
from ..setting.custom_setting import CustomSetting
class DynamicSection(CustomSection):
"""
Section that dynamically generates settings from a command output.
The generator_command should return a JSON array of objects.
Each object is used to populate the setting_template.
Example YAML:
sections:
- name: "Network Interfaces"
type: dynamic
generator_command: "ip -j link show"
setting_template:
name: "{ifname}"
type: boolean
get_command: "ip link show {ifname} | grep -q UP && echo True || echo False"
set_command: "ip link set {ifname} {value}"
"""
def __init__(self, section_data, module):
# Вызываем BaseSection напрямую, минуя CustomSection.__init__,
# который создаёт settings из section_data
BaseSection.__init__(self, section_data, module)
self.logger = logging.getLogger(f"{self.__class__.__name__}[{self.name}]")
self.generator_command = section_data.get('generator_command')
self.setting_template = section_data.get('setting_template', {})
self._generate_settings()
self.settings_dict = {s.orig_name: s for s in self.settings}
self._callback_buffer = []
self.module.add_section(self)
def _generate_settings(self):
items = self._execute_generator()
if not items:
self.logger.warning("Generator returned no items")
return
for item in items:
try:
setting_data = self._apply_template(self.setting_template, item)
setting = CustomSetting(setting_data, self.module, self)
self.settings.append(setting)
except Exception as e:
self.logger.error(f"Error creating setting from item {item}: {e}")
self.settings = sorted(self.settings, key=lambda s: s.weight, reverse=True)
self.logger.info(f"Generated {len(self.settings)} settings")
def _execute_generator(self):
if not self.generator_command:
self.logger.error("No generator_command specified")
return []
try:
result = subprocess.run(
self.generator_command,
shell=True,
capture_output=True,
text=True,
timeout=30
)
if result.returncode != 0:
self.logger.error(f"Generator command failed: {result.stderr}")
return []
output = result.stdout.strip()
if not output:
return []
data = json.loads(output)
# Normalize to list
if isinstance(data, dict):
# Single object -> wrap in list
return [data]
elif isinstance(data, list):
# Normalize simple lists: ["a", "b"] -> [{"_item": "a"}, {"_item": "b"}]
if data and not isinstance(data[0], dict):
return [{"_item": item, "_index": i} for i, item in enumerate(data)]
return data
else:
# Scalar value
return [{"_item": data, "_index": 0}]
except json.JSONDecodeError as e:
self.logger.error(f"Failed to parse generator output as JSON: {e}")
# Try line-by-line parsing
lines = result.stdout.strip().split('\n')
if lines:
return [{"_item": line.strip(), "_index": i} for i, line in enumerate(lines) if line.strip()]
return []
except subprocess.TimeoutExpired:
self.logger.error("Generator command timed out")
return []
except Exception as e:
self.logger.error(f"Generator execution error: {e}")
return []
def _apply_template(self, template, item):
"""
Recursively apply item values to template.
- "{key}" in string context -> string substitution
- "{key}" as entire value -> preserves type (dict, list, etc.)
- Supports nested paths: "{a.b.c}"
"""
if isinstance(template, str):
# Check if entire string is a single placeholder
match = re.fullmatch(r'\{(\w+(?:\.\w+)*)\}', template.strip())
if match:
# Return value as-is (preserves type)
value = self._get_nested(item, match.group(1))
if value is not None:
return value
return template
# String interpolation with multiple placeholders
return self._format_string(template, item)
elif isinstance(template, dict):
return {k: self._apply_template(v, item) for k, v in template.items()}
elif isinstance(template, list):
return [self._apply_template(v, item) for v in template]
return template
def _format_string(self, template, item):
"""Format string with item values, supporting nested paths"""
def replacer(match):
path = match.group(1)
value = self._get_nested(item, path)
if value is not None:
return str(value)
return match.group(0)
return re.sub(r'\{(\w+(?:\.\w+)*)\}', replacer, template)
def _get_nested(self, data, path):
"""Get value by dot-separated path: 'a.b.c' -> data['a']['b']['c']"""
try:
for key in path.split('.'):
if isinstance(data, dict):
data = data.get(key)
elif isinstance(data, (list, tuple)) and key.isdigit():
data = data[int(key)]
else:
return None
if data is None:
return None
return data
except (KeyError, IndexError, TypeError):
return None
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment