Update app.py

app.py

@@ -5,6 +5,8 @@ from datetime import datetime, timedelta
 from huggingface_hub import Repository
 import threading
 import time
+import json
+
 
 # Configuration
 DATA_STORAGE_REPO = "CIV3283/Data_Storage"
@@ -14,6 +16,371 @@ MIN_IDLE_MINUTES = 10  # Minimum idle time required for space assignment
 ALLOCATION_RECORD_FILE = "allocation_records.csv"  # New file for tracking allocations
 ALLOCATION_LOCK_DURATION = 5  # Lock duration in minutes
 
+# 缓存配置
+ACTIVITY_CACHE_FILE = "space_activity_cache.json"
+CACHE_UPDATE_INTERVAL = 600  # 10分钟 = 600秒
+CACHE_LOCK = threading.Lock()
+
+class SpaceActivityCache:
+    def __init__(self, repo_dir):
+        self.repo_dir = repo_dir
+        self.cache_file_path = os.path.join(repo_dir, ACTIVITY_CACHE_FILE)
+        self._cache_data = {}
+        self._last_update = None
+        self._update_thread = None
+        self._stop_event = threading.Event()
+        
+        # 启动时立即更新一次缓存
+        self._update_cache()
+        
+        # 启动后台更新线程
+        self.start_background_updates()
+    
+    def start_background_updates(self):
+        """启动后台缓存更新线程"""
+        if self._update_thread and self._update_thread.is_alive():
+            return  # 已经在运行
+        
+        self._stop_event.clear()
+        self._update_thread = threading.Thread(target=self._background_update_worker, daemon=True)
+        self._update_thread.start()
+        print("[SpaceActivityCache] Background update thread started")
+    
+    def stop_background_updates(self):
+        """停止后台更新线程"""
+        if self._update_thread:
+            self._stop_event.set()
+            print("[SpaceActivityCache] Background update thread stopping...")
+    
+    def _background_update_worker(self):
+        """后台更新工作线程"""
+        while not self._stop_event.is_set():
+            try:
+                # 等待指定间隔或停止信号
+                if self._stop_event.wait(timeout=CACHE_UPDATE_INTERVAL):
+                    break  # 收到停止信号
+                
+                # 执行缓存更新
+                print("[SpaceActivityCache] Starting scheduled cache update...")
+                self._update_cache()
+                print("[SpaceActivityCache] Scheduled cache update completed")
+                
+            except Exception as e:
+                print(f"[SpaceActivityCache] Error in background update: {e}")
+                # 发生错误时等待较短时间后重试
+                if not self._stop_event.wait(timeout=60):  # 1分钟后重试
+                    continue
+    
+    def _update_cache(self):
+        """更新缓存数据"""
+        try:
+            print("[SpaceActivityCache] Updating activity cache...")
+            start_time = time.time()
+            
+            # 获取所有可用空间
+            available_spaces = self._get_available_spaces()
+            
+            new_cache_data = {
+                'last_updated': datetime.now().isoformat(),
+                'update_interval_seconds': CACHE_UPDATE_INTERVAL,
+                'spaces': {}
+            }
+            
+            # 为每个空间读取最后活动时间
+            for space_name in available_spaces:
+                csv_file = os.path.join(self.repo_dir, f"{space_name}_query_log.csv")
+                last_activity, status = self._get_last_activity_from_file(csv_file)
+                
+                new_cache_data['spaces'][space_name] = {
+                    'last_activity': last_activity.isoformat() if last_activity != datetime.min else None,
+                    'status': status,
+                    'cache_update_time': datetime.now().isoformat()
+                }
+                
+                print(f"[SpaceActivityCache] {space_name}: {status}")
+            
+            # 线程安全地更新缓存
+            with CACHE_LOCK:
+                self._cache_data = new_cache_data
+                self._last_update = datetime.now()
+                
+                # 保存到文件
+                self._save_cache_to_file()
+            
+            elapsed_time = time.time() - start_time
+            print(f"[SpaceActivityCache] Cache updated in {elapsed_time:.2f}s, {len(available_spaces)} spaces processed")
+            
+        except Exception as e:
+            print(f"[SpaceActivityCache] Error updating cache: {e}")
+            import traceback
+            print(f"[SpaceActivityCache] Traceback: {traceback.format_exc()}")
+    
+    def _get_available_spaces(self):
+        """获取可用空间列表"""
+        available_spaces = set()
+        try:
+            for filename in os.listdir(self.repo_dir):
+                if filename.endswith('_query_log.csv') and '_Student_' in filename:
+                    space_name = filename.replace('_query_log.csv', '')
+                    available_spaces.add(space_name)
+            return sorted(list(available_spaces))
+        except Exception as e:
+            print(f"[SpaceActivityCache] Error getting available spaces: {e}")
+            return []
+    
+    def _get_last_activity_from_file(self, csv_file_path):
+        """从文件读取最后活动时间（使用原来的方法）"""
+        try:
+            if not os.path.exists(csv_file_path):
+                return datetime.min, "file_not_found"
+            
+            # 检查文件大小
+            file_size = os.path.getsize(csv_file_path)
+            if file_size <= 100:
+                return datetime.min, "empty_or_header_only"
+            
+            # 使用CSV reader读取最后一行
+            with open(csv_file_path, 'r', encoding='utf-8') as f:
+                csv_reader = csv.reader(f)
+                
+                # 跳过header
+                try:
+                    header = next(csv_reader)
+                except StopIteration:
+                    return datetime.min, "empty_file"
+                
+                # 读取所有行，获取最后一行
+                rows = []
+                try:
+                    for row in csv_reader:
+                        if row:  # 跳过空行
+                            rows.append(row)
+                except Exception as csv_error:
+                    print(f"[SpaceActivityCache] CSV parsing error for {csv_file_path}: {csv_error}")
+                    return datetime.min, "csv_parse_error"
+            
+            if not rows:
+                return datetime.min, "no_data_rows"
+            
+            # 解析最后一行的时间戳
+            last_row = rows[-1]
+            if len(last_row) >= 3:
+                timestamp_str = last_row[2].strip()  # timestamp column
+                try:
+                    parsed_time = datetime.strptime(timestamp_str, '%Y-%m-%d %H:%M:%S')
+                    return parsed_time, f"active_last_at_{parsed_time.strftime('%Y-%m-%d_%H:%M:%S')}"
+                except ValueError as ve:
+                    print(f"[SpaceActivityCache] Date parsing error for '{timestamp_str}': {ve}")
+                    return datetime.min, "date_parse_error"
+            else:
+                return datetime.min, "invalid_row_format"
+                
+        except Exception as e:
+            print(f"[SpaceActivityCache] Error reading {csv_file_path}: {e}")
+            return datetime.min, "read_error"
+    
+    def _save_cache_to_file(self):
+        """保存缓存到文件"""
+        try:
+            with open(self.cache_file_path, 'w', encoding='utf-8') as f:
+                json.dump(self._cache_data, f, indent=2, ensure_ascii=False)
+            print(f"[SpaceActivityCache] Cache saved to {self.cache_file_path}")
+        except Exception as e:
+            print(f"[SpaceActivityCache] Error saving cache to file: {e}")
+    
+    def _load_cache_from_file(self):
+        """从文件加载缓存"""
+        try:
+            if os.path.exists(self.cache_file_path):
+                with open(self.cache_file_path, 'r', encoding='utf-8') as f:
+                    data = json.load(f)
+                    print(f"[SpaceActivityCache] Cache loaded from file")
+                    return data
+        except Exception as e:
+            print(f"[SpaceActivityCache] Error loading cache from file: {e}")
+        return {}
+    
+    def get_space_activity(self, space_name):
+        """获取指定空间的活动信息"""
+        with CACHE_LOCK:
+            if not self._cache_data or 'spaces' not in self._cache_data:
+                # 缓存为空，尝试从文件加载
+                self._cache_data = self._load_cache_from_file()
+                if not self._cache_data:
+                    # 文件也没有，立即更新一次
+                    print("[SpaceActivityCache] No cache available, updating immediately...")
+                    self._update_cache()
+            
+            spaces_data = self._cache_data.get('spaces', {})
+            space_info = spaces_data.get(space_name, {})
+            
+            if not space_info:
+                return datetime.min, "not_in_cache"
+            
+            # 解析时间戳
+            last_activity_str = space_info.get('last_activity')
+            if last_activity_str:
+                try:
+                    last_activity = datetime.fromisoformat(last_activity_str)
+                except:
+                    last_activity = datetime.min
+            else:
+                last_activity = datetime.min
+            
+            status = space_info.get('status', 'unknown')
+            return last_activity, status
+    
+    def get_all_spaces_activity(self):
+        """获取所有空间的活动信息"""
+        with CACHE_LOCK:
+            if not self._cache_data or 'spaces' not in self._cache_data:
+                self._cache_data = self._load_cache_from_file()
+                if not self._cache_data:
+                    print("[SpaceActivityCache] No cache available, updating immediately...")
+                    self._update_cache()
+            
+            result = {}
+            spaces_data = self._cache_data.get('spaces', {})
+            
+            for space_name, space_info in spaces_data.items():
+                last_activity_str = space_info.get('last_activity')
+                if last_activity_str:
+                    try:
+                        last_activity = datetime.fromisoformat(last_activity_str)
+                    except:
+                        last_activity = datetime.min
+                else:
+                    last_activity = datetime.min
+                
+                result[space_name] = {
+                    'last_activity': last_activity,
+                    'status': space_info.get('status', 'unknown')
+                }
+            
+            return result
+    
+    def get_cache_info(self):
+        """获取缓存状态信息"""
+        with CACHE_LOCK:
+            if self._cache_data and 'last_updated' in self._cache_data:
+                last_updated_str = self._cache_data['last_updated']
+                try:
+                    last_updated = datetime.fromisoformat(last_updated_str)
+                    age_minutes = (datetime.now() - last_updated).total_seconds() / 60
+                    return {
+                        'last_updated': last_updated,
+                        'age_minutes': age_minutes,
+                        'spaces_count': len(self._cache_data.get('spaces', {})),
+                        'is_fresh': age_minutes < (CACHE_UPDATE_INTERVAL / 60) * 1.5  # 允许1.5倍的间隔
+                    }
+                except:
+                    pass
+            
+            return {
+                'last_updated': None,
+                'age_minutes': float('inf'),
+                'spaces_count': 0,
+                'is_fresh': False
+            }
+    
+    def force_update(self):
+        """强制立即更新缓存"""
+        print("[SpaceActivityCache] Force updating cache...")
+        self._update_cache()
+
+# 全局缓存实例
+activity_cache = None
+
+def init_activity_cache(repo_dir):
+    """初始化活动缓存"""
+    global activity_cache
+    if activity_cache is None:
+        activity_cache = SpaceActivityCache(repo_dir)
+        print("[init_activity_cache] Activity cache initialized")
+    return activity_cache
+
+def analyze_space_activity_cached(available_spaces, repo_dir):
+    """使用缓存的空间活动分析"""
+    global activity_cache
+    
+    # 确保缓存已初始化
+    if activity_cache is None:
+        activity_cache = init_activity_cache(repo_dir)
+    
+    space_activity = []
+    current_time = datetime.now()
+    
+    # 获取缓存信息
+    cache_info = activity_cache.get_cache_info()
+    print(f"[analyze_space_activity_cached] Using cache (age: {cache_info['age_minutes']:.1f} min, fresh: {cache_info['is_fresh']})")
+    
+    # 如果缓存太旧，可以选择强制更新（可选）
+    if not cache_info['is_fresh']:
+        print("[analyze_space_activity_cached] Cache is stale, but using it anyway for speed")
+        # 可以选择在后台触发更新：activity_cache.force_update()
+    
+    # 获取所有空间的缓存活动数据
+    all_spaces_activity = activity_cache.get_all_spaces_activity()
+    
+    # Read allocation records to filter out recently allocated spaces
+    active_allocations = read_allocation_records(repo_dir)
+    
+    print(f"[analyze_space_activity_cached] Analyzing {len(available_spaces)} spaces using cached data...")
+    
+    for space_name in available_spaces:
+        # 从缓存获取活动信息
+        cached_info = all_spaces_activity.get(space_name)
+        
+        if cached_info:
+            last_activity = cached_info['last_activity']
+            cached_status = cached_info['status']
+        else:
+            # 缓存中没有这个空间，可能是新空间
+            last_activity = datetime.min
+            cached_status = "not_in_cache"
+        
+        # Calculate idle time in minutes
+        if last_activity == datetime.min or 'empty' in cached_status or 'not_found' in cached_status:
+            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 (cached)"
+            last_activity_str = last_activity.strftime('%Y-%m-%d %H:%M:%S')
+        
+        # Check if space is recently allocated (remote)
+        is_recently_allocated_remote = space_name in active_allocations
+        if is_recently_allocated_remote:
+            alloc_info = active_allocations[space_name]
+            minutes_until_free = (alloc_info['expires_at'] - current_time).total_seconds() / 60
+            status += f" (Recently allocated remotely, free in {minutes_until_free:.1f} min)"
+        
+        # Check if space is recently allocated (local)
+        is_recently_allocated_local, local_student = local_tracker.is_recently_allocated_locally(space_name)
+        if is_recently_allocated_local:
+            status += f" (Recently allocated locally to {local_student})"
+        
+        space_activity.append({
+            'space_name': space_name,
+            'last_activity': last_activity,
+            'last_activity_str': last_activity_str,
+            'idle_minutes': idle_minutes,
+            'status': status,
+            'is_recently_allocated_remote': is_recently_allocated_remote,
+            'is_recently_allocated_local': is_recently_allocated_local,
+            'cached_status': cached_status
+        })
+        
+        print(f"[analyze_space_activity_cached] {space_name}: {status}")
+    
+    # Sort by idle time (most idle first)
+    space_activity.sort(key=lambda x: x['idle_minutes'], reverse=True)
+    
+    return space_activity
+
+
+
 # 新增：本地防撞车机制
 class LocalAllocationTracker:
     def __init__(self):
