🔍 Code Extractor

Panel showing pending tasks for the current user

Purpose

Panel showing pending tasks for the current user

Source Code

class UserTasksPanel:
    """Panel showing pending tasks for the current user"""
    
    def __init__(self, parent_app=None):
        self.parent_app = parent_app
        self.user = None
        self.notification_area = pn.pane.Markdown("")
        self.tasks_container = pn.Column(sizing_mode='stretch_width')
        self.tasks_table = None
        
    def set_user(self, user):
        """Set user and reload tasks"""
        self.user = user
        if user:
            self.load_tasks()
        return True
    
    def get_panel(self):
        """Get the tasks panel for embedding with optimized sizing for sidebar."""
        try:
            # Create a more compact header for sidebar use
            header = pn.pane.Markdown(
                "### 📋 My Tasks",
                margin=(0, 0, 5, 0),
                styles={'font-size': '16px', 'margin-bottom': '5px'}
            )
            
            # Create notification area with reduced size
            compact_notification = pn.pane.Markdown(
                self.notification_area.object if hasattr(self.notification_area, 'object') else "",
                margin=(0, 0, 5, 0),
                styles={'font-size': '12px', 'color': '#666'}
            )
            
            # Configure tasks container for sidebar
            if hasattr(self, 'tasks_table') and self.tasks_table:
                # Reconfigure the existing table for sidebar display
                self.tasks_table.height = 300  # Explicit height for sidebar
                self.tasks_table.width = 270   # Fit within sidebar width
                self.tasks_table.page_size = 3  # Smaller page size for compact display
                self.tasks_table.sizing_mode = 'stretch_width'
                
                # Hide less critical columns for space
                self.tasks_table.hidden_columns = ['task_id', 'resource_id', 'status']
                
                # Adjust column widths for compact display
                self.tasks_table.widths = {
                    'doc_number': 80,
                    'title': 120,
                    'task_type': 100,
                    'due_date': 80,
                    'action': 60
                }
                
                tasks_content = self.tasks_table
            else:
                # Create placeholder if no table exists yet
                tasks_content = pn.pane.Markdown(
                    "*Loading tasks...*",
                    styles={'text-align': 'center', 'color': '#666', 'font-style': 'italic'}
                )
            
            # Create compact container optimized for sidebar
            container = pn.Column(
                header,
                compact_notification,
                tasks_content,
                sizing_mode='stretch_width',
                margin=0,
                styles={
                    'max-height': '400px',  # Constrain height
                    'overflow-y': 'auto',   # Enable scrolling
                    'padding': '5px'        # Minimal padding
                }
            )
            
            return container
            
        except Exception as e:
            logger.error(f"Error creating tasks panel: {e}")
            return pn.Column(
                pn.pane.Markdown("### 📋 My Tasks"),
                pn.pane.Markdown("*Error loading tasks panel*", styles={'color': 'red'}),
                sizing_mode='stretch_width'
            )
    
    def load_tasks(self):
        """Load all tasks for the current user with compact display configuration."""
        try:
            self.notification_area.object = "Loading your tasks..."
            self.tasks_container.clear()
            
            if not self.user:
                self.notification_area.object = "Please log in to view your tasks."
                return
            
            # Get all task data
            review_tasks = self._get_review_tasks()
            approval_tasks = self._get_approval_tasks()
            completion_tasks = self._get_completion_tasks()
            
            # Combine all tasks
            all_tasks = review_tasks + approval_tasks + completion_tasks
            
            if not all_tasks:
                self.tasks_container.append(pn.pane.Markdown("*You have no pending tasks*"))
                self.notification_area.object = ""
                return
            
            # Create a DataFrame for the tasks
            df = pd.DataFrame(all_tasks)
            
            # Truncate long titles for compact display
            if 'title' in df.columns:
                df['title'] = df['title'].apply(lambda x: x[:25] + '...' if len(str(x)) > 25 else str(x))
            
            # Create a compact tasks table optimized for sidebar
            self.tasks_table = Tabulator(
                df,
                formatters={
                    'task_type': {'type': 'html'}, 
                    'action': {'type': 'html'}
                },
                pagination='local',
                page_size=3,  # Reduced from 5 to 3 for compact display
                sizing_mode='stretch_width',
                selectable=1,
                height=300,  # Reduced height for sidebar
                width=270,   # Explicit width for sidebar
                hidden_columns=['task_id', 'resource_id', 'status'],  # Hide status to save space
                show_index=False,  # Remove index column
                layout='fit_columns',  # FIXED: Changed from 'fitColumns' to 'fit_columns'
                # Configure column widths for compact display
                widths={
                    'doc_number': 70,
                    'title': 110,
                    'task_type': 90,
                    'due_date': 70,
                    'action': 50
                }
            )
            
            # Add click handler
            self.tasks_table.on_click(self._task_selected)
            
            # Add to container
            self.tasks_container.append(self.tasks_table)
            self.notification_area.object = f"Showing {len(all_tasks)} task(s)"
            
        except Exception as e:
            logger.error(f"Error loading user tasks: {e}")
            import traceback
            logger.error(traceback.format_exc())
            self.notification_area.object = f"**Error:** {str(e)}"
            self.tasks_container.clear()
            self.tasks_container.append(pn.pane.Markdown("Error loading tasks. Please try again."))
    
    def _get_review_tasks(self):
        """Get pending review tasks for the user"""
        try:
            # Get reviews assigned to this user that are pending/active
            result = get_user_assigned_reviews(
                user=self.user,
                status_filter=["PENDING", "ACTIVE"],
                include_completed=False
            )
            
            assignments = result.get('assignments', [])
            if not assignments:
                return []
            
            tasks = []
            for assignment in assignments:
                # Extract data
                review = assignment.get('review_cycle', {})
                document = assignment.get('document', {})
                reviewer_assignment = assignment.get('assignment', {})
                
                # Determine if this is an active task in a sequential review
                is_sequential = review.get('sequential', False)
                assignment_status = reviewer_assignment.get('status', '')
                
                # Create task indicator based on review type and status
                if assignment_status == "ACTIVE" or not is_sequential:
                    task_type = self._format_task_type("REVIEW", assignment_status)
                    
                    # Get due date and convert to string if it's a Neo4j datetime
                    due_date = review.get('due_date', review.get('dueDate', ''))
                    if hasattr(due_date, 'to_native'):
                        # Convert Neo4j DateTime to Python datetime
                        due_date = due_date.to_native().strftime('%Y-%m-%d')
                    elif isinstance(due_date, (datetime, pd.Timestamp)):
                        # Regular datetime objects
                        due_date = due_date.strftime('%Y-%m-%d')
                    
                    # Create task
                    task = {
                        'task_id': reviewer_assignment.get('UID', ''),
                        'resource_id': review.get('UID', ''),
                        'doc_number': document.get('doc_number', ''),
                        'title': document.get('title', ''),
                        'task_type': task_type,
                        'status': assignment_status,
                        'due_date': due_date,
                        'action': '<button class="btn btn-sm btn-primary">Review</button>',
                        'resource_type': 'review'
                    }
                    tasks.append(task)
            
            return tasks
            
        except Exception as e:
            logger.error(f"Error getting review tasks: {e}")
            import traceback
            logger.error(traceback.format_exc())
            return []
    
    def _get_approval_tasks(self):
        """Get pending approval tasks for the user"""
        try:
            # Get approvals assigned to this user that are pending/active
            from CDocs.controllers.approval_controller import get_user_assigned_approvals
            
            result = get_user_assigned_approvals(
                user=self.user,
                status_filter=["PENDING", "ACTIVE"],
                include_completed=False
            )
            
            assignments = result.get('assignments', [])
            if not assignments:
                return []
            
            tasks = []
            for assignment_data in assignments:
                # Extract data
                approval_cycle = assignment_data.get('approval_cycle', {})
                document = assignment_data.get('document', {})
                assignment = assignment_data.get('assignment', {})
                
                # Determine if this is active in a sequential workflow
                is_sequential = approval_cycle.get('sequential', False)
                assignment_status = assignment.get('status', '')
                
                # Only show active approvals or pending ones in parallel workflows
                if assignment_status == "ACTIVE" or (not is_sequential and assignment_status == "PENDING"):
                    task_type = self._format_task_type("APPROVAL", assignment_status)
                    
                    # Get due date and convert to string if it's a Neo4j datetime
                    due_date = approval_cycle.get('due_date', approval_cycle.get('dueDate', ''))
                    if hasattr(due_date, 'to_native'):
                        # Convert Neo4j DateTime to Python datetime
                        due_date = due_date.to_native().strftime('%Y-%m-%d')
                    elif isinstance(due_date, (datetime, pd.Timestamp)):
                        # Regular datetime objects
                        due_date = due_date.strftime('%Y-%m-%d')
                    
                    # Create task
                    task = {
                        'task_id': assignment.get('uid', assignment.get('UID', '')),
                        'resource_id': approval_cycle.get('uid', approval_cycle.get('UID', '')),
                        'doc_number': document.get('doc_number', ''),
                        'title': document.get('title', ''),
                        'task_type': task_type,
                        'status': assignment_status,
                        'due_date': due_date,
                        'action': '<button class="btn btn-sm btn-primary">Approve</button>',
                        'resource_type': 'approval'
                    }
                    tasks.append(task)
            
            return tasks
            
        except Exception as e:
            logger.error(f"Error getting approval tasks: {e}")
            import traceback
            logger.error(traceback.format_exc())
            return []
    
    def _get_completion_tasks(self):
        """Get review cycles and approval workflows that need completing"""
        try:
            # Get completed reviews that need closing
            completed_reviews = self._get_completed_reviews_to_close()
            
            # Get completed approvals that need closing
            completed_approvals = self._get_completed_approvals_to_close()
            
            # Combine tasks
            tasks = completed_reviews + completed_approvals
            return tasks
            
        except Exception as e:
            logger.error(f"Error getting completion tasks: {e}")
            return []
    
    def _get_completed_reviews_to_close(self):
        """Get completed reviews that need to be closed"""
        try:
            # Check if user has permissions to close reviews
            from CDocs.config import permissions
            if not permissions.user_has_permission(self.user, ["MANAGE_REVIEWS", "EDIT_DOCUMENT"]):
                return []
            
            # Get documents owned by this user
            from CDocs.controllers.document_controller import get_documents
            docs_result = get_documents(
                user=self.user,
                status="IN_REVIEW",
                owner=self.user.uid
            )
            
            documents = docs_result.get('documents', [])
            if not documents:
                return []
            
            tasks = []
            for doc in documents:
                # Get review cycles for this document
                from CDocs.controllers.review_controller import get_document_review_cycles
                doc_uid = doc.get('UID', '')
                if not doc_uid:
                    continue
                
                review_result = get_document_review_cycles(document_uid=doc_uid)
                review_cycles = review_result.get('review_cycles', [])
                
                # Check for completed review cycles
                for review_cycle in review_cycles:
                    if review_cycle.get('status') == 'COMPLETED':
                        task_type = self._format_task_type("COMPLETION", "REVIEW")
                        
                        # Create task
                        task = {
                            'task_id': review_cycle.get('UID', ''),
                            'resource_id': doc_uid,
                            'doc_number': doc.get('doc_number', doc.get('docNumber', '')),
                            'title': doc.get('title', ''),
                            'task_type': task_type,
                            'status': 'COMPLETED',
                            'due_date': '',  # No due date for completion tasks
                            'action': '<button class="btn btn-sm btn-success">Close Review</button>',
                            'resource_type': 'complete_review'
                        }
                        tasks.append(task)
            
            return tasks
            
        except Exception as e:
            logger.error(f"Error getting completed reviews: {e}")
            return []
    
    def _get_completed_approvals_to_close(self):
        """Get completed approvals that need to be closed"""
        try:
            # Check if user has permissions to close approvals
            from CDocs.config import permissions
            if not permissions.user_has_permission(self.user, ["MANAGE_APPROVALS"]):
                return []
            
            # Get documents owned by this user
            from CDocs.controllers.document_controller import get_documents
            docs_result = get_documents(
                user=self.user,
                status="IN_APPROVAL", 
                owner=self.user.uid
            )
            
            documents = docs_result.get('documents', [])
            if not documents:
                return []
            
            tasks = []
            for doc in documents:
                # Get approval workflows for this document
                from CDocs.controllers.approval_controller import get_document_approval_cycles
                doc_uid = doc.get('UID', '')
                if not doc_uid:
                    continue
                
                approval_result = get_document_approval_cycles(document_uid=doc_uid)
                approvals = approval_result.get('approval_cycles', [])
                logger.info("Approvals: %s", approvals)
                
                # Check for completed approval workflows
                for approval in approvals:
                    if approval.get('status') == 'COMPLETED':
                        task_type = self._format_task_type("COMPLETION", "APPROVAL")
                        
                        # Create task
                        task = {
                            'task_id': approval.get('UID', ''),
                            'resource_id': doc_uid,
                            'doc_number': doc.get('doc_number', doc.get('docNumber', '')),
                            'title': doc.get('title', ''),
                            'task_type': task_type,
                            'status': 'COMPLETED',
                            'due_date': '',  # No due date for completion tasks
                            'action': '<button class="btn btn-sm btn-success">Close Approval</button>',
                            'resource_type': 'complete_approval'
                        }
                        tasks.append(task)
            
            return tasks
            
        except Exception as e:
            logger.error(f"Error getting completed approvals: {e}")
            import traceback
            logger.error(traceback.format_exc())
            return []
    
    def _format_task_type(self, task_category, task_status):
        """Format task type with color coding and icons"""
        if task_category == "REVIEW":
            if task_status == "ACTIVE":
                return '<span style="color:#ff9900; font-weight:bold;"><i class="fas fa-exclamation-circle"></i> Active Review</span>'
            else:
                return '<span style="color:#0066cc; font-weight:bold;"><i class="fas fa-clipboard-list"></i> Pending Review</span>'
        elif task_category == "APPROVAL":
            if task_status == "ACTIVE":
                return '<span style="color:#cc3300; font-weight:bold;"><i class="fas fa-signature"></i> Active Approval</span>'
            else:
                return '<span style="color:#660066; font-weight:bold;"><i class="fas fa-stamp"></i> Pending Approval</span>'
        elif task_category == "COMPLETION":
            if task_status == "REVIEW":
                return '<span style="color:#009933; font-weight:bold;"><i class="fas fa-check-circle"></i> Close Review</span>'
            else:
                return '<span style="color:#009933; font-weight:bold;"><i class="fas fa-check-double"></i> Close Approval</span>'
        else:
            return '<span style="color:#666666;"><i class="fas fa-tasks"></i> Task</span>'
    
    def _task_selected(self, event):
        """Handle task selection from table"""
        try:
            # Get the currently selected row
            row_index = None
            
            # Handle different event types
            if hasattr(event, 'row') and event.row is not None:
                row_index = event.row
            elif hasattr(event, 'new') and event.new:
                row_index = event.new[0] if isinstance(event.new, list) else event.new
                
            if row_index is None or not hasattr(self, 'tasks_table'):
                return
                
            # Get task data from the table
            df = self.tasks_table.value
            if row_index >= len(df):
                return


            # Extract data
            task_row = df.iloc[row_index]
            resource_type = task_row.get('resource_type', '')
            resource_id = task_row.get('resource_id', '')
            task_id = task_row.get('task_id', '')
            
            # Handle different task types
            if resource_type == 'review':
                # Open review panel
                if hasattr(self.parent_app, 'load_reviews'):
                    self.parent_app.load_reviews()  # First load the review panel
                    
                    # Then load the specific review
                    if hasattr(self.parent_app, 'review_panel') and hasattr(self.parent_app.review_panel, '_load_review'):
                        self.parent_app.review_panel._load_review(resource_id)
            
            elif resource_type == 'approval':
                # Open approvals panel
                if hasattr(self.parent_app, 'load_approvals'):
                    self.parent_app.load_approvals()
                    
                    # Then load the specific approval
                    if hasattr(self.parent_app, 'approval_panel') and hasattr(self.parent_app.approval_panel, '_load_approval'):
                        self.parent_app.approval_panel._load_approval(resource_id)
            
            elif resource_type == 'complete_review':
                # Open document detail view and then trigger review cycle closing
                if hasattr(self.parent_app, 'load_document'):
                    self.parent_app.load_document(resource_id)
                    
            elif resource_type == 'complete_approval':
                # Open document detail view and then trigger approval workflow closing
                if hasattr(self.parent_app, 'load_document'):
                    self.parent_app.load_document(resource_id)
            
        except Exception as e:
            logger.error(f"Error handling task selection: {e}")
            import traceback
            logger.error(traceback.format_exc())
    
    def refresh_for_sidebar(self):
        """Refresh the panel specifically for sidebar display."""
        try:
            # Reload tasks with current user
            if self.user:
                self.load_tasks()
            
            # Ensure proper sizing for sidebar
            if hasattr(self, 'tasks_table') and self.tasks_table:
                self.tasks_table.height = 300
                self.tasks_table.width = 270
                self.tasks_table.page_size = 3
                
        except Exception as e:
            logger.error(f"Error refreshing tasks for sidebar: {e}")
    
    def get_panel_for_dashboard(self):
        """Get a clean tasks table for dashboard display without headers."""
        try:
            if not self.user:
                return pn.pane.Markdown("*Please log in to view tasks*")
            
            # Ensure tasks are loaded
            if not hasattr(self, 'tasks_table') or not self.tasks_table:
                self.load_tasks()
            
            # Configure for dashboard display
            if hasattr(self, 'tasks_table') and self.tasks_table:
                # Configure for full-width dashboard display
                self.tasks_table.height = 200
                self.tasks_table.sizing_mode = 'stretch_width'
                self.tasks_table.page_size = 5
                
                # Show more columns in full-width mode
                self.tasks_table.hidden_columns = ['task_id', 'resource_id']
                
                # Better column widths for full display
                self.tasks_table.widths = {
                    'doc_number': 120,
                    'title': 300,
                    'task_type': 150,
                    'status': 100,
                    'due_date': 100,
                    'action': 80
                }
                
                return self.tasks_table
            else:
                return pn.pane.Markdown("*No tasks available*")
                
        except Exception as e:
            logger.error(f"Error getting panel for dashboard: {e}")
            return pn.pane.Markdown("*Error loading tasks*")

