Refactor: add find_monitors_by_globs() and find_and_get_monitors()
This commit is contained in:
Pim van Pelt
@@ -171,3 +171,105 @@ class KumaClient:
|
||||
except Exception as e:
|
||||
print(f"Error getting group members: {e}")
|
||||
return []
|
||||
|
||||
def find_monitors_by_globs(self, monitor_patterns=None, group_patterns=None):
|
||||
"""Find monitor IDs by name patterns and/or group patterns.
|
||||
|
||||
Args:
|
||||
monitor_patterns: List of monitor name patterns (supports wildcards)
|
||||
group_patterns: List of group name patterns (supports wildcards)
|
||||
|
||||
Returns:
|
||||
List of monitor IDs (integers) that match the criteria
|
||||
"""
|
||||
try:
|
||||
# Check if we have either monitor patterns or group patterns
|
||||
if not monitor_patterns and not group_patterns:
|
||||
print(
|
||||
"Error: Either monitor or group patterns required. "
|
||||
"Specify at least one pattern."
|
||||
)
|
||||
return []
|
||||
|
||||
matched_monitors = []
|
||||
|
||||
# Find monitors by patterns if specified
|
||||
if monitor_patterns:
|
||||
pattern_monitors = self.find_monitors_by_pattern(monitor_patterns)
|
||||
matched_monitors.extend(pattern_monitors)
|
||||
|
||||
# Find monitors by groups if specified
|
||||
if group_patterns:
|
||||
group_monitors = self.get_monitors_in_groups(group_patterns)
|
||||
# Convert to same format as find_monitors_by_pattern
|
||||
group_monitor_objs = [
|
||||
{"id": m.get("id"), "name": m.get("name")} for m in group_monitors
|
||||
]
|
||||
matched_monitors.extend(group_monitor_objs)
|
||||
|
||||
# Remove duplicates while preserving order
|
||||
seen = set()
|
||||
unique_monitors = []
|
||||
for monitor in matched_monitors:
|
||||
if monitor["id"] not in seen:
|
||||
seen.add(monitor["id"])
|
||||
unique_monitors.append(monitor)
|
||||
|
||||
matched_monitors = unique_monitors
|
||||
|
||||
if not matched_monitors:
|
||||
print(
|
||||
"Error: No monitors found matching the specified patterns or groups"
|
||||
)
|
||||
return []
|
||||
|
||||
# Return list of monitor IDs
|
||||
return [monitor["id"] for monitor in matched_monitors]
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error finding monitors by globs: {e}")
|
||||
return []
|
||||
|
||||
def get_monitor_details(self, monitor_ids):
|
||||
"""Get monitor details for display purposes.
|
||||
|
||||
Args:
|
||||
monitor_ids: List of monitor IDs
|
||||
|
||||
Returns:
|
||||
List of dicts with 'id' and 'name' keys
|
||||
"""
|
||||
try:
|
||||
all_monitors = self.api.get_monitors()
|
||||
return [
|
||||
{
|
||||
"id": mid,
|
||||
"name": next(
|
||||
(
|
||||
m.get("name", f"Monitor {mid}")
|
||||
for m in all_monitors
|
||||
if m.get("id") == mid
|
||||
),
|
||||
f"Monitor {mid}",
|
||||
),
|
||||
}
|
||||
for mid in monitor_ids
|
||||
]
|
||||
except Exception as e:
|
||||
print(f"Error getting monitor details: {e}")
|
||||
return []
|
||||
|
||||
def find_and_get_monitors(self, monitor_patterns=None, group_patterns=None):
|
||||
"""Find monitors by patterns/groups and return detailed info.
|
||||
|
||||
Args:
|
||||
monitor_patterns: List of monitor name patterns (supports wildcards)
|
||||
group_patterns: List of group name patterns (supports wildcards)
|
||||
|
||||
Returns:
|
||||
List of dicts with 'id' and 'name' keys, or empty list if none found
|
||||
"""
|
||||
monitor_ids = self.find_monitors_by_globs(monitor_patterns, group_patterns)
|
||||
if not monitor_ids:
|
||||
return []
|
||||
return self.get_monitor_details(monitor_ids)
|
||||
|
||||
@@ -49,46 +49,10 @@ class MaintenanceCommands:
|
||||
):
|
||||
"""Add a new maintenance"""
|
||||
try:
|
||||
# Check if we have either monitor patterns or group patterns
|
||||
if not monitor_patterns and not group_patterns:
|
||||
print(
|
||||
"Error: Either --monitor or --group flag is required. "
|
||||
"Specify at least one pattern."
|
||||
)
|
||||
return
|
||||
|
||||
matched_monitors = []
|
||||
|
||||
# Find monitors by patterns if specified
|
||||
if monitor_patterns:
|
||||
pattern_monitors = self.client.find_monitors_by_pattern(
|
||||
monitor_patterns
|
||||
)
|
||||
matched_monitors.extend(pattern_monitors)
|
||||
|
||||
# Find monitors by groups if specified
|
||||
if group_patterns:
|
||||
group_monitors = self.client.get_monitors_in_groups(group_patterns)
|
||||
# Convert to the same format as find_monitors_by_pattern
|
||||
group_monitor_objs = [
|
||||
{"id": m.get("id"), "name": m.get("name")} for m in group_monitors
|
||||
]
|
||||
matched_monitors.extend(group_monitor_objs)
|
||||
|
||||
# Remove duplicates while preserving order
|
||||
seen = set()
|
||||
unique_monitors = []
|
||||
for monitor in matched_monitors:
|
||||
if monitor["id"] not in seen:
|
||||
seen.add(monitor["id"])
|
||||
unique_monitors.append(monitor)
|
||||
|
||||
matched_monitors = unique_monitors
|
||||
|
||||
matched_monitors = self.client.find_and_get_monitors(
|
||||
monitor_patterns, group_patterns
|
||||
)
|
||||
if not matched_monitors:
|
||||
print(
|
||||
"Error: No monitors found matching the specified patterns or groups"
|
||||
)
|
||||
return
|
||||
|
||||
print(f"Found {len(matched_monitors)} matching monitors:")
|
||||
|
||||
@@ -74,46 +74,10 @@ class MonitorCommands:
|
||||
def pause_monitors(self, monitor_patterns=None, group_patterns=None):
|
||||
"""Pause monitors by patterns and/or groups"""
|
||||
try:
|
||||
# Check if we have either monitor patterns or group patterns
|
||||
if not monitor_patterns and not group_patterns:
|
||||
print(
|
||||
"Error: Either --monitor or --group flag is required. "
|
||||
"Specify at least one pattern."
|
||||
)
|
||||
return
|
||||
|
||||
matched_monitors = []
|
||||
|
||||
# Find monitors by patterns if specified
|
||||
if monitor_patterns:
|
||||
pattern_monitors = self.client.find_monitors_by_pattern(
|
||||
monitor_patterns
|
||||
)
|
||||
matched_monitors.extend(pattern_monitors)
|
||||
|
||||
# Find monitors by groups if specified
|
||||
if group_patterns:
|
||||
group_monitors = self.client.get_monitors_in_groups(group_patterns)
|
||||
# Convert to the same format as find_monitors_by_pattern
|
||||
group_monitor_objs = [
|
||||
{"id": m.get("id"), "name": m.get("name")} for m in group_monitors
|
||||
]
|
||||
matched_monitors.extend(group_monitor_objs)
|
||||
|
||||
# Remove duplicates while preserving order
|
||||
seen = set()
|
||||
unique_monitors = []
|
||||
for monitor in matched_monitors:
|
||||
if monitor["id"] not in seen:
|
||||
seen.add(monitor["id"])
|
||||
unique_monitors.append(monitor)
|
||||
|
||||
matched_monitors = unique_monitors
|
||||
|
||||
matched_monitors = self.client.find_and_get_monitors(
|
||||
monitor_patterns, group_patterns
|
||||
)
|
||||
if not matched_monitors:
|
||||
print(
|
||||
"Error: No monitors found matching the specified patterns or groups"
|
||||
)
|
||||
return
|
||||
|
||||
print(f"Found {len(matched_monitors)} matching monitors to pause:")
|
||||
@@ -142,62 +106,36 @@ class MonitorCommands:
|
||||
):
|
||||
"""Resume monitors by patterns and/or groups, or all paused monitors"""
|
||||
try:
|
||||
# Check if we have either monitor patterns, group patterns, or --all flag
|
||||
if not monitor_patterns and not group_patterns and not resume_all:
|
||||
print("Error: Either --monitor, --group, or --all flag is required.")
|
||||
return
|
||||
|
||||
matched_monitors = []
|
||||
|
||||
if resume_all:
|
||||
# Get all monitors and filter for inactive (paused) ones
|
||||
all_monitors = self.client.api.get_monitors()
|
||||
paused_monitors = [
|
||||
{"id": m.get("id"), "name": m.get("name")}
|
||||
for m in all_monitors
|
||||
if not m.get("active", True)
|
||||
monitor_ids = [
|
||||
m.get("id") for m in all_monitors if not m.get("active", True)
|
||||
]
|
||||
matched_monitors.extend(paused_monitors)
|
||||
else:
|
||||
# Find monitors by patterns if specified
|
||||
if monitor_patterns:
|
||||
pattern_monitors = self.client.find_monitors_by_pattern(
|
||||
monitor_patterns
|
||||
)
|
||||
matched_monitors.extend(pattern_monitors)
|
||||
|
||||
# Find monitors by groups if specified
|
||||
if group_patterns:
|
||||
group_monitors = self.client.get_monitors_in_groups(group_patterns)
|
||||
# Convert to the same format as find_monitors_by_pattern
|
||||
group_monitor_objs = [
|
||||
{"id": m.get("id"), "name": m.get("name")}
|
||||
for m in group_monitors
|
||||
]
|
||||
matched_monitors.extend(group_monitor_objs)
|
||||
|
||||
# Remove duplicates while preserving order
|
||||
seen = set()
|
||||
unique_monitors = []
|
||||
for monitor in matched_monitors:
|
||||
if monitor["id"] not in seen:
|
||||
seen.add(monitor["id"])
|
||||
unique_monitors.append(monitor)
|
||||
|
||||
matched_monitors = unique_monitors
|
||||
|
||||
if not matched_monitors:
|
||||
if resume_all:
|
||||
if not monitor_ids:
|
||||
print("No paused monitors found to resume")
|
||||
else:
|
||||
print(
|
||||
"Error: No monitors found matching the specified patterns or groups"
|
||||
)
|
||||
return
|
||||
|
||||
if resume_all:
|
||||
return
|
||||
matched_monitors = [
|
||||
{
|
||||
"id": mid,
|
||||
"name": next(
|
||||
(
|
||||
m.get("name", f"Monitor {mid}")
|
||||
for m in all_monitors
|
||||
if m.get("id") == mid
|
||||
),
|
||||
f"Monitor {mid}",
|
||||
),
|
||||
}
|
||||
for mid in monitor_ids
|
||||
]
|
||||
print(f"Found {len(matched_monitors)} paused monitors to resume:")
|
||||
else:
|
||||
matched_monitors = self.client.find_and_get_monitors(
|
||||
monitor_patterns, group_patterns
|
||||
)
|
||||
if not matched_monitors:
|
||||
return
|
||||
print(f"Found {len(matched_monitors)} matching monitors to resume:")
|
||||
for monitor in matched_monitors:
|
||||
print(f" - {monitor['name']} (ID: {monitor['id']})")
|
||||
|
||||
Reference in New Issue
Block a user