@@ -343,89 +710,7 @@ def simple_push_allocation_record(repo, space_name, student_id):
         print(f"[simple_push_allocation_record] Allocation recorded locally, sync will happen eventually")
         return False
 
-def get_last_activity_time_optimized(csv_file_path):
-    """优化版本：使用文件修改时间而不是读取文件内容"""
-    try:
-        if not os.path.exists(csv_file_path):
-            print(f"[get_last_activity_time_optimized] File not found: {csv_file_path}")
-            return datetime.min  # Never used
-        
-        # 获取文件的修改时间戳
-        file_mtime = os.path.getmtime(csv_file_path)
-        last_modified = datetime.fromtimestamp(file_mtime)
-        
-        print(f"[get_last_activity_time_optimized] {os.path.basename(csv_file_path)} last modified: {last_modified}")
-        return last_modified
-        
-    except Exception as e:
-        print(f"[get_last_activity_time_optimized] Error getting file mtime for {csv_file_path}: {e}")
-        return datetime.min
-
-def get_file_size_for_validation(csv_file_path):
-    """获取文件大小，用于验证文件是否有实际内容"""
-    try:
-        if not os.path.exists(csv_file_path):
-            return 0
-        return os.path.getsize(csv_file_path)
-    except:
-        return 0
 
-def analyze_space_activity_optimized(available_spaces, repo_dir):
-    """优化版本：使用文件修改时间分析空间活动"""
-    space_activity = []
-    current_time = datetime.now()
-    
-    # Read allocation records to filter out recently allocated spaces
-    active_allocations = read_allocation_records(repo_dir)
-    
-    print(f"[analyze_space_activity_optimized] Analyzing {len(available_spaces)} spaces using file modification time...")
-    
-    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_optimized(csv_file)
-        file_size = get_file_size_for_validation(csv_file)
-        
-        # Calculate idle time in minutes
-        if last_activity == datetime.min or file_size <= 100:  # 小于100字节认为是空文件或只有header
-            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')
-        
-        # Check if space is recently allocated (remote)
-        is_recently_allocated_remote = space_name in active_allocations
-        if is_recently_allocated_remote:
-            alloc_info = active_allocations[space_name]
-            minutes_until_free = (alloc_info['expires_at'] - current_time).total_seconds() / 60
-            status += f" (Recently allocated remotely, free in {minutes_until_free:.1f} min)"
-        
-        # Check if space is recently allocated (local)
-        is_recently_allocated_local, local_student = local_tracker.is_recently_allocated_locally(space_name)
-        if is_recently_allocated_local:
-            status += f" (Recently allocated locally to {local_student})"
-        
-        space_activity.append({
-            'space_name': space_name,
-            'last_activity': last_activity,
-            'last_activity_str': last_activity_str,
-            'idle_minutes': idle_minutes,
-            'status': status,
-            'is_recently_allocated_remote': is_recently_allocated_remote,
-            'is_recently_allocated_local': is_recently_allocated_local,
-            'file_size': file_size  # 添加文件大小信息用于调试
-        })
-        
-        print(f"[analyze_space_activity_optimized] {space_name}: {status} (file size: {file_size} bytes)")
-    
-    # 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 with proper line breaks"""
@@ -595,26 +880,30 @@ def redirect_to_space(redirect_url, selected_space, status_display):
     
     return gr.HTML(redirect_html)
 
