rag-manager/ingest_pipeline/cli/tui/screens/dialogs.py

"""Dialog screens for confirmations and user interactions."""

from typing import TYPE_CHECKING

from textual.app import ComposeResult
from textual.binding import Binding
from textual.containers import Container, Horizontal
from textual.screen import Screen
from textual.widgets import Button, Footer, Header, LoadingIndicator, Static
from typing_extensions import override

from ..models import CollectionInfo

if TYPE_CHECKING:
    from .dashboard import CollectionOverviewScreen
    from .documents import DocumentManagementScreen


class ConfirmDeleteScreen(Screen[None]):
    """Screen for confirming collection deletion."""

    collection: CollectionInfo
    parent_screen: "CollectionOverviewScreen"

    BINDINGS = [
        Binding("escape", "app.pop_screen", "Cancel"),
        Binding("y", "confirm_delete", "Yes"),
        Binding("n", "app.pop_screen", "No"),
        Binding("enter", "confirm_delete", "Confirm"),
    ]

    def __init__(self, collection: CollectionInfo, parent_screen: "CollectionOverviewScreen"):
        super().__init__()
        self.collection = collection
        self.parent_screen = parent_screen

    @override
    def compose(self) -> ComposeResult:
        yield Header()
        yield Container(
            Static("⚠️  Confirm Deletion", classes="title warning"),
            Static(f"Are you sure you want to delete collection '{self.collection['name']}'?"),
            Static(f"Backend: {self.collection['backend']}"),
            Static(f"Documents: {self.collection['count']:,}"),
            Static("This action cannot be undone!", classes="warning"),
            Static("Press Y to confirm, N or Escape to cancel", classes="subtitle"),
            Horizontal(
                Button("✅ Yes, Delete (Y)", id="yes_btn", variant="error"),
                Button("❌ Cancel (N)", id="no_btn", variant="default"),
                classes="action_buttons",
            ),
            classes="main_container center",
        )
        yield Footer()

    def on_mount(self) -> None:
        """Initialize the screen with focus on cancel button for safety."""
        self.query_one("#no_btn").focus()

    def on_button_pressed(self, event: Button.Pressed) -> None:
        """Handle button presses."""
        if event.button.id == "yes_btn":
            self.action_confirm_delete()
        elif event.button.id == "no_btn":
            self.app.pop_screen()

    def action_confirm_delete(self) -> None:
        """Confirm deletion."""
        self.run_worker(self.delete_collection())

    async def delete_collection(self) -> None:
        """Delete the collection."""
        try:
            if self.collection["type"] == "weaviate" and self.parent_screen.weaviate:
                # Delete Weaviate collection
                if self.parent_screen.weaviate and self.parent_screen.weaviate.client:
                    self.parent_screen.weaviate.client.collections.delete(self.collection["name"])
                self.notify(
                    f"Deleted Weaviate collection: {self.collection['name']}",
                    severity="information",
                )
            elif self.collection["type"] == "openwebui" and self.parent_screen.openwebui:
                # Delete OpenWebUI knowledge base
                response = await self.parent_screen.openwebui.client.delete(
                    f"/api/v1/knowledge/{self.collection['name']}"
                )
                response.raise_for_status()
                self.notify(
                    f"Deleted OpenWebUI collection: {self.collection['name']}",
                    severity="information",
                )

            # Refresh parent screen
            self.parent_screen.refresh_collections()  # Don't await, let it run as a worker
            self.app.pop_screen()

        except Exception as e:
            self.notify(f"Failed to delete collection: {e}", severity="error")


class ConfirmDocumentDeleteScreen(Screen[None]):
    """Screen for confirming document deletion."""

    doc_ids: list[str]
    collection: CollectionInfo
    parent_screen: "DocumentManagementScreen"

    BINDINGS = [
        Binding("escape", "app.pop_screen", "Cancel"),
        Binding("y", "confirm_delete", "Yes"),
        Binding("n", "app.pop_screen", "No"),
        Binding("enter", "confirm_delete", "Confirm"),
    ]

    def __init__(
        self,
        doc_ids: list[str],
        collection: CollectionInfo,
        parent_screen: "DocumentManagementScreen",
    ):
        super().__init__()
        self.doc_ids = doc_ids
        self.collection = collection
        self.parent_screen = parent_screen

    @override
    def compose(self) -> ComposeResult:
        yield Header()
        yield Container(
            Static("⚠️  Confirm Document Deletion", classes="title warning"),
            Static(
                f"Are you sure you want to delete {len(self.doc_ids)} documents from '{self.collection['name']}'?"
            ),
            Static("This action cannot be undone!", classes="warning"),
            Static("Press Y to confirm, N or Escape to cancel", classes="subtitle"),
            Horizontal(
                Button("✅ Yes, Delete (Y)", id="yes_btn", variant="error"),
                Button("❌ Cancel (N)", id="no_btn", variant="default"),
                classes="action_buttons",
            ),
            LoadingIndicator(id="loading"),
            classes="main_container center",
        )
        yield Footer()

    def on_mount(self) -> None:
        """Initialize the screen with focus on cancel button for safety."""
        self.query_one("#loading").display = False
        self.query_one("#no_btn").focus()

    def on_button_pressed(self, event: Button.Pressed) -> None:
        """Handle button presses."""
        if event.button.id == "yes_btn":
            self.action_confirm_delete()
        elif event.button.id == "no_btn":
            self.app.pop_screen()

    def action_confirm_delete(self) -> None:
        """Confirm deletion."""
        self.run_worker(self.delete_documents())

    async def delete_documents(self) -> None:
        """Delete the selected documents."""
        loading = self.query_one("#loading")
        loading.display = True

        try:
            if self.parent_screen.weaviate:
                # Delete documents
                results = await self.parent_screen.weaviate.delete_documents(self.doc_ids)

                # Count successful deletions
                successful = sum(1 for success in results.values() if success)
                failed = len(results) - successful

                if successful > 0:
                    self.notify(f"Deleted {successful} documents", severity="information")
                if failed > 0:
                    self.notify(f"Failed to delete {failed} documents", severity="error")

                # Clear selection and refresh parent screen
                self.parent_screen.selected_docs.clear()
                await self.parent_screen.load_documents()
                self.app.pop_screen()

        except Exception as e:
            self.notify(f"Failed to delete documents: {e}", severity="error")
        finally:
            loading.display = False