Device search is working. Working on The multidevice seach, WIP
This commit is contained in:
@@ -0,0 +1,133 @@
|
||||
import difflib
|
||||
|
||||
from textual.containers import Horizontal
|
||||
from textual.css.query import NoMatches
|
||||
from textual.message import Message
|
||||
from textual.widget import Widget
|
||||
from textual.widgets import Button, SelectionList, Static, Switch, TextArea
|
||||
|
||||
|
||||
class MultiAgentSelector(Widget):
|
||||
class AgentsSelected(Message):
|
||||
def __init__(self, selected_agents):
|
||||
super().__init__()
|
||||
self.selected_agents = selected_agents
|
||||
|
||||
def __init__(self, all_agents: list[dict]):
|
||||
super().__init__()
|
||||
self.all_agents = all_agents
|
||||
self._match_type = "fuzzy"
|
||||
|
||||
@property
|
||||
def match_type(self):
|
||||
return self._match_type
|
||||
|
||||
@match_type.setter
|
||||
def match_type(self, value):
|
||||
self._match_type = value
|
||||
|
||||
def compose(self):
|
||||
yield Static("🔍 Multi-Agent Selector", id="selector_title")
|
||||
text_area = TextArea(
|
||||
id="device_input",
|
||||
placeholder="Paste device names here (one per line)",
|
||||
)
|
||||
text_area.styles.height = 10
|
||||
text_area.styles.overflow_y = "auto"
|
||||
yield text_area
|
||||
|
||||
with Horizontal() as switch_container:
|
||||
switch_container.styles.height = "auto"
|
||||
switch_container.styles.align = ("left", "middle")
|
||||
switch_label = Static("Match Type: Fuzzy", id="match_switch_label")
|
||||
switch_label.styles.width = "auto"
|
||||
switch_label.styles.padding = (0, 1)
|
||||
yield switch_label
|
||||
|
||||
switch = Switch(value=False, id="match_switch")
|
||||
switch.styles.width = "auto"
|
||||
yield switch
|
||||
|
||||
yield Button("Search", id="search_button")
|
||||
|
||||
yield SelectionList(id="match_results")
|
||||
yield Static(id="unmatched_label")
|
||||
|
||||
yield Horizontal(
|
||||
Button("Select All", id="select_all"),
|
||||
Button("Select None", id="select_none"),
|
||||
)
|
||||
|
||||
yield Button("Continue with Selected", id="submit_selection", variant="primary")
|
||||
|
||||
def on_switch_changed(self, event: Switch.Changed):
|
||||
self.match_type = "exact" if event.value else "fuzzy"
|
||||
self.query_one("#match_switch_label", Static).update(
|
||||
f"Match Type: {self.match_type.capitalize()}"
|
||||
)
|
||||
|
||||
def on_button_pressed(self, event: Button.Pressed):
|
||||
btn_id = event.button.id # can be None for internal buttons
|
||||
|
||||
# Only query when needed
|
||||
try:
|
||||
match_list = self.query_one("#match_results", SelectionList)
|
||||
except NoMatches:
|
||||
# UI not mounted yet or id changed—just ignore gracefully
|
||||
return
|
||||
|
||||
if btn_id == "select_all":
|
||||
match_list.select_all()
|
||||
event.stop()
|
||||
elif btn_id == "select_none":
|
||||
match_list.deselect_all()
|
||||
event.stop()
|
||||
elif btn_id == "submit_selection":
|
||||
selected = list(match_list.selected)
|
||||
self.post_message(self.AgentsSelected(selected))
|
||||
event.stop()
|
||||
elif btn_id == "search_button":
|
||||
self.update_matches()
|
||||
event.stop()
|
||||
|
||||
def update_matches(self):
|
||||
raw_input = self.query_one("#device_input", TextArea).text.strip()
|
||||
device_names = [line.strip() for line in raw_input.split("\n") if line.strip()]
|
||||
matched, unmatched = self.match_devices(device_names)
|
||||
match_list = self.query_one("#match_results", SelectionList)
|
||||
match_list.clear_options()
|
||||
for name in matched:
|
||||
match_list.add_option((name, name))
|
||||
unmatched_label = self.query_one("#unmatched_label", Static)
|
||||
if unmatched:
|
||||
unmatched_label.update(f"⚠️ No matches for: {', '.join(unmatched)}")
|
||||
else:
|
||||
unmatched_label.update("")
|
||||
|
||||
def match_devices(self, device_names: list[str]) -> tuple[list[str], list[str]]:
|
||||
if not self.all_agents or not device_names:
|
||||
return [], device_names
|
||||
agent_names = [agent["hostname"] for agent in self.all_agents]
|
||||
matched = set()
|
||||
unmatched = []
|
||||
for name in device_names:
|
||||
if self.match_type == "exact":
|
||||
# Case-insensitive exact match
|
||||
name_lower = name.lower()
|
||||
exact_match = None
|
||||
for agent_name in agent_names:
|
||||
if agent_name.lower() == name_lower:
|
||||
exact_match = agent_name
|
||||
break
|
||||
if exact_match:
|
||||
matched.add(exact_match)
|
||||
else:
|
||||
unmatched.append(name)
|
||||
else:
|
||||
# Fuzzy match
|
||||
matches = difflib.get_close_matches(name, agent_names, n=5, cutoff=0.5)
|
||||
if matches:
|
||||
matched.update(matches)
|
||||
else:
|
||||
unmatched.append(name)
|
||||
return sorted(matched), unmatched
|
||||
@@ -0,0 +1,199 @@
|
||||
import logging
|
||||
|
||||
from rich.text import Text
|
||||
from textual.containers import Horizontal, Vertical
|
||||
from textual.widget import Widget
|
||||
from textual.widgets import Input, OptionList, Static, Tree
|
||||
from textual.widgets.option_list import Option
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class PolicyTreeWidget(Widget):
|
||||
"""Widget for displaying and searching a hierarchical policy tree."""
|
||||
|
||||
def __init__(self, policies, devices):
|
||||
super().__init__()
|
||||
self.policies = policies
|
||||
self.devices = devices
|
||||
self.last_highlighted_node = None
|
||||
|
||||
def compose(self):
|
||||
# Left: Policy Tree
|
||||
policy_tree = Tree("Policies", id="policy_tree")
|
||||
policy_tree.styles.width = "2fr"
|
||||
policy_tree.styles.height = "100%"
|
||||
|
||||
# Right: Search + Details
|
||||
label = Static("Device Search:")
|
||||
search_box = Input(
|
||||
placeholder="Search policies or devices...", id="tree_search"
|
||||
)
|
||||
details_pane = Static("", id="details_pane")
|
||||
|
||||
with Horizontal():
|
||||
yield policy_tree
|
||||
with Vertical() as right_pane:
|
||||
right_pane.styles.width = "3fr"
|
||||
yield label
|
||||
yield search_box
|
||||
yield details_pane
|
||||
|
||||
def on_mount(self) -> None:
|
||||
"""Build the tree after mounting."""
|
||||
self._build_tree()
|
||||
|
||||
def _build_tree(self) -> None:
|
||||
"""Build the policy tree structure."""
|
||||
policy_tree = self.query_one("#policy_tree", Tree)
|
||||
node_map = {}
|
||||
|
||||
# Top-level policies
|
||||
for _, policy in self.policies.iterrows():
|
||||
if policy["parent"] == "global-policy-settings":
|
||||
node = policy_tree.root.add(label=policy["name"], data=policy.to_dict())
|
||||
node_map[policy["groupid"]] = node
|
||||
|
||||
# Child policies
|
||||
for _, policy in self.policies.iterrows():
|
||||
parent_id = policy["parent"]
|
||||
if parent_id in node_map:
|
||||
parent_node = node_map[parent_id]
|
||||
node = parent_node.add(label=policy["name"], data=policy.to_dict())
|
||||
node_map[policy["groupid"]] = node
|
||||
|
||||
# Devices under policies
|
||||
for _, device in self.devices.iterrows():
|
||||
group_id = device["groupid"]
|
||||
if group_id in node_map:
|
||||
parent_node = node_map[group_id]
|
||||
label = device["hostname"]
|
||||
parent_node.add(label=label, data=device.to_dict())
|
||||
|
||||
def _collect_tree_nodes(self, node, all_nodes):
|
||||
"""Helper to recursively collect all nodes from a tree."""
|
||||
all_nodes.append(node)
|
||||
for child in node.children:
|
||||
self._collect_tree_nodes(child, all_nodes)
|
||||
|
||||
def _remove_match_selector(self):
|
||||
"""Safely remove match selector widgets."""
|
||||
try:
|
||||
existing = self.query("#match_selector")
|
||||
for widget in existing:
|
||||
if widget.is_attached:
|
||||
widget.remove()
|
||||
except Exception as exc:
|
||||
logger.debug("Failed to remove match_selector: %s", exc)
|
||||
|
||||
def on_tree_node_selected(self, message: Tree.NodeSelected) -> None:
|
||||
"""Handle tree node selection."""
|
||||
node = message.node
|
||||
data = node.data
|
||||
details_pane = self.query_one("#details_pane", Static)
|
||||
|
||||
# Reset previous highlight
|
||||
if self.last_highlighted_node is not None:
|
||||
original_label = str(self.last_highlighted_node.label).strip()
|
||||
# Remove any styling
|
||||
if isinstance(self.last_highlighted_node.label, Text):
|
||||
original_label = self.last_highlighted_node.label.plain
|
||||
self.last_highlighted_node.set_label(original_label)
|
||||
|
||||
# Apply highlight to current node
|
||||
label_text = str(node.label).strip()
|
||||
if isinstance(node.label, Text):
|
||||
label_text = node.label.plain
|
||||
highlighted_label = Text(label_text, style="reverse bold")
|
||||
node.set_label(highlighted_label)
|
||||
self.last_highlighted_node = node
|
||||
|
||||
# Update details pane
|
||||
if data:
|
||||
details = "\n".join(f"{key}: {value}" for key, value in data.items())
|
||||
else:
|
||||
details = f"Selected: {node.label}"
|
||||
details_pane.update(details)
|
||||
|
||||
# Stop event from bubbling
|
||||
message.stop()
|
||||
|
||||
def on_input_submitted(self, message: Input.Submitted) -> None:
|
||||
"""Handle search input submission."""
|
||||
# Remove existing match selector FIRST
|
||||
self._remove_match_selector()
|
||||
|
||||
query = message.value.strip().lower()
|
||||
tree = self.query_one("#policy_tree", Tree)
|
||||
details_pane = self.query_one("#details_pane", Static)
|
||||
|
||||
all_nodes = []
|
||||
self._collect_tree_nodes(tree.root, all_nodes)
|
||||
|
||||
label_to_node = {}
|
||||
for node in all_nodes:
|
||||
label_text = str(node.label).lower()
|
||||
label_to_node[label_text] = node
|
||||
if node.data:
|
||||
for key, value in node.data.items():
|
||||
if isinstance(value, str):
|
||||
label_to_node[value.lower()] = node
|
||||
|
||||
# Wildcard-style substring match
|
||||
matches = sorted([label for label in label_to_node if query in label])
|
||||
|
||||
if matches:
|
||||
# Try to reuse existing match_selector or create new one
|
||||
try:
|
||||
option_list = self.query_one("#match_selector", OptionList)
|
||||
option_list.clear_options()
|
||||
option_list.display = True # Ensure it's visible
|
||||
except:
|
||||
option_list = OptionList(id="match_selector")
|
||||
# Mount to the details pane's parent (the Vertical container)
|
||||
details_pane.parent.mount(option_list)
|
||||
|
||||
for label in matches:
|
||||
option_list.add_option(Option(label, id=f"match_{label}"))
|
||||
|
||||
details_pane.update(f"Found {len(matches)} matches. Select one below.")
|
||||
else:
|
||||
# Hide or remove the match_selector when no matches
|
||||
self._remove_match_selector()
|
||||
details_pane.update("No matches found.")
|
||||
|
||||
def on_option_list_option_selected(self, event: OptionList.OptionSelected) -> None:
|
||||
"""Handle selection from search results."""
|
||||
selected_id = event.option.id.replace("match_", "")
|
||||
tree = self.query_one("#policy_tree", Tree)
|
||||
details_pane = self.query_one("#details_pane", Static)
|
||||
|
||||
# Find the node
|
||||
all_nodes = []
|
||||
self._collect_tree_nodes(tree.root, all_nodes)
|
||||
|
||||
label_to_node = {str(node.label).lower(): node for node in all_nodes}
|
||||
match_node = label_to_node.get(selected_id.lower())
|
||||
|
||||
if match_node:
|
||||
# Expand path (original working logic)
|
||||
node = match_node
|
||||
path = []
|
||||
while node:
|
||||
path.insert(0, node)
|
||||
node = node.parent
|
||||
for node in path:
|
||||
node.expand()
|
||||
|
||||
tree.select_node(match_node)
|
||||
tree.scroll_to_node(match_node)
|
||||
|
||||
match_node.set_label(Text(str(match_node.label), style="reverse bold"))
|
||||
details_pane.update(f"Selected: {match_node.label}")
|
||||
|
||||
# Remove the match_selector after selection
|
||||
try:
|
||||
option_list = self.query_one("#match_selector", OptionList)
|
||||
option_list.remove()
|
||||
except:
|
||||
pass
|
||||
@@ -0,0 +1,45 @@
|
||||
from textual.containers import Vertical
|
||||
from textual.message import Message
|
||||
from textual.widget import Widget
|
||||
from textual.widgets import Button, Static
|
||||
|
||||
|
||||
class ThemeSelector(Widget):
|
||||
"""Widget for selecting and applying Textual themes."""
|
||||
|
||||
class ThemeSelected(Message):
|
||||
"""Message posted when a theme is selected."""
|
||||
|
||||
def __init__(self, theme_name: str):
|
||||
super().__init__()
|
||||
self.theme_name = theme_name
|
||||
|
||||
AVAILABLE_THEMES = [
|
||||
("textual-dark", "textual-dark"),
|
||||
("textual-light", "textual-light"),
|
||||
("nord", "nord"),
|
||||
("gruvbox", "gruvbox"),
|
||||
("catppuccin-mocha", "catppuccin-mocha"),
|
||||
("dracula", "dracula"),
|
||||
("tokyo-night", "tokyo-night"),
|
||||
("monokai", "monokai"),
|
||||
("flexoki", "flexoki"),
|
||||
("catppuccin-latte", "catppuccin-latte"),
|
||||
("solarized-light", "solarized-light"),
|
||||
]
|
||||
|
||||
def compose(self):
|
||||
yield Static("Theme Options", id="theme_title")
|
||||
|
||||
with Vertical() as column:
|
||||
column.styles.width = "1fr"
|
||||
column.styles.height = "auto"
|
||||
|
||||
for label, btn_id in self.AVAILABLE_THEMES:
|
||||
yield Button(label, id=f"set_theme_{btn_id}", compact=True)
|
||||
|
||||
def on_button_pressed(self, event: Button.Pressed) -> None:
|
||||
button_id = event.button.id
|
||||
if button_id and button_id.startswith("set_theme_"):
|
||||
theme_name = button_id.replace("set_theme_", "")
|
||||
self.post_message(self.ThemeSelected(theme_name))
|
||||
Reference in New Issue
Block a user