-def load_balance_user(student_id):
-    """Main load balancing function with enhanced collision avoidance"""
-    print(f"[load_balance_user] Starting enhanced load balancing for student ID: {student_id}")
+def load_balance_user_final(student_id):
+    """最终版本的负载均衡函数"""
+    print(f"[load_balance_user_final] Starting cached 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)
     
+    # 初始化活动缓存
+    init_activity_cache(LOCAL_DATA_DIR)
+    
     # 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 (including both remote and local allocation records)
-    space_activity = analyze_space_activity_optimized(available_spaces, LOCAL_DATA_DIR)
+    # 使用缓存版本的分析函数
+    space_activity = analyze_space_activity_cached(available_spaces, LOCAL_DATA_DIR)
     
-    # Select space with enhanced collision avoidance
-    return select_space_with_enhanced_collision_avoidance(space_activity, student_id, repo)
+    # 使用缓存版本的选择函数
+    return select_space_with_enhanced_collision_avoidance_cached(space_activity, student_id, repo)
+
 
 def get_url_params(request: gr.Request):
     """Extract URL parameters from request"""
@@ -627,8 +916,8 @@ def get_url_params(request: gr.Request):
             return "Load Distributor", None
     return "Load Distributor", None
 
-def handle_user_access(request: gr.Request):
-    """Handle user access and perform load balancing"""
+def handle_user_access_final(request: gr.Request):
+    """最终版本的用户访问处理"""
     title, check_id = get_url_params(request)
     
     if not check_id:
@@ -645,9 +934,9 @@ def handle_user_access(request: gr.Request):
         """
         return title, gr.HTML(error_html)
     
-    # Valid student ID - perform load balancing
+    # Valid student ID - perform cached load balancing
     try:
-        result = load_balance_user(check_id)
+        result = load_balance_user_final(check_id)
         return title, result
     except Exception as e:
         # Handle any errors during load balancing
@@ -663,16 +952,52 @@ def handle_user_access(request: gr.Request):
         """
         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("")
+# 修改后的主Gradio界面
+def create_main_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("")
+        
+        # 添加隐藏的缓存信息显示（用于调试，默认不显示）
+        with gr.Accordion("🔧 System Information", open=False, visible=False):
+            cache_info = gr.JSON(label="Cache Status", visible=False)
+            
+            def update_cache_info():
+                return get_cache_status()
+            
+            # 每30秒更新一次缓存信息（如果界面是打开的）
+            interface.load(
+                fn=update_cache_info,
+                outputs=cache_info,
+                every=30
+            )
+        
+        # 主要的负载均衡逻辑
+        interface.load(
+            fn=handle_user_access_final,
+            outputs=[title_display, content_display]
+        )
     
-    # Initialize on page load
-    interface.load(
-        fn=handle_user_access,
-        outputs=[title_display, content_display]
-    )
+    return interface
 
+# 修改后的main函数
 if __name__ == "__main__":
-    interface.launch()
+    # 创建主界面
+    main_interface = create_main_interface()
+      
+    # 启动主界面
+    main_interface.launch()
+
+# 在程序退出时优雅地关闭后台线程
+import atexit
+
+def cleanup_on_exit():
+    """程序退出时的清理函数"""
+    global activity_cache
+    if activity_cache:
+        print("[cleanup_on_exit] Stopping background cache updates...")
+        activity_cache.stop_background_updates()
+
+# 注册退出时的清理函数
+atexit.register(cleanup_on_exit)