Create app.py

app.py

@@ -0,0 +1,342 @@
+import gradio as gr
+import os
+import csv
+from datetime import datetime
+from huggingface_hub import Repository
+
+# Configuration
+DATA_STORAGE_REPO = "CIV3283/Data_Storage"
+DATA_BRANCH_NAME = "data_branch"
+LOCAL_DATA_DIR = "temp_data_storage"
+MIN_IDLE_MINUTES = 30  # Minimum idle time required for space assignment
+
+# Environment variables
+HF_HUB_TOKEN = os.environ.get("HF_HUB_TOKEN", None)
+if HF_HUB_TOKEN is None:
+    raise ValueError("Set HF_HUB_TOKEN in Space Settings -> Secrets")
+
+def init_data_storage_connection():
+    """Initialize connection to centralized data storage repository"""
+    try:
+        repo = Repository(
+            local_dir=LOCAL_DATA_DIR,
+            clone_from=DATA_STORAGE_REPO,
+            revision=DATA_BRANCH_NAME,
+            repo_type="space",
+            use_auth_token=HF_HUB_TOKEN
+        )
+        
+        print(f"[init_data_storage_connection] Pulling latest data from {DATA_STORAGE_REPO}...")
+        repo.git_pull(rebase=True)
+        
+        print(f"[init_data_storage_connection] Successfully connected to data storage")
+        print(f"[init_data_storage_connection] Local directory: {LOCAL_DATA_DIR}")
+        print(f"[init_data_storage_connection] Branch: {DATA_BRANCH_NAME}")
+        
+        return repo
+        
+    except Exception as e:
+        print(f"[init_data_storage_connection] Error: {e}")
+        return None
+
+def get_available_spaces(repo_dir):
+    """Dynamically get space list from CIV3283/Data_Storage"""
+    available_spaces = set()
+    
+    try:
+        # Scan all *_query_log.csv files
+        for filename in os.listdir(repo_dir):
+            if filename.endswith('_query_log.csv') and '_Student_' in filename:
+                # Extract space name from filename: CIV3283_Student_01_query_log.csv → CIV3283_Student_01
+                space_name = filename.replace('_query_log.csv', '')
+                available_spaces.add(space_name)
+        
+        spaces_list = sorted(list(available_spaces))
+        print(f"[get_available_spaces] Found {len(spaces_list)} student spaces: {spaces_list}")
+        return spaces_list
+        
+    except Exception as e:
+        print(f"[get_available_spaces] Error: {e}")
+        return []
+
+def get_last_activity_time(csv_file_path):
+    """Get the timestamp of the last activity from query log file"""
+    try:
+        if not os.path.exists(csv_file_path):
+            print(f"[get_last_activity_time] File not found: {csv_file_path}")
+            return datetime.min  # Never used
+        
+        with open(csv_file_path, 'r', encoding='utf-8') as f:
+            lines = f.readlines()
+            
+        if len(lines) <= 1:  # Only header or empty file
+            print(f"[get_last_activity_time] No data in file: {csv_file_path}")
+            return datetime.min  # Never used
+        
+        # Get last line and parse timestamp (3rd column, index 2)
+        last_line = lines[-1].strip()
+        if not last_line:
+            return datetime.min
+        
+        # Parse CSV line
+        csv_reader = csv.reader([last_line])
+        row = next(csv_reader)
+        
+        if len(row) >= 3:
+            timestamp_str = row[2].strip()  # timestamp column
+            try:
+                return datetime.strptime(timestamp_str, '%Y-%m-%d %H:%M:%S')
+            except ValueError as ve:
+                print(f"[get_last_activity_time] Date parsing error: {ve}")
+                return datetime.min
+        else:
+            print(f"[get_last_activity_time] Invalid CSV format in {csv_file_path}")
+            return datetime.min
+            
+    except Exception as e:
+        print(f"[get_last_activity_time] Error reading {csv_file_path}: {e}")
+        return datetime.min
+
+def analyze_space_activity(available_spaces, repo_dir):
+    """Analyze activity status of all spaces and return sorted list"""
+    space_activity = []
+    current_time = datetime.now()
+    
+    print(f"[analyze_space_activity] Analyzing {len(available_spaces)} spaces...")
+    
+    for space_name in available_spaces:
+        csv_file = os.path.join(repo_dir, f"{space_name}_query_log.csv")
+        last_activity = get_last_activity_time(csv_file)
+        
+        # Calculate idle time in minutes
+        if last_activity == datetime.min:
+            idle_minutes = float('inf')  # Never used
+            status = "Never used"
+            last_activity_str = "Never"
+        else:
+            idle_minutes = (current_time - last_activity).total_seconds() / 60
+            status = f"Idle for {idle_minutes:.1f} minutes"
+            last_activity_str = last_activity.strftime('%Y-%m-%d %H:%M:%S')
+        
+        space_activity.append({
+            'space_name': space_name,
+            'last_activity': last_activity,
+            'last_activity_str': last_activity_str,
+            'idle_minutes': idle_minutes,
+            'status': status
+        })
+        
+        print(f"[analyze_space_activity] {space_name}: {status}")
+    
+    # Sort by idle time (most idle first)
+    space_activity.sort(key=lambda x: x['idle_minutes'], reverse=True)
+    
+    return space_activity
+
+def create_status_display(space_activity):
+    """Create formatted status display for all spaces"""
+    status_display = "📊 **Current Space Status (sorted by availability):**\n\n"
+    
+    for i, space in enumerate(space_activity, 1):
+        status_display += f"{i}. **{space['space_name']}**\n"
+        status_display += f"   • Status: {space['status']}\n"
+        status_display += f"   • Last activity: {space['last_activity_str']}\n\n"
+    
+    return status_display
+
+def select_space_with_boundary_check(space_activity, student_id):
+    """Select space with strict boundary conditions"""
+    
+    # Create and display current status
+    status_display = create_status_display(space_activity)
+    print(f"[select_space_with_boundary_check] Space analysis:\n{status_display}")
+    
+    # Check for available spaces (idle for at least MIN_IDLE_MINUTES)
+    available_spaces = [s for s in space_activity if s['idle_minutes'] >= MIN_IDLE_MINUTES]
+    
+    if not available_spaces:
+        # All spaces are busy (used within the last 30 minutes)
+        most_idle = space_activity[0] if space_activity else None
+        
+        if most_idle:
+            if most_idle['idle_minutes'] == float('inf'):
+                idle_info = "Never used"
+            else:
+                idle_info = f"{most_idle['idle_minutes']:.1f} minutes ago"
+            
+            error_msg = (
+                f"🚫 **All learning assistants are currently busy**\n\n"
+                f"All spaces have been used within the last {MIN_IDLE_MINUTES} minutes.\n\n"
+                f"**Current Status:**\n"
+                f"• Most idle space: {most_idle['space_name']}\n"
+                f"• Last used: {idle_info}\n\n"
+                f"**Please try again in a few minutes.**"
+            )
+        else:
+            error_msg = "🚫 No learning assistant spaces are available at the moment."
+        
+        print(f"[select_space_with_boundary_check] All spaces busy: {error_msg}")
+        raise gr.Error(error_msg, duration=10)
+    
+    # Select the most idle space
+    selected_space = available_spaces[0]
+    print(f"[select_space_with_boundary_check] Selected space: {selected_space['space_name']}")
+    
+    # Build redirect URL
+    redirect_url = f"https://huggingface.co/spaces/{selected_space['space_name']}/?check={student_id}"
+    
+    return redirect_to_space(redirect_url, selected_space, status_display)
+
+def redirect_to_space(redirect_url, selected_space, status_display):
+    """Execute redirect with detailed information display"""
+    
+    if selected_space['idle_minutes'] == float('inf'):
+        idle_info = "Never used (completely fresh)"
+    else:
+        idle_info = f"{selected_space['idle_minutes']:.1f} minutes"
+    
+    redirect_html = f"""
+    <div style="max-width: 900px; margin: 0 auto; padding: 20px; font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;">
+        <div style="text-align: center; margin-bottom: 30px; padding: 20px; background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); color: white; border-radius: 12px;">
+            <h1 style="margin: 0 0 10px 0;">🎯 Learning Assistant Assigned</h1>
+            <h2 style="margin: 0; font-weight: normal; opacity: 0.9;">{selected_space['space_name']}</h2>
+            <p style="margin: 10px 0 0 0; font-size: 16px;">
+                ✨ This space was idle for: <strong>{idle_info}</strong>
+            </p>
+        </div>
+        
+        <div style="background: #f8f9fa; padding: 25px; border-radius: 12px; margin-bottom: 25px; border-left: 4px solid #007bff;">
+            <h3 style="margin-top: 0; color: #333;">📊 Space Selection Analysis</h3>
+            <div style="background: white; padding: 20px; border-radius: 8px; font-family: 'Courier New', monospace; font-size: 14px; line-height: 1.6; overflow-x: auto; border: 1px solid #e9ecef;">
+{status_display}
+            </div>
+            <p style="margin-bottom: 0; color: #666; font-size: 14px; margin-top: 15px;">
+                <strong>Selection Algorithm:</strong> Spaces idle for ≥{MIN_IDLE_MINUTES} minutes are eligible. The most idle space is automatically selected for optimal load distribution.
+            </p>
+        </div>
+        
+        <div style="text-align: center; background: linear-gradient(135deg, #28a745, #20c997); color: white; padding: 25px; border-radius: 12px; box-shadow: 0 4px 6px rgba(0,0,0,0.1);">
+            <h3 style="margin-top: 0;">🔄 Redirecting to Your Learning Assistant</h3>
+            <p style="margin: 15px 0;">You will be automatically redirected in <span id="countdown">5</span> seconds.</p>
+            
+            <div style="margin: 25px 0;">
+                <a href="{redirect_url}" 
+                   style="display: inline-block; background: rgba(255,255,255,0.2); color: white; padding: 12px 24px; 
+                          text-decoration: none; border-radius: 25px; font-weight: bold; border: 2px solid rgba(255,255,255,0.3);
+                          transition: all 0.3s ease;"
+                   onmouseover="this.style.background='rgba(255,255,255,0.3)'"
+                   onmouseout="this.style.background='rgba(255,255,255,0.2)'">
+                    Click Here if Not Redirected Automatically
+                </a>
+            </div>
+            
+            <details style="margin-top: 20px; text-align: left;">
+                <summary style="cursor: pointer; color: rgba(255,255,255,0.9);">🔗 View Target URL</summary>
+                <p style="margin: 10px 0 0 0; padding: 10px; background: rgba(0,0,0,0.1); border-radius: 5px; word-break: break-all; font-family: monospace; font-size: 12px;">
+                    {redirect_url}
+                </p>
+            </details>
+        </div>
+    </div>
+    
+    <script>
+        let countdown = 5;
+        const countdownElement = document.getElementById('countdown');
+        
+        const timer = setInterval(function() {{
+            countdown--;
+            if (countdownElement) {{
+                countdownElement.textContent = countdown;
+            }}
+            
+            if (countdown <= 0) {{
+                clearInterval(timer);
+                window.location.href = "{redirect_url}";
+            }}
+        }}, 1000);
+    </script>
+    """
+    
+    return gr.HTML(redirect_html)
+
+def load_balance_user(student_id):
+    """Main load balancing function"""
+    print(f"[load_balance_user] Starting load balancing for student ID: {student_id}")
+    
+    # Initialize connection to data storage
+    repo = init_data_storage_connection()
+    if not repo:
+        raise gr.Error("🚫 Unable to connect to the data storage system. Please try again later.", duration=8)
+    
+    # Get available spaces dynamically
+    available_spaces = get_available_spaces(LOCAL_DATA_DIR)
+    
+    if not available_spaces:
+        raise gr.Error("🚫 No student learning assistant spaces found in the system. Please contact your instructor.", duration=8)
+    
+    # Analyze space activity
+    space_activity = analyze_space_activity(available_spaces, LOCAL_DATA_DIR)
+    
+    # Select space with boundary checking
+    return select_space_with_boundary_check(space_activity, student_id)
+
+def get_url_params(request: gr.Request):
+    """Extract URL parameters from request"""
+    if request:
+        query_params = dict(request.query_params)
+        check_id = query_params.get('check', None)
+        if check_id:
+            return f"Load Distributor - Student {check_id}", check_id
+        else:
+            return "Load Distributor", None
+    return "Load Distributor", None
+
+def handle_user_access(request: gr.Request):
+    """Handle user access and perform load balancing"""
+    title, check_id = get_url_params(request)
+    
+    if not check_id:
+        # No student ID provided
+        error_html = """
+        <div style="max-width: 600px; margin: 50px auto; padding: 30px; text-align: center; 
+                    background: #fff3cd; border: 1px solid #ffeaa7; border-radius: 12px;">
+            <h2 style="color: #856404;">⚠️ Invalid Access</h2>
+            <p style="color: #856404; font-size: 16px; line-height: 1.6;">
+                This load distributor requires a valid student ID parameter.<br><br>
+                <strong>Please access this system through the official link provided in Moodle.</strong>
+            </p>
+        </div>
+        """
+        return title, gr.HTML(error_html)
+    
+    # Valid student ID - perform load balancing
+    try:
+        result = load_balance_user(check_id)
+        return title, result
+    except Exception as e:
+        # Handle any errors during load balancing
+        error_html = f"""
+        <div style="max-width: 600px; margin: 50px auto; padding: 30px; text-align: center; 
+                    background: #f8d7da; border: 1px solid #f5c6cb; border-radius: 12px;">
+            <h2 style="color: #721c24;">🚫 Load Balancing Error</h2>
+            <p style="color: #721c24; font-size: 16px; line-height: 1.6;">
+                {str(e)}<br><br>
+                Please try again in a few moments or contact your instructor if the problem persists.
+            </p>
+        </div>
+        """
+        return title, gr.HTML(error_html)
+
+# Create Gradio interface
+with gr.Blocks(title="CIV3283 Load Distributor") as interface:
+    title_display = gr.Markdown("# 🔄 CIV3283 Learning Assistant Load Distributor", elem_id="title")
+    content_display = gr.HTML("")
+    
+    # Initialize on page load
+    interface.load(
+        fn=handle_user_access,
+        outputs=[title_display, content_display]
+    )
+
+if __name__ == "__main__":
+    interface.launch()