import os import sys import json import asyncio from datetime import datetime import msal import aiohttp from rich.panel import Panel from rich import print as rprint from textual.app import App, ComposeResult from textual.binding import Binding from textual.containers import Container, Horizontal, Vertical from textual.widgets import ( Header, Footer, Static, Label, DataTable, Button, ListView, ListItem, LoadingIndicator, OptionList ) from textual.reactive import reactive from textual.worker import Worker, get_current_worker from textual import work from textual.widgets.option_list import Option # Import our DocumentViewerScreen sys.path.append(os.path.join(os.path.dirname(__file__), "maildir_gtd")) from maildir_gtd.screens.DocumentViewer import DocumentViewerScreen class OneDriveTUI(App): """A Textual app for OneDrive integration with MSAL authentication.""" CSS_PATH = "drive_view_tui.tcss" # Reactive variables is_authenticated = reactive(False) selected_drive_id = reactive("") drive_name = reactive("") current_view = reactive("Following") # Track current view: "Following" or "Root" # App bindings BINDINGS = [ Binding("q", "quit", "Quit"), Binding("r", "refresh", "Refresh"), Binding("f", "toggle_follow", "Toggle Follow"), Binding("o", "open_url", "Open URL"), Binding("enter", "open_url", "Open URL"), Binding("v", "view_document", "View Document"), Binding("tab", "next_view", "Switch View"), ] def __init__(self): super().__init__() self.access_token = None self.drives = [] self.followed_items = [] self.current_items = [] # Store currently displayed items self.msal_app = None self.cache = msal.SerializableTokenCache() # Read Azure app credentials from environment variables self.client_id = os.getenv("AZURE_CLIENT_ID") self.tenant_id = os.getenv("AZURE_TENANT_ID") self.scopes = ["https://graph.microsoft.com/Files.ReadWrite.All"] self.cache_file = "token_cache.bin" def compose(self) -> ComposeResult: """Create child widgets for the app.""" yield Header(show_clock=True) with Container(id="main_container"): yield LoadingIndicator(id="loading") yield Label("Authenticating with Microsoft Graph API...", id="status_label") with Container(id="auth_container"): yield Label("", id="auth_message") yield Button("Login", id="login_button", variant="primary") with Container(id="content_container", classes="hide"): with Horizontal(): with Vertical(id="navigation_container"): yield OptionList( Option("Following", id="following"), Option("Root", id="root"), id="view_options" ) with Vertical(id="items_container"): yield DataTable(id="items_table") yield Label("No items found", id="no_items_label", classes="hide") yield Footer() async def on_mount(self) -> None: """Initialize the app when mounted.""" self.query_one("#login_button").styles.width = "20" self.query_one("#view_options").border_title = "My Files" # Initialize the table table = self.query_one("#items_table") table.add_columns("Type", "Name", "Last Modified", "Size", "Web URL") # Load cached token if available if os.path.exists(self.cache_file): with open(self.cache_file, "r") as f: self.cache.deserialize(f.read()) # Initialize MSAL app self.initialize_msal() # Try silent authentication first self.authenticate_silent() def initialize_msal(self) -> None: """Initialize the MSAL application.""" if not self.client_id or not self.tenant_id: self.notify( "Please set AZURE_CLIENT_ID and AZURE_TENANT_ID environment variables.", severity="error", timeout=10, ) return authority = f"https://login.microsoftonline.com/{self.tenant_id}" self.msal_app = msal.PublicClientApplication( self.client_id, authority=authority, token_cache=self.cache ) def authenticate_silent(self) -> None: """Try silent authentication first.""" if not self.msal_app: return accounts = self.msal_app.get_accounts() if accounts: self.query_one("#status_label").update("Trying silent authentication...") self.get_token_silent(accounts[0]) else: self.query_one("#status_label").update("Please log in to continue.") self.query_one("#auth_container").remove_class("hide") self.query_one("#loading").remove() @work async def get_token_silent(self, account): """Get token silently.""" token_response = self.msal_app.acquire_token_silent(self.scopes, account=account) if token_response and "access_token" in token_response: self.access_token = token_response["access_token"] self.is_authenticated = True self.load_initial_data() else: self.query_one("#status_label").update("Silent authentication failed. Please log in.") self.query_one("#auth_container").remove_class("hide") self.query_one("#loading").remove() async def on_button_pressed(self, event: Button.Pressed) -> None: """Handle button presses.""" if event.button.id == "login_button": self.initiate_device_flow() @work async def initiate_device_flow(self): """Initiate the MSAL device code flow.""" self.query_one("#loading").styles.display = "block" self.query_one("#status_label").update("Initiating device code flow...") # Initiate device flow flow = self.msal_app.initiate_device_flow(scopes=self.scopes) if "user_code" not in flow: self.notify("Failed to create device flow", severity="error") return # Display the device code message self.query_one("#auth_message").update(flow["message"]) # Wait for the user to authenticate token_response = self.msal_app.acquire_token_by_device_flow(flow) if "access_token" not in token_response: self.notify("Failed to acquire token", severity="error") return # Save token to cache with open(self.cache_file, "w") as f: f.write(self.cache.serialize()) self.access_token = token_response["access_token"] self.is_authenticated = True # Proceed with loading drives and followed items self.load_initial_data() @work async def load_initial_data(self): """Load initial data after authentication.""" self.query_one("#status_label").update("Loading drives...") # Load drives first if not self.access_token: return headers = {"Authorization": f"Bearer {self.access_token}"} try: async with aiohttp.ClientSession() as session: async with session.get( "https://graph.microsoft.com/v1.0/me/drives", headers=headers ) as response: if response.status != 200: self.notify(f"Failed to load drives: {response.status}", severity="error") return drives_data = await response.json() self.drives = drives_data.get("value", []) except Exception as e: self.notify(f"Error loading drives: {str(e)}", severity="error") # Hide auth container and show content container self.query_one("#auth_container").add_class("hide") self.query_one("#content_container").remove_class("hide") self.query_one("#loading").remove() # Find and select the OneDrive drive for drive in self.drives: if drive.get("name") == "OneDrive": self.selected_drive_id = drive.get("id") self.drive_name = drive.get("name") break # Set Following as default view option_list = self.query_one("#view_options") option_list.highlighted = 0 # Select "Following" option # If we have a selected drive, load followed items if self.selected_drive_id: self.load_followed_items() async def on_option_list_option_selected(self, event: OptionList.OptionSelected) -> None: """Handle option list selection.""" selected_option = event.option.id if selected_option == "following": self.current_view = "Following" if self.selected_drive_id: self.load_followed_items() elif selected_option == "root": self.current_view = "Root" if self.selected_drive_id: self.load_root_items() @work async def load_root_items(self): """Load root items from the selected drive.""" if not self.access_token or not self.selected_drive_id: return self.query_one("#status_label").update("Loading root items...") headers = {"Authorization": f"Bearer {self.access_token}"} try: url = f"https://graph.microsoft.com/v1.0/me/drive/root/children" async with aiohttp.ClientSession() as session: async with session.get(url, headers=headers) as response: if response.status != 200: self.notify(f"Failed to load root items: {response.status}", severity="error") return items_data = await response.json() self.current_items = items_data.get("value", []) # Update the table with the root items self.update_items_table(self.current_items, is_root_view=True) except Exception as e: self.notify(f"Error loading root items: {str(e)}", severity="error") self.query_one("#status_label").update("Ready") @work async def load_followed_items(self): """Load followed items from the selected drive.""" if not self.access_token or not self.selected_drive_id: return self.query_one("#status_label").update("Loading followed items...") headers = {"Authorization": f"Bearer {self.access_token}"} try: url = f"https://graph.microsoft.com/v1.0/me/drive/following" async with aiohttp.ClientSession() as session: async with session.get(url, headers=headers) as response: if response.status != 200: self.notify(f"Failed to load followed items: {response.status}", severity="error") return items_data = await response.json() self.followed_items = items_data.get("value", []) self.current_items = self.followed_items # Update the table with the followed items self.update_items_table(self.followed_items) except Exception as e: self.notify(f"Error loading followed items: {str(e)}", severity="error") self.query_one("#status_label").update("Ready") def update_items_table(self, items, is_root_view=False): """Update the table with the given items.""" table = self.query_one("#items_table") table.clear() if not items: self.query_one("#no_items_label").remove_class("hide") return self.query_one("#no_items_label").add_class("hide") for item in items: name = item.get("name", "Unknown") is_folder = bool(item.get("folder")) # Add folder icon if it's a folder and we're in root view item_type = "📁" if is_folder else "📄" # Format the last modified date last_modified = item.get("lastModifiedDateTime", "") if last_modified: try: date_obj = datetime.fromisoformat(last_modified.replace('Z', '+00:00')) last_modified = date_obj.strftime("%Y-%m-%d %H:%M") except: pass # Format the size size = item.get("size", 0) if size: if size < 1024: size_str = f"{size} B" elif size < 1024 * 1024: size_str = f"{size / 1024:.1f} KB" elif size < 1024 * 1024 * 1024: size_str = f"{size / (1024 * 1024):.1f} MB" else: size_str = f"{size / (1024 * 1024 * 1024):.1f} GB" else: size_str = "N/A" web_url = item.get("webUrl", "") table.add_row(item_type, name, last_modified, size_str, web_url) async def action_next_view(self) -> None: """Switch to the next view (Following/Root).""" option_list = self.query_one("#view_options") if self.current_view == "Following": option_list.highlighted = 1 # Switch to Root self.current_view = "Root" self.load_root_items() else: option_list.highlighted = 0 # Switch to Following self.current_view = "Following" self.load_followed_items() self.notify(f"Switched to {self.current_view} view") async def action_refresh(self) -> None: """Refresh the data.""" if self.is_authenticated and self.selected_drive_id: if self.current_view == "Following": self.load_followed_items() elif self.current_view == "Root": self.load_root_items() self.notify("Refreshed items") async def action_toggle_follow(self) -> None: """Toggle follow status for selected item.""" # This would be implemented to follow/unfollow the selected item # Currently just a placeholder for the key binding self.notify("Toggle follow functionality not implemented yet") async def action_open_url(self) -> None: """Open the web URL of the selected item.""" table = self.query_one("#items_table") if table.cursor_row is not None: selected_row = table.get_row_at(table.cursor_row) if selected_row and len(selected_row) > 4: web_url = selected_row[4] if web_url: self.notify(f"Opening URL: {web_url}") # Use Textual's built-in open_url method self.app.open_url(web_url) async def action_view_document(self) -> None: """View the selected document using the DocumentViewerScreen.""" table = self.query_one("#items_table") if table.cursor_row is None: return # Get the name of the selected item selected_row = table.get_row_at(table.cursor_row) if not selected_row: return selected_name = selected_row[1] # Find the item in our list to get its ID selected_item = None for item in self.current_items: if item.get("name") == selected_name: selected_item = item break if not selected_item: self.notify("Could not find the selected item details", severity="error") return # Check if it's a folder - cannot view folders if selected_item.get("folder"): self.notify("Cannot preview folders. Use 'Open URL' to view in browser.", severity="warning") return # Get the item ID item_id = selected_item.get("id") if not item_id: self.notify("Item ID not found", severity="error") return # Open the document viewer screen viewer = DocumentViewerScreen(item_id, selected_name, self.access_token, selected_item.get("parentReference").get("driveId")) self.push_screen(viewer) async def action_quit(self) -> None: """Quit the application.""" self.exit() if __name__ == "__main__": app = OneDriveTUI() app.run()