635 lines
26 KiB
Python
635 lines
26 KiB
Python
"""Main dashboard screen with collections overview."""
|
|
|
|
import logging
|
|
from typing import TYPE_CHECKING, Final
|
|
|
|
from textual import work
|
|
from textual.app import ComposeResult
|
|
from textual.binding import Binding
|
|
from textual.containers import Container, Grid, Horizontal
|
|
from textual.css.query import NoMatches
|
|
from textual.reactive import reactive, var
|
|
from textual.screen import Screen
|
|
from textual.widgets import (
|
|
Button,
|
|
Footer,
|
|
Header,
|
|
LoadingIndicator,
|
|
Rule,
|
|
Static,
|
|
TabbedContent,
|
|
TabPane,
|
|
)
|
|
from typing_extensions import override
|
|
|
|
from ....core.models import StorageBackend
|
|
from ....storage.base import BaseStorage
|
|
from ....storage.openwebui import OpenWebUIStorage
|
|
from ....storage.weaviate import WeaviateStorage
|
|
from ..models import CollectionInfo
|
|
from ..utils.storage_manager import StorageManager
|
|
from ..widgets import EnhancedDataTable, MetricsCard, StatusIndicator
|
|
|
|
if TYPE_CHECKING:
|
|
from ....storage.r2r.storage import R2RStorage
|
|
else: # pragma: no cover - optional dependency fallback
|
|
R2RStorage = BaseStorage
|
|
|
|
|
|
LOGGER: Final[logging.Logger] = logging.getLogger(__name__)
|
|
|
|
|
|
class CollectionOverviewScreen(Screen[None]):
|
|
"""Enhanced dashboard with modern design and metrics."""
|
|
|
|
total_documents: int = 0
|
|
total_collections: int = 0
|
|
active_backends: int = 0
|
|
|
|
BINDINGS = [
|
|
Binding("q", "quit", "Quit"),
|
|
Binding("r", "refresh", "Refresh"),
|
|
Binding("i", "ingest", "Ingest"),
|
|
Binding("m", "manage", "Manage"),
|
|
Binding("s", "search", "Search"),
|
|
Binding("ctrl+d", "delete", "Delete"),
|
|
Binding("ctrl+1", "tab_dashboard", "Dashboard"),
|
|
Binding("ctrl+2", "tab_collections", "Collections"),
|
|
Binding("ctrl+3", "tab_analytics", "Analytics"),
|
|
Binding("tab", "next_tab", "Next Tab"),
|
|
Binding("shift+tab", "prev_tab", "Prev Tab"),
|
|
Binding("f1", "help", "Help"),
|
|
]
|
|
|
|
collections: var[list[CollectionInfo]] = var([])
|
|
is_loading: var[bool] = var(False)
|
|
selected_collection: reactive[CollectionInfo | None] = reactive(None)
|
|
storage_manager: StorageManager
|
|
weaviate: WeaviateStorage | None
|
|
openwebui: OpenWebUIStorage | None
|
|
r2r: R2RStorage | BaseStorage | None
|
|
|
|
def __init__(
|
|
self,
|
|
storage_manager: StorageManager,
|
|
weaviate: WeaviateStorage | None,
|
|
openwebui: OpenWebUIStorage | None,
|
|
r2r: R2RStorage | BaseStorage | None,
|
|
) -> None:
|
|
super().__init__()
|
|
self.storage_manager = storage_manager
|
|
self.weaviate = weaviate
|
|
self.openwebui = openwebui
|
|
self.r2r = r2r
|
|
self.total_documents = 0
|
|
self.total_collections = 0
|
|
self.active_backends = 0
|
|
|
|
@override
|
|
def compose(self) -> ComposeResult:
|
|
yield Header(show_clock=True)
|
|
|
|
with TabbedContent():
|
|
# Dashboard Tab
|
|
with TabPane("Dashboard", id="dashboard"):
|
|
yield Container(
|
|
Static("🚀 Collection Management System", classes="title"),
|
|
Static("Modern document ingestion and management platform", classes="subtitle"),
|
|
Rule(line_style="heavy"),
|
|
# Metrics Grid
|
|
Container(
|
|
Grid(
|
|
MetricsCard(
|
|
"Collections", str(self.total_collections), "Active collections"
|
|
),
|
|
MetricsCard("Documents", str(self.total_documents), "Total indexed"),
|
|
MetricsCard(
|
|
"Backends", str(self.active_backends), "Connected services"
|
|
),
|
|
MetricsCard("Status", "Online", "System health"),
|
|
classes="responsive-grid metrics-grid",
|
|
),
|
|
classes="center",
|
|
),
|
|
Rule(line_style="dashed"),
|
|
# Quick Actions
|
|
Container(
|
|
Static("⚡ Quick Actions", classes="section-title"),
|
|
Horizontal(
|
|
Button("🔄 Refresh Data", id="quick_refresh", variant="primary"),
|
|
Button("📥 New Ingestion", id="quick_ingest", variant="success"),
|
|
Button("🔍 Search All", id="quick_search", variant="default"),
|
|
Button("⚙️ Settings", id="quick_settings", variant="default"),
|
|
classes="action_buttons",
|
|
),
|
|
classes="card",
|
|
),
|
|
# Recent Activity
|
|
Container(
|
|
Static("📊 Recent Activity", classes="section-title"),
|
|
Static(
|
|
"Loading recent activity...", id="activity_feed", classes="status-text"
|
|
),
|
|
classes="card",
|
|
),
|
|
classes="main_container",
|
|
)
|
|
|
|
# Collections Tab
|
|
with TabPane("Collections", id="collections"):
|
|
yield Container(
|
|
Static("📚 Collection Overview", classes="title"),
|
|
# Collection controls
|
|
Horizontal(
|
|
Button("🔄 Refresh", id="refresh_btn", variant="primary"),
|
|
Button("📥 Ingest", id="ingest_btn", variant="success"),
|
|
Button("🔧 Manage", id="manage_btn", variant="warning"),
|
|
Button("🗑️ Delete", id="delete_btn", variant="error"),
|
|
Button("🔍 Search", id="search_btn", variant="default"),
|
|
classes="button_bar",
|
|
),
|
|
# Collection table with enhanced navigation
|
|
EnhancedDataTable(id="collections_table", classes="enhanced-table"),
|
|
# Status bar
|
|
Container(
|
|
Static("Ready", id="status_text", classes="status-text"),
|
|
StatusIndicator("Ready", id="connection_status"),
|
|
classes="status-bar",
|
|
),
|
|
LoadingIndicator(id="loading", classes="pulse"),
|
|
classes="main_container",
|
|
)
|
|
|
|
# Analytics Tab
|
|
with TabPane("Analytics", id="analytics"):
|
|
yield Container(
|
|
Static("📈 Analytics & Insights", classes="title"),
|
|
# Analytics content
|
|
Container(
|
|
Static("🚧 Analytics Dashboard", classes="section-title"),
|
|
Static("Advanced analytics and insights coming soon!", classes="subtitle"),
|
|
# Placeholder charts area
|
|
Container(
|
|
Static("📊 Document Distribution", classes="chart-title"),
|
|
Static(
|
|
"Chart placeholder - integrate with visualization library",
|
|
classes="chart-placeholder",
|
|
),
|
|
classes="card",
|
|
),
|
|
Container(
|
|
Static("⏱️ Ingestion Timeline", classes="chart-title"),
|
|
Static("Timeline chart placeholder", classes="chart-placeholder"),
|
|
classes="card",
|
|
),
|
|
classes="analytics-grid",
|
|
),
|
|
classes="main_container",
|
|
)
|
|
|
|
yield Footer()
|
|
|
|
async def on_mount(self) -> None:
|
|
"""Initialize the screen with enhanced loading."""
|
|
self.query_one("#loading").display = False
|
|
self.update_metrics()
|
|
self.refresh_collections() # Don't await, let it run as a worker
|
|
|
|
def update_metrics(self) -> None:
|
|
"""Update dashboard metrics with enhanced calculations."""
|
|
self._calculate_metrics()
|
|
self._update_metrics_cards()
|
|
self._update_activity_feed()
|
|
|
|
def _calculate_metrics(self) -> None:
|
|
"""Calculate basic metrics from collections."""
|
|
self.total_collections = len(self.collections)
|
|
self.total_documents = sum(col["count"] for col in self.collections)
|
|
# Calculate active backends from storage manager if individual storages are None
|
|
if self.weaviate is None and self.openwebui is None and self.r2r is None:
|
|
self.active_backends = len(self.storage_manager.get_available_backends())
|
|
else:
|
|
self.active_backends = sum([bool(self.weaviate), bool(self.openwebui), bool(self.r2r)])
|
|
|
|
def _update_metrics_cards(self) -> None:
|
|
"""Update the metrics cards display."""
|
|
try:
|
|
dashboard_tab = self.query_one("#dashboard")
|
|
metrics_cards_query = dashboard_tab.query(MetricsCard)
|
|
if len(metrics_cards_query) >= 4:
|
|
metrics_cards = list(metrics_cards_query)
|
|
self._update_card_values(metrics_cards)
|
|
self._update_status_card(metrics_cards[3])
|
|
except NoMatches:
|
|
return
|
|
except Exception as exc:
|
|
LOGGER.exception("Failed to update dashboard metrics", exc_info=exc)
|
|
|
|
def _update_card_values(self, metrics_cards: list[MetricsCard]) -> None:
|
|
"""Update individual metric card values."""
|
|
metrics_cards[0].query_one(".metrics-value", Static).update(f"{self.total_collections:,}")
|
|
metrics_cards[1].query_one(".metrics-value", Static).update(f"{self.total_documents:,}")
|
|
metrics_cards[2].query_one(".metrics-value", Static).update(str(self.active_backends))
|
|
|
|
def _update_status_card(self, status_card: MetricsCard) -> None:
|
|
"""Update the system status card."""
|
|
if self.active_backends > 0 and self.total_collections > 0:
|
|
status_text, status_class = "🟢 Healthy", "status-active"
|
|
elif self.active_backends > 0:
|
|
status_text, status_class = "🟡 Ready", "status-warning"
|
|
else:
|
|
status_text, status_class = "🔴 Offline", "status-error"
|
|
|
|
status_card.query_one(".metrics-value", Static).update(status_text)
|
|
status_card.add_class(status_class)
|
|
|
|
def _update_activity_feed(self) -> None:
|
|
"""Update the activity feed with collection data."""
|
|
try:
|
|
dashboard_tab = self.query_one("#dashboard")
|
|
activity_feed = dashboard_tab.query_one("#activity_feed", Static)
|
|
activity_text = self._generate_activity_text()
|
|
activity_feed.update(activity_text)
|
|
except NoMatches:
|
|
return
|
|
except Exception as exc:
|
|
LOGGER.exception("Failed to update dashboard activity feed", exc_info=exc)
|
|
|
|
def _generate_activity_text(self) -> str:
|
|
"""Generate activity feed text from collections."""
|
|
if not self.collections:
|
|
return "🚀 No collections found. Start by creating your first ingestion!\n💡 Press 'I' to begin or use the Quick Actions above."
|
|
|
|
recent_activity = [self._format_collection_item(col) for col in self.collections[:3]]
|
|
activity_text = "\n".join(recent_activity)
|
|
|
|
if len(self.collections) > 3:
|
|
total_docs = sum(c["count"] for c in self.collections)
|
|
activity_text += (
|
|
f"\n📊 Total: {len(self.collections)} collections with {total_docs:,} documents"
|
|
)
|
|
|
|
return activity_text
|
|
|
|
def _format_collection_item(self, col: CollectionInfo) -> str:
|
|
"""Format a single collection item for the activity feed."""
|
|
content_type = self._get_content_type_icon(col["name"])
|
|
size_mb = col["size_mb"]
|
|
backend_info = col["backend"]
|
|
|
|
# Check if this represents a multi-backend ingestion result
|
|
if isinstance(backend_info, list):
|
|
if len(backend_info) > 1:
|
|
# Ensure all elements are strings for safe joining
|
|
backend_strings = [str(b) for b in backend_info if b is not None]
|
|
backend_list = " + ".join(backend_strings) if backend_strings else "unknown"
|
|
return f"{content_type} {col['name']}: {col['count']:,} docs ({size_mb:.1f} MB) → {backend_list}"
|
|
elif len(backend_info) == 1:
|
|
backend_name = str(backend_info[0]) if backend_info[0] is not None else "unknown"
|
|
return f"{content_type} {col['name']}: {col['count']:,} docs ({size_mb:.1f} MB) - {backend_name}"
|
|
else:
|
|
return f"{content_type} {col['name']}: {col['count']:,} docs ({size_mb:.1f} MB) - unknown"
|
|
else:
|
|
backend_display = str(backend_info) if backend_info is not None else "unknown"
|
|
return f"{content_type} {col['name']}: {col['count']:,} docs ({size_mb:.1f} MB) - {backend_display}"
|
|
|
|
def _get_content_type_icon(self, name: str) -> str:
|
|
"""Get appropriate icon for collection content type."""
|
|
name_lower = name.lower()
|
|
if "web" in name_lower:
|
|
return "🌐"
|
|
elif "doc" in name_lower:
|
|
return "📖"
|
|
elif "repo" in name_lower:
|
|
return "📦"
|
|
return "📄"
|
|
|
|
@work(exclusive=True)
|
|
async def refresh_collections(self) -> None:
|
|
"""Refresh collection data with enhanced multi-backend loading feedback."""
|
|
self.is_loading = True
|
|
loading_indicator = self.query_one("#loading")
|
|
status_text = self.query_one("#status_text", Static)
|
|
|
|
loading_indicator.display = True
|
|
status_text.update("🔄 Refreshing collections...")
|
|
|
|
try:
|
|
# Use storage manager for unified backend handling
|
|
if not self.storage_manager.is_initialized:
|
|
status_text.update("🔗 Initializing storage backends...")
|
|
backend_results = await self.storage_manager.initialize_all_backends()
|
|
|
|
# Report per-backend initialization status
|
|
success_count = sum(backend_results.values())
|
|
total_count = len(backend_results)
|
|
status_text.update(f"✅ Initialized {success_count}/{total_count} backends")
|
|
|
|
# Get collections from all backends via storage manager
|
|
status_text.update("📚 Loading collections from all backends...")
|
|
collections = await self.storage_manager.get_all_collections()
|
|
|
|
# Update metrics calculation for multi-backend support
|
|
self.active_backends = len(self.storage_manager.get_available_backends())
|
|
|
|
self.collections = collections
|
|
await self.update_collections_table()
|
|
self.update_metrics()
|
|
|
|
# Enhanced status reporting for multi-backend
|
|
backend_names = ", ".join(
|
|
backend.value for backend in self.storage_manager.get_available_backends()
|
|
)
|
|
status_text.update(f"✨ Ready - {len(collections)} collections from {backend_names}")
|
|
|
|
# Update connection status with multi-backend awareness
|
|
connection_status = self.query_one("#connection_status", StatusIndicator)
|
|
if collections and self.active_backends > 0:
|
|
connection_status.update_status(f"✓ {self.active_backends} Active")
|
|
else:
|
|
connection_status.update_status("No Data")
|
|
|
|
except Exception as e:
|
|
status_text.update(f"❌ Error: {e}")
|
|
self.notify(f"Failed to refresh: {e}", severity="error", markup=False)
|
|
finally:
|
|
self.is_loading = False
|
|
loading_indicator.display = False
|
|
|
|
async def update_collections_table(self) -> None:
|
|
"""Update the collections table with enhanced formatting."""
|
|
try:
|
|
table = self.query_one("#collections_table", EnhancedDataTable)
|
|
table.clear(columns=True)
|
|
|
|
# Add enhanced columns with more metadata
|
|
table.add_columns("Collection", "Backend", "Documents", "Size", "Type", "Status", "Updated")
|
|
|
|
# Add rows with enhanced formatting
|
|
for collection in self.collections:
|
|
try:
|
|
# Format size
|
|
size_str = f"{collection['size_mb']:.1f} MB"
|
|
if collection["size_mb"] > 1000:
|
|
size_str = f"{collection['size_mb'] / 1000:.1f} GB"
|
|
|
|
# Format document count
|
|
doc_count = f"{collection['count']:,}"
|
|
|
|
# Determine content type based on collection name or other metadata
|
|
content_type = "📄 Mixed"
|
|
if "web" in collection["name"].lower():
|
|
content_type = "🌐 Web"
|
|
elif "doc" in collection["name"].lower():
|
|
content_type = "📖 Docs"
|
|
elif "repo" in collection["name"].lower():
|
|
content_type = "📦 Code"
|
|
|
|
table.add_row(
|
|
collection["name"],
|
|
collection["backend"],
|
|
doc_count,
|
|
size_str,
|
|
content_type,
|
|
collection["status"],
|
|
collection["last_updated"],
|
|
)
|
|
except Exception as e:
|
|
LOGGER.warning(f"Failed to add collection row for {collection.get('name', 'unknown')}: {e}")
|
|
continue
|
|
|
|
if self.collections:
|
|
try:
|
|
table.move_cursor(row=0)
|
|
except Exception as e:
|
|
LOGGER.warning(f"Failed to move table cursor: {e}")
|
|
|
|
self.get_selected_collection()
|
|
except Exception as e:
|
|
LOGGER.exception(f"Failed to update collections table: {e}")
|
|
self.notify(f"Failed to update table: {e}", severity="error", markup=False)
|
|
|
|
def update_search_controls(self, collection: CollectionInfo | None) -> None:
|
|
"""Enable or disable search controls based on backend support."""
|
|
try:
|
|
search_button = self.query_one("#search_btn", Button)
|
|
quick_search_button = self.query_one("#quick_search", Button)
|
|
except Exception:
|
|
return
|
|
|
|
is_weaviate = bool(collection and collection.get("type") == "weaviate")
|
|
search_button.disabled = not is_weaviate
|
|
quick_search_button.disabled = not is_weaviate
|
|
|
|
def get_selected_collection(self) -> CollectionInfo | None:
|
|
"""Get the currently selected collection."""
|
|
table = self.query_one("#collections_table", EnhancedDataTable)
|
|
try:
|
|
row_index = table.cursor_coordinate.row
|
|
except (AttributeError, IndexError):
|
|
self.selected_collection = None
|
|
self.update_search_controls(None)
|
|
return None
|
|
|
|
if 0 <= row_index < len(self.collections):
|
|
collection = self.collections[row_index]
|
|
self.selected_collection = collection
|
|
self.update_search_controls(collection)
|
|
return collection
|
|
|
|
self.selected_collection = None
|
|
self.update_search_controls(None)
|
|
return None
|
|
|
|
# Action methods
|
|
def action_refresh(self) -> None:
|
|
"""Refresh collections."""
|
|
self.refresh_collections()
|
|
|
|
def action_ingest(self) -> None:
|
|
"""Show enhanced ingestion dialog."""
|
|
if selected := self.get_selected_collection():
|
|
from .ingestion import IngestionScreen
|
|
|
|
self.app.push_screen(IngestionScreen(selected, self.storage_manager))
|
|
else:
|
|
self.notify("🔍 Please select a collection first", severity="warning")
|
|
|
|
def action_manage(self) -> None:
|
|
"""Manage documents in selected collection."""
|
|
if selected := self.get_selected_collection():
|
|
if storage_backend := self._get_storage_for_collection(selected):
|
|
from .documents import DocumentManagementScreen
|
|
|
|
self.app.push_screen(DocumentManagementScreen(selected, storage_backend))
|
|
else:
|
|
self.notify(
|
|
"🚧 No storage backend available for this collection", severity="warning"
|
|
)
|
|
else:
|
|
self.notify("🔍 Please select a collection first", severity="warning")
|
|
|
|
def _get_storage_for_collection(self, collection: CollectionInfo) -> BaseStorage | None:
|
|
"""Get the appropriate storage backend for a collection."""
|
|
collection_type = collection.get("type", "")
|
|
|
|
# Map collection types to storage backends (try direct instances first)
|
|
if collection_type == "weaviate" and self.weaviate:
|
|
return self.weaviate
|
|
elif collection_type == "openwebui" and self.openwebui:
|
|
return self.openwebui
|
|
elif collection_type == "r2r" and self.r2r:
|
|
return self.r2r
|
|
|
|
# Fall back to storage manager if direct instances not available
|
|
if collection_type == "weaviate":
|
|
return self.storage_manager.get_backend(StorageBackend.WEAVIATE)
|
|
elif collection_type == "openwebui":
|
|
return self.storage_manager.get_backend(StorageBackend.OPEN_WEBUI)
|
|
elif collection_type == "r2r":
|
|
return self.storage_manager.get_backend(StorageBackend.R2R)
|
|
|
|
# Fall back to checking available backends by backend name
|
|
backend_name = collection.get("backend", "")
|
|
if isinstance(backend_name, str):
|
|
if "weaviate" in backend_name.lower():
|
|
return self.weaviate or self.storage_manager.get_backend(StorageBackend.WEAVIATE)
|
|
elif "openwebui" in backend_name.lower():
|
|
return self.openwebui or self.storage_manager.get_backend(StorageBackend.OPEN_WEBUI)
|
|
elif "r2r" in backend_name.lower():
|
|
return self.r2r or self.storage_manager.get_backend(StorageBackend.R2R)
|
|
|
|
return None
|
|
|
|
def action_search(self) -> None:
|
|
"""Search in selected collection."""
|
|
if selected := self.get_selected_collection():
|
|
if selected["type"] != "weaviate":
|
|
self.notify(
|
|
"🔐 Search is currently available only for Weaviate collections",
|
|
severity="warning",
|
|
)
|
|
return
|
|
from .search import SearchScreen
|
|
|
|
self.app.push_screen(SearchScreen(selected, self.weaviate, self.openwebui))
|
|
else:
|
|
self.notify("🔍 Please select a collection first", severity="warning")
|
|
|
|
def action_delete(self) -> None:
|
|
"""Delete selected collection."""
|
|
if selected := self.get_selected_collection():
|
|
from .dialogs import ConfirmDeleteScreen
|
|
|
|
self.app.push_screen(ConfirmDeleteScreen(selected, self))
|
|
else:
|
|
self.notify("🔍 Please select a collection first", severity="warning")
|
|
|
|
def action_tab_dashboard(self) -> None:
|
|
"""Switch to dashboard tab."""
|
|
tabbed_content: TabbedContent = self.query_one(TabbedContent)
|
|
tabbed_content.active = "dashboard"
|
|
|
|
def action_tab_collections(self) -> None:
|
|
"""Switch to collections tab."""
|
|
tabbed_content: TabbedContent = self.query_one(TabbedContent)
|
|
tabbed_content.active = "collections"
|
|
|
|
def action_tab_analytics(self) -> None:
|
|
"""Switch to analytics tab."""
|
|
tabbed_content: TabbedContent = self.query_one(TabbedContent)
|
|
tabbed_content.active = "analytics"
|
|
|
|
def action_next_tab(self) -> None:
|
|
"""Switch to next tab."""
|
|
tabbed_content: TabbedContent = self.query_one(TabbedContent)
|
|
tab_ids = ["dashboard", "collections", "analytics"]
|
|
current = tabbed_content.active
|
|
try:
|
|
current_index = tab_ids.index(current)
|
|
next_index = (current_index + 1) % len(tab_ids)
|
|
tabbed_content.active = tab_ids[next_index]
|
|
except (ValueError, AttributeError):
|
|
tabbed_content.active = tab_ids[0]
|
|
|
|
def action_prev_tab(self) -> None:
|
|
"""Switch to previous tab."""
|
|
tabbed_content: TabbedContent = self.query_one(TabbedContent)
|
|
tab_ids = ["dashboard", "collections", "analytics"]
|
|
current = tabbed_content.active
|
|
try:
|
|
current_index = tab_ids.index(current)
|
|
prev_index = (current_index - 1) % len(tab_ids)
|
|
tabbed_content.active = tab_ids[prev_index]
|
|
except (ValueError, AttributeError):
|
|
tabbed_content.active = tab_ids[0]
|
|
|
|
def action_help(self) -> None:
|
|
"""Show help screen."""
|
|
from .help import HelpScreen
|
|
|
|
help_md = """
|
|
# 🚀 Modern Collection Management System
|
|
|
|
## Navigation
|
|
- **Tab** / **Shift+Tab**: Switch between tabs
|
|
- **Ctrl+1/2/3**: Direct tab access
|
|
- **Enter**: Activate selected item
|
|
- **Escape**: Go back/cancel
|
|
- **Arrow Keys**: Navigate within tables
|
|
- **Home/End**: Jump to first/last row
|
|
- **Page Up/Down**: Scroll by page
|
|
|
|
## Collections
|
|
- **R**: Refresh collections
|
|
- **I**: Start ingestion
|
|
- **M**: Manage documents
|
|
- **S**: Search collection
|
|
- **Ctrl+D**: Delete collection
|
|
|
|
## Table Navigation
|
|
- **Up/Down** or **J/K**: Navigate rows
|
|
- **Space**: Toggle selection
|
|
- **Ctrl+A**: Select all
|
|
- **Ctrl+Shift+A**: Clear selection
|
|
|
|
## General
|
|
- **Q** / **Ctrl+C**: Quit application
|
|
- **F1**: Show this help
|
|
|
|
Enjoy the enhanced interface! 🎉
|
|
"""
|
|
self.app.push_screen(HelpScreen(help_md))
|
|
|
|
def on_button_pressed(self, event: Button.Pressed) -> None:
|
|
"""Handle button presses with enhanced feedback."""
|
|
button_id = event.button.id
|
|
|
|
# Add visual feedback
|
|
event.button.add_class("pressed")
|
|
self.call_later(self.remove_pressed_class, event.button)
|
|
|
|
if getattr(event.button, "disabled", False):
|
|
self.notify(
|
|
"🔐 Search is currently limited to Weaviate collections",
|
|
severity="warning",
|
|
)
|
|
return
|
|
|
|
if button_id in ["refresh_btn", "quick_refresh"]:
|
|
self.action_refresh()
|
|
elif button_id in ["ingest_btn", "quick_ingest"]:
|
|
self.action_ingest()
|
|
elif button_id == "manage_btn":
|
|
self.action_manage()
|
|
elif button_id == "delete_btn":
|
|
self.action_delete()
|
|
elif button_id in ["search_btn", "quick_search"]:
|
|
self.action_search()
|
|
elif button_id == "quick_settings":
|
|
self.notify("⚙️ Settings panel coming soon!", severity="information")
|
|
|
|
def remove_pressed_class(self, button: Button) -> None:
|
|
"""Remove pressed visual feedback class."""
|
|
button.remove_class("pressed")
|