Parameters

Name	Type	Default	Kind
bases	-	-

Parameter Details

bases: Parameter of type

Return Value

Returns unspecified type

Class Interface

Methods

Required Imports

import logging
import pandas as pd
import panel as pn
from panel.widgets import Tabulator
from panel.pane import Markdown

Usage Example

# Example usage:
# result = UserTasksPanel(bases)

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function create_user_tasks_panel 68.8% similar

  Factory function that creates and initializes a UserTasksPanel instance, optionally setting the current user from a parent application.

  From: /tf/active/vicechatdev/CDocs/ui/user_tasks_panel.py

• class User 51.9% similar

  A user management class that handles authentication, authorization, user profiles, preferences, file management, and logging for a Panel-based web application with Neo4j backend.

  From: /tf/active/vicechatdev/userclass.py

• class ApprovalPanel 49.7% similar

  Approval management interface component

  From: /tf/active/vicechatdev/CDocs/ui/approval_panel_bis.py

• class ApprovalPanel_v1 48.6% similar

  Approval management interface component

  From: /tf/active/vicechatdev/CDocs/ui/approval_panel.py

• class AdminPanel 44.5% similar

  Admin configuration interface component

  From: /tf/active/vicechatdev/CDocs/ui/admin_panel.py

class UserTasksPanel:
    """Panel showing pending tasks for the current user"""
    
    def __init__(self, parent_app=None):
        self.parent_app = parent_app
        self.user = None
        self.notification_area = pn.pane.Markdown("")
        self.tasks_container = pn.Column(sizing_mode='stretch_width')
        self.tasks_table = None
        
    def set_user(self, user):
        """Set user and reload tasks"""
        self.user = user
        if user:
            self.load_tasks()
        return True
    
    def get_panel(self):
        """Get the tasks panel for embedding with optimized sizing for sidebar."""
        try:
            # Create a more compact header for sidebar use
            header = pn.pane.Markdown(
                "### 📋 My Tasks",
                margin=(0, 0, 5, 0),
                styles={'font-size': '16px', 'margin-bottom': '5px'}
            )
            
            # Create notification area with reduced size
            compact_notification = pn.pane.Markdown(
                self.notification_area.object if hasattr(self.notification_area, 'object') else "",
                margin=(0, 0, 5, 0),
                styles={'font-size': '12px', 'color': '#666'}
            )
            
            # Configure tasks container for sidebar
            if hasattr(self, 'tasks_table') and self.tasks_table:
                # Reconfigure the existing table for sidebar display
                self.tasks_table.height = 300  # Explicit height for sidebar
                self.tasks_table.width = 270   # Fit within sidebar width
                self.tasks_table.page_size = 3  # Smaller page size for compact display
                self.tasks_table.sizing_mode = 'stretch_width'
                
                # Hide less critical columns for space
                self.tasks_table.hidden_columns = ['task_id', 'resource_id', 'status']
                
                # Adjust column widths for compact display
                self.tasks_table.widths = {
                    'doc_number': 80,
                    'title': 120,
                    'task_type': 100,
                    'due_date': 80,
                    'action': 60
                }
                
                tasks_content = self.tasks_table
            else:
                # Create placeholder if no table exists yet
                tasks_content = pn.pane.Markdown(
                    "*Loading tasks...*",
                    styles={'text-align': 'center', 'color': '#666', 'font-style': 'italic'}
                )
            
            # Create compact container optimized for sidebar
            container = pn.Column(
                header,
                compact_notification,
                tasks_content,
                sizing_mode='stretch_width',
                margin=0,
                styles={
                    'max-height': '400px',  # Constrain height
                    'overflow-y': 'auto',   # Enable scrolling
                    'padding': '5px'        # Minimal padding
                }
            )
            
            return container
            
        except Exception as e:
            logger.error(f"Error creating tasks panel: {e}")
            return pn.Column(
                pn.pane.Markdown("### 📋 My Tasks"),
                pn.pane.Markdown("*Error loading tasks panel*", styles={'color': 'red'}),
                sizing_mode='stretch_width'
            )
    
    def load_tasks(self):
        """Load all tasks for the current user with compact display configuration."""
        try:
            self.notification_area.object = "Loading your tasks..."
            self.tasks_container.clear()
            
            if not self.user:
                self.notification_area.object = "Please log in to view your tasks."
                return
            
            # Get all task data
            review_tasks = self._get_review_tasks()
            approval_tasks = self._get_approval_tasks()
            completion_tasks = self._get_completion_tasks()
            
            # Combine all tasks
            all_tasks = review_tasks + approval_tasks + completion_tasks
            
            if not all_tasks:
                self.tasks_container.append(pn.pane.Markdown("*You have no pending tasks*"))
                self.notification_area.object = ""
                return
            
            # Create a DataFrame for the tasks
            df = pd.DataFrame(all_tasks)
            
            # Truncate long titles for compact display
            if 'title' in df.columns:
                df['title'] = df['title'].apply(lambda x: x[:25] + '...' if len(str(x)) > 25 else str(x))
            
            # Create a compact tasks table optimized for sidebar
            self.tasks_table = Tabulator(
                df,
                formatters={
                    'task_type': {'type': 'html'}, 
                    'action': {'type': 'html'}
                },
                pagination='local',
                page_size=3,  # Reduced from 5 to 3 for compact display
                sizing_mode='stretch_width',
                selectable=1,
                height=300,  # Reduced height for sidebar
                width=270,   # Explicit width for sidebar
                hidden_columns=['task_id', 'resource_id', 'status'],  # Hide status to save space
                show_index=False,  # Remove index column
                layout='fit_columns',  # FIXED: Changed from 'fitColumns' to 'fit_columns'
                # Configure column widths for compact display
                widths={
                    'doc_number': 70,
                    'title': 110,
                    'task_type': 90,
                    'due_date': 70,
                    'action': 50
                }
            )
            
            # Add click handler
            self.tasks_table.on_click(self._task_selected)
            
            # Add to container
            self.tasks_container.append(self.tasks_table)
            self.notification_area.object = f"Showing {len(all_tasks)} task(s)"
            
        except Exception as e:
            logger.error(f"Error loading user tasks: {e}")
            import traceback
            logger.error(traceback.format_exc())
            self.notification_area.object = f"**Error:** {str(e)}"
            self.tasks_container.clear()
            self.tasks_container.append(pn.pane.Markdown("Error loading tasks. Please try again."))
    
    def _get_review_tasks(self):
        """Get pending review tasks for the user"""
        try:
            # Get reviews assigned to this user that are pending/active
            result = get_user_assigned_reviews(
                user=self.user,
                status_filter=["PENDING", "ACTIVE"],
                include_completed=False
            )
            
            assignments = result.get('assignments', [])
            if not assignments:
                return []
            
            tasks = []
            for assignment in assignments:
                # Extract data
                review = assignment.get('review_cycle', {})
                document = assignment.get('document', {})
                reviewer_assignment = assignment.get('assignment', {})
                
                # Determine if this is an active task in a sequential review
                is_sequential = review.get('sequential', False)
                assignment_status = reviewer_assignment.get('status', '')
                
                # Create task indicator based on review type and status
                if assignment_status == "ACTIVE" or not is_sequential:
                    task_type = self._format_task_type("REVIEW", assignment_status)
                    
                    # Get due date and convert to string if it's a Neo4j datetime
                    due_date = review.get('due_date', review.get('dueDate', ''))
                    if hasattr(due_date, 'to_native'):
                        # Convert Neo4j DateTime to Python datetime
                        due_date = due_date.to_native().strftime('%Y-%m-%d')
                    elif isinstance(due_date, (datetime, pd.Timestamp)):
                        # Regular datetime objects
                        due_date = due_date.strftime('%Y-%m-%d')
                    
                    # Create task
                    task = {
                        'task_id': reviewer_assignment.get('UID', ''),
                        'resource_id': review.get('UID', ''),
                        'doc_number': document.get('doc_number', ''),
                        'title': document.get('title', ''),
                        'task_type': task_type,
                        'status': assignment_status,
                        'due_date': due_date,
                        'action': '<button class="btn btn-sm btn-primary">Review</button>',
                        'resource_type': 'review'
                    }
                    tasks.append(task)
            
            return tasks
            
        except Exception as e:
            logger.error(f"Error getting review tasks: {e}")
            import traceback
            logger.error(traceback.format_exc())
            return []
    
    def _get_approval_tasks(self):
        """Get pending approval tasks for the user"""
        try:
            # Get approvals assigned to this user that are pending/active
            from CDocs.controllers.approval_controller import get_user_assigned_approvals
            
            result = get_user_assigned_approvals(
                user=self.user,
                status_filter=["PENDING", "ACTIVE"],
                include_completed=False
            )
            
            assignments = result.get('assignments', [])
            if not assignments:
                return []
            
            tasks = []
            for assignment_data in assignments:
                # Extract data
                approval_cycle = assignment_data.get('approval_cycle', {})
                document = assignment_data.get('document', {})
                assignment = assignment_data.get('assignment', {})
                
                # Determine if this is active in a sequential workflow
                is_sequential = approval_cycle.get('sequential', False)
                assignment_status = assignment.get('status', '')
                
                # Only show active approvals or pending ones in parallel workflows
                if assignment_status == "ACTIVE" or (not is_sequential and assignment_status == "PENDING"):
                    task_type = self._format_task_type("APPROVAL", assignment_status)
                    
                    # Get due date and convert to string if it's a Neo4j datetime
                    due_date = approval_cycle.get('due_date', approval_cycle.get('dueDate', ''))
                    if hasattr(due_date, 'to_native'):
                        # Convert Neo4j DateTime to Python datetime
                        due_date = due_date.to_native().strftime('%Y-%m-%d')
                    elif isinstance(due_date, (datetime, pd.Timestamp)):
                        # Regular datetime objects
                        due_date = due_date.strftime('%Y-%m-%d')
                    
                    # Create task
                    task = {
                        'task_id': assignment.get('uid', assignment.get('UID', '')),
                        'resource_id': approval_cycle.get('uid', approval_cycle.get('UID', '')),
                        'doc_number': document.get('doc_number', ''),
                        'title': document.get('title', ''),
                        'task_type': task_type,
                        'status': assignment_status,
                        'due_date': due_date,
                        'action': '<button class="btn btn-sm btn-primary">Approve</button>',
                        'resource_type': 'approval'
                    }
                    tasks.append(task)
            
            return tasks
            
        except Exception as e:
            logger.error(f"Error getting approval tasks: {e}")
            import traceback
            logger.error(traceback.format_exc())
            return []
    
    def _get_completion_tasks(self):
        """Get review cycles and approval workflows that need completing"""
        try:
            # Get completed reviews that need closing
            completed_reviews = self._get_completed_reviews_to_close()
            
            # Get completed approvals that need closing
            completed_approvals = self._get_completed_approvals_to_close()
            
            # Combine tasks
            tasks = completed_reviews + completed_approvals
            return tasks
            
        except Exception as e:
            logger.error(f"Error getting completion tasks: {e}")
            return []
    
    def _get_completed_reviews_to_close(self):
        """Get completed reviews that need to be closed"""
        try:
            # Check if user has permissions to close reviews
            from CDocs.config import permissions
            if not permissions.user_has_permission(self.user, ["MANAGE_REVIEWS", "EDIT_DOCUMENT"]):
                return []
            
            # Get documents owned by this user
            from CDocs.controllers.document_controller import get_documents
            docs_result = get_documents(
                user=self.user,
                status="IN_REVIEW",
                owner=self.user.uid
            )
            
            documents = docs_result.get('documents', [])
            if not documents:
                return []
            
            tasks = []
            for doc in documents:
                # Get review cycles for this document
                from CDocs.controllers.review_controller import get_document_review_cycles
                doc_uid = doc.get('UID', '')
                if not doc_uid:
                    continue
                
                review_result = get_document_review_cycles(document_uid=doc_uid)
                review_cycles = review_result.get('review_cycles', [])
                
                # Check for completed review cycles
                for review_cycle in review_cycles:
                    if review_cycle.get('status') == 'COMPLETED':
                        task_type = self._format_task_type("COMPLETION", "REVIEW")
                        
                        # Create task
                        task = {
                            'task_id': review_cycle.get('UID', ''),
                            'resource_id': doc_uid,
                            'doc_number': doc.get('doc_number', doc.get('docNumber', '')),
                            'title': doc.get('title', ''),
                            'task_type': task_type,
                            'status': 'COMPLETED',
                            'due_date': '',  # No due date for completion tasks
                            'action': '<button class="btn btn-sm btn-success">Close Review</button>',
                            'resource_type': 'complete_review'
                        }
                        tasks.append(task)
            
            return tasks
            
        except Exception as e:
            logger.error(f"Error getting completed reviews: {e}")
            return []
    
    def _get_completed_approvals_to_close(self):
        """Get completed approvals that need to be closed"""
        try:
            # Check if user has permissions to close approvals
            from CDocs.config import permissions
            if not permissions.user_has_permission(self.user, ["MANAGE_APPROVALS"]):
                return []
            
            # Get documents owned by this user
            from CDocs.controllers.document_controller import get_documents
            docs_result = get_documents(
                user=self.user,
                status="IN_APPROVAL", 
                owner=self.user.uid
            )
            
            documents = docs_result.get('documents', [])
            if not documents:
                return []
            
            tasks = []
            for doc in documents:
                # Get approval workflows for this document
                from CDocs.controllers.approval_controller import get_document_approval_cycles
                doc_uid = doc.get('UID', '')
                if not doc_uid:
                    continue
                
                approval_result = get_document_approval_cycles(document_uid=doc_uid)
                approvals = approval_result.get('approval_cycles', [])
                logger.info("Approvals: %s", approvals)
                
                # Check for completed approval workflows
                for approval in approvals:
                    if approval.get('status') == 'COMPLETED':
                        task_type = self._format_task_type("COMPLETION", "APPROVAL")
                        
                        # Create task
                        task = {
                            'task_id': approval.get('UID', ''),
                            'resource_id': doc_uid,
                            'doc_number': doc.get('doc_number', doc.get('docNumber', '')),
                            'title': doc.get('title', ''),
                            'task_type': task_type,
                            'status': 'COMPLETED',
                            'due_date': '',  # No due date for completion tasks
                            'action': '<button class="btn btn-sm btn-success">Close Approval</button>',
                            'resource_type': 'complete_approval'
                        }
                        tasks.append(task)
            
            return tasks
            
        except Exception as e:
            logger.error(f"Error getting completed approvals: {e}")
            import traceback
            logger.error(traceback.format_exc())
            return []
    
    def _format_task_type(self, task_category, task_status):
        """Format task type with color coding and icons"""
        if task_category == "REVIEW":
            if task_status == "ACTIVE":
                return '<span style="color:#ff9900; font-weight:bold;"><i class="fas fa-exclamation-circle"></i> Active Review</span>'
            else:
                return '<span style="color:#0066cc; font-weight:bold;"><i class="fas fa-clipboard-list"></i> Pending Review</span>'
        elif task_category == "APPROVAL":
            if task_status == "ACTIVE":
                return '<span style="color:#cc3300; font-weight:bold;"><i class="fas fa-signature"></i> Active Approval</span>'
            else:
                return '<span style="color:#660066; font-weight:bold;"><i class="fas fa-stamp"></i> Pending Approval</span>'
        elif task_category == "COMPLETION":
            if task_status == "REVIEW":
                return '<span style="color:#009933; font-weight:bold;"><i class="fas fa-check-circle"></i> Close Review</span>'
            else:
                return '<span style="color:#009933; font-weight:bold;"><i class="fas fa-check-double"></i> Close Approval</span>'
        else:
            return '<span style="color:#666666;"><i class="fas fa-tasks"></i> Task</span>'
    
    def _task_selected(self, event):
        """Handle task selection from table"""
        try:
            # Get the currently selected row
            row_index = None
            
            # Handle different event types
            if hasattr(event, 'row') and event.row is not None:
                row_index = event.row
            elif hasattr(event, 'new') and event.new:
                row_index = event.new[0] if isinstance(event.new, list) else event.new
                
            if row_index is None or not hasattr(self, 'tasks_table'):
                return
                
            # Get task data from the table
            df = self.tasks_table.value
            if row_index >= len(df):
                return


            # Extract data
            task_row = df.iloc[row_index]
            resource_type = task_row.get('resource_type', '')
            resource_id = task_row.get('resource_id', '')
            task_id = task_row.get('task_id', '')
            
            # Handle different task types
            if resource_type == 'review':
                # Open review panel
                if hasattr(self.parent_app, 'load_reviews'):
                    self.parent_app.load_reviews()  # First load the review panel
                    
                    # Then load the specific review
                    if hasattr(self.parent_app, 'review_panel') and hasattr(self.parent_app.review_panel, '_load_review'):
                        self.parent_app.review_panel._load_review(resource_id)
            
            elif resource_type == 'approval':
                # Open approvals panel
                if hasattr(self.parent_app, 'load_approvals'):
                    self.parent_app.load_approvals()
                    
                    # Then load the specific approval
                    if hasattr(self.parent_app, 'approval_panel') and hasattr(self.parent_app.approval_panel, '_load_approval'):
                        self.parent_app.approval_panel._load_approval(resource_id)
            
            elif resource_type == 'complete_review':
                # Open document detail view and then trigger review cycle closing
                if hasattr(self.parent_app, 'load_document'):
                    self.parent_app.load_document(resource_id)
                    
            elif resource_type == 'complete_approval':
                # Open document detail view and then trigger approval workflow closing
                if hasattr(self.parent_app, 'load_document'):
                    self.parent_app.load_document(resource_id)
            
        except Exception as e:
            logger.error(f"Error handling task selection: {e}")
            import traceback
            logger.error(traceback.format_exc())
    
    def refresh_for_sidebar(self):
        """Refresh the panel specifically for sidebar display."""
        try:
            # Reload tasks with current user
            if self.user:
                self.load_tasks()
            
            # Ensure proper sizing for sidebar
            if hasattr(self, 'tasks_table') and self.tasks_table:
                self.tasks_table.height = 300
                self.tasks_table.width = 270
                self.tasks_table.page_size = 3
                
        except Exception as e:
            logger.error(f"Error refreshing tasks for sidebar: {e}")
    
    def get_panel_for_dashboard(self):
        """Get a clean tasks table for dashboard display without headers."""
        try:
            if not self.user:
                return pn.pane.Markdown("*Please log in to view tasks*")
            
            # Ensure tasks are loaded
            if not hasattr(self, 'tasks_table') or not self.tasks_table:
                self.load_tasks()
            
            # Configure for dashboard display
            if hasattr(self, 'tasks_table') and self.tasks_table:
                # Configure for full-width dashboard display
                self.tasks_table.height = 200
                self.tasks_table.sizing_mode = 'stretch_width'
                self.tasks_table.page_size = 5
                
                # Show more columns in full-width mode
                self.tasks_table.hidden_columns = ['task_id', 'resource_id']
                
                # Better column widths for full display
                self.tasks_table.widths = {
                    'doc_number': 120,
                    'title': 300,
                    'task_type': 150,
                    'status': 100,
                    'due_date': 100,
                    'action': 80
                }
                
                return self.tasks_table
            else:
                return pn.pane.Markdown("*No tasks available*")
                
        except Exception as e:
            logger.error(f"Error getting panel for dashboard: {e}")
            return pn.pane.Markdown("*Error loading tasks*")