Commit 244440d3 authored by Roman Alifanov's avatar Roman Alifanov

Merge branch 'feature/dynamic-sections' into 'master'

sections: init dynamic section See merge request !3
parents 5f610328 ef6c500a
from .classic import ClassicSection from .classic import ClassicSection
from .custom import CustomSection from .custom import CustomSection
from .dynamic import DynamicSection
class SectionFactory: class SectionFactory:
def __init__(self): def __init__(self):
self.sections = { self.sections = {
'classic': ClassicSection, 'classic': ClassicSection,
'custom': CustomSection, 'custom': CustomSection,
'dynamic': DynamicSection,
} }
def create_section(self, section_data, module): def create_section(self, section_data, module):
......
class BaseSection(): class BaseSection():
def __init__(self, section_data, module): def __init__(self, section_data, module):
self.section_data = section_data self.section_data = section_data
self.module = module
self.settings = []
self.name = module.get_translation(section_data['name']) self.name = module.get_translation(section_data['name'])
self.weight = section_data.get('weight', 0) self.weight = section_data.get('weight', 0)
self.page = section_data.get('page') self.page = section_data.get('page')
...@@ -10,7 +10,6 @@ class ClassicSection(BaseSection): ...@@ -10,7 +10,6 @@ class ClassicSection(BaseSection):
self.logger = logging.getLogger(f"{self.__class__.__name__}[{self.name}]") self.logger = logging.getLogger(f"{self.__class__.__name__}[{self.name}]")
self.settings = [Setting(s, module) for s in section_data.get('settings', [])] 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.settings = sorted(self.settings, key=lambda s: s.weight, reverse=True)
self.module = module
self.module.add_section(self) self.module.add_section(self)
......
...@@ -13,7 +13,6 @@ class CustomSection(BaseSection): ...@@ -13,7 +13,6 @@ class CustomSection(BaseSection):
self.logger = logging.getLogger(f"{self.__class__.__name__}[{self.name}]") 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 = [CustomSetting(s, module, self) for s in section_data.get('settings', [])]
self.settings_dict = {s.orig_name: s for s in self.settings} self.settings_dict = {s.orig_name: s for s in self.settings}
self.module = module
self.module.add_section(self) self.module.add_section(self)
self._callback_buffer = [] 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