Update app.py

app.py

@@ -1,9 +1,8 @@
 from flask import Flask, render_template_string
 import requests
-import json
-from threading import Thread, Lock
+import threading
 import time
-from datetime import datetime, timedelta
+from concurrent.futures import ThreadPoolExecutor
 
 app = Flask(__name__)
 
@@ -227,25 +226,20 @@ htmlTemplate = f"""
 
   <script>
     const username = 'yangtb24';
-    let serversData = {{ servers_data|tojson }};  // 从 Flask 传入初始数据
+    const serversData = {{ servers_data|tojson }};
 
-    function updateServerCard(data, spaceId) {{
-      const serverId = data.replica;
-      const serverElement = document.getElementById(`server-${serverId}`);
-      const owner = username; // Simplified, since we know the owner
-
-      if (!serverElement) {{
+    function createServerCard(serverId, spaceId, owner) {{
         const card = document.createElement('div');
         card.id = `server-${serverId}`;
         card.className = 'server-card';
         card.innerHTML = `
           <div class="server-header">
             <div class="server-name">
-              <div class="status-dot status-online"></div>
+              <div class="status-dot status-offline"></div>
               <svg class="server-flag" width="20" height="20" viewBox="0 0 24 24" fill="currentColor">
                 <path d="M21 3H3C1.9 3 1 3.9 1 5v3c0 1.1.9 2 2 2h18c1.1 0 2-.9 2-2V5c0-1.1-.9-2-2-2zm-1 5H4V6h16v2zm1 4H3c-1.1 0-2 .9-2 2v3c0 1.1.9 2 2 2h18c1.1 0 2-.9 2-2v-3c0-1.1-.9-2-2-2zm-1 5H4v-2h16v2zm1 4H3c-1.1 0-2 .9-2 2v3c0 1.1.9 2 2 2h18c1.1 0 2-.9 2-2v-3c0-1.1-.9-2-2-2zm-1 5H4v-2h16v2z"/>
               </svg>
-              <div>${{serverId}} (${{owner}}/${{spaceId}})</div>
+              <div>${serverId} (${owner}/${spaceId})</div>
             </div>
           </div>
           <div class="metric-grid">
@@ -274,6 +268,15 @@ htmlTemplate = f"""
           </div>
         `;
         document.getElementById('servers').appendChild(card);
+    }}
+
+    function updateServerCard(data, spaceId) {{
+      const serverId = data.replica;
+      const serverElement = document.getElementById(`server-${serverId}`);
+      const owner = data.owner
+
+      if (!serverElement) {{
+          createServerCard(serverId, spaceId, owner);
       }}
 
       const card = document.getElementById(`server-${serverId}`);
@@ -290,182 +293,200 @@ htmlTemplate = f"""
 
       card.querySelector('.upload').textContent = `${formatBytes(uploadBps)}/s`;
       card.querySelector('.download').textContent = `${formatBytes(downloadBps)}/s`;
+      updateSummary(serverId, data);
     }}
-    
-        function updateSummary() {
-        let online = 0;
-        let offline = 0;
-        let totalUpload = 0;
-        let totalDownload = 0;
-        let totalServers = 0;
-
-        for (const spaceId in serversData) {
-            for (const replicaId in serversData[spaceId]) {
-                totalServers++;
-                const server = serversData[spaceId][replicaId];
-                const isOnline = server.isOnline;
-                const serverCard = document.getElementById(`server-${replicaId}`);
-
-
-                if (serverCard) {
-                    const statusDot = serverCard.querySelector('.status-dot');
-                    statusDot.className = `status-dot status-${isOnline ? 'online' : 'offline'}`;
-                }
-
-
-                if (isOnline) {
-                    online++;
-                    totalUpload += server.tx_bps;
-                    totalDownload += server.rx_bps;
-
-                } else {
-                    offline++;
-                }
-            }
-        }
 
-        document.getElementById('totalServers').textContent = totalServers;
-        document.getElementById('onlineServers').textContent = online;
-        document.getElementById('offlineServers').textContent = offline;
-        document.getElementById('totalUpload').textContent = `${formatBytes(totalUpload)}/s`;
-        document.getElementById('totalDownload').textContent = `${formatBytes(totalDownload)}/s`;
-    }
 
-    function formatBytes(bytes) {
+    function updateSummary(serverId, data) {{
+
+      const now = Date.now();
+      let online = 0;
+      let offline = 0;
+      let totalUpload = 0;
+      let totalDownload = 0;
+
+      for (const serverId in serversData) {{
+          const server = serversData[serverId];
+          const isOnline = server.isOnline;
+          const serverCard = document.getElementById(`server-${serverId}`);
+
+          if (serverCard) {{
+            const statusDot = serverCard.querySelector('.status-dot');
+            statusDot.className = `status-dot status-${isOnline ? 'online' : 'offline'}`;
+
+            if (isOnline && server.data) {{
+                const uploadText = formatBytes(server.data.tx_bps);
+                const downloadText = formatBytes(server.data.rx_bps);
+                totalUpload += server.data.tx_bps;
+                totalDownload += server.data.rx_bps;
+            }}
+          }}
+          isOnline ? online++ : offline++;
+      }}
+      document.getElementById('totalServers').textContent = Object.keys(serversData).length;
+      document.getElementById('onlineServers').textContent = online;
+      document.getElementById('offlineServers').textContent = offline;
+      document.getElementById('totalUpload').textContent = `${formatBytes(totalUpload)}/s`;
+      document.getElementById('totalDownload').textContent = `${formatBytes(totalDownload)}/s`;
+    }}
+
+
+
+    function formatBytes(bytes) {{
       if (bytes === 0) return '0 B';
       const k = 1024;
       const sizes = ['B', 'KB', 'MB', 'GB', 'TB'];
       const i = Math.floor(Math.log(bytes) / Math.log(k));
       return parseFloat((bytes / Math.pow(k, i)).toFixed(2)) + ' ' + sizes[i];
-    }
-    
-    //初始更新
-    for (const spaceId in serversData) {{
-      for (const replicaId in serversData[spaceId]) {{
-          const server = serversData[spaceId][replicaId];
-          updateServerCard(server, spaceId);
-      }}
+    }}
+
+
+    // Initial render based on initial data
+    for (const serverId in serversData) {{
+        const server = serversData[serverId];
+        if (server.data) {
+            updateServerCard(server.data, server.spaceId);
+        } else {
+            createServerCard(serverId, server.spaceId, server.owner)
+        }
     }}
     updateSummary();
 
-    // 定时更新
-    setInterval(() => {{
-      fetch('/update_data')
-        .then(response => response.json())
-        .then(data => {{
-          serversData = data.servers_data;
-          for (const spaceId in serversData) {{
-              for (const replicaId in serversData[spaceId]) {{
-                  const server = serversData[spaceId][replicaId];
-                  updateServerCard(server, spaceId);
-              }}
-          }}
-          updateSummary();
-        }});
-    }}, 2000);
+
   </script>
 </body>
 </html>
 """
 
 
-USERNAME = 'yangtb24'
-servers_data = {}  # {space_id: {replica_id: {metrics}, ...}, ...}
-data_lock = Lock()
-last_fetch_time = {}  # 记录每个space的上次获取时间
-FETCH_INTERVAL = 5  # 5秒获取间隔
-MAX_OFFLINE_TIME = 10  # 10秒判定为离线
-
-
-def fetch_instances(username):
-    try:
-        response = requests.get(f"https://huggingface.co/api/spaces?author={username}")
-        response.raise_for_status()  # 检查请求是否成功
-        user_instances = response.json()
-        return [{"id": instance["id"].split('/')[1], "owner": username} for instance in user_instances]
-    except requests.RequestException as e:
-        print(f"获取实例列表失败：{e}")
-        return []
-
-def fetch_metrics(username, space_id):
-    try:
-        response = requests.get(f"https://api.hf.space/v1/{username}/{space_id}/live-metrics/sse", stream=True)
-        response.raise_for_status()
-
-        for line in response.iter_lines():
-            if line:
-                decoded_line = line.decode('utf-8')
-                if decoded_line.startswith("event: metric"):
-                    try:
-                        data_str = decoded_line.split("data: ", 1)[1]
-                        data = json.loads(data_str)
-                        with data_lock:
-                            if space_id not in servers_data:
-                                servers_data[space_id] = {}
-                            data['isOnline'] = True #初始都设置为在线
-                            servers_data[space_id][data['replica']] = data
-                            last_fetch_time[space_id] = datetime.now()
-
-                    except json.JSONDecodeError as e:
-                        print(f"解析数据失败 ({space_id}): {e}")
-
-    except requests.RequestException as e:
-        print(f"连接失败 ({username}/{space_id}): {e}")
-        with data_lock:
-            # 如果请求失败,设置isOnline为False
-            if space_id in servers_data:
-                for replica_id in servers_data[space_id]:
-                    servers_data[space_id][replica_id]['isOnline'] = False
-
-def update_metrics():
-    """定期更新所有实例的指标"""
-    while True:
-        instances = fetch_instances(USERNAME)
-        threads = []
+class MetricsManager:
+    def __init__(self, username):
+        self.username = username
+        self.executor = ThreadPoolExecutor(max_workers=10)  # Adjust as needed
+        self.servers_data = {}
+        self.lock = threading.Lock()  # Protect shared data
+        self.stop_event = threading.Event()
+
+
+    def fetch_instances(self):
+        try:
+            response = requests.get(f"https://huggingface.co/api/spaces?author={self.username}")
+            response.raise_for_status()  # Raise an exception for bad status codes
+            user_instances = response.json()
+            return [
+                {"id": instance["id"].split("/")[1], "owner": self.username}
+                for instance in user_instances
+            ]
+        except requests.RequestException as e:
+            print(f"Error fetching instances: {e}")
+            return []
+
+    def connect(self, instance_id, username):
+        if self.stop_event.is_set():
+             return
+        url = f"https://api.hf.space/v1/{username}/{instance_id}/live-metrics/sse"
+        try:
+            # Use stream=True for streaming response
+            with requests.get(url, stream=True, timeout=(5, 60)) as response:  # Connect timeout, read timeout
+                response.raise_for_status()
+
+                if response.encoding is None:
+                    response.encoding = 'utf-8'
+
+                lines_iter = response.iter_lines(decode_unicode=True)
+
+                # Skip initial lines until we find the first "event: metric"
+                for line in lines_iter:
+                    if line.strip() == "event: metric":
+                        break
+
+                # Now process the actual metric data
+                for line in lines_iter:
+                    if self.stop_event.is_set():
+                        break
+                    if line.strip().startswith("data:"):
+                        try:
+                            data_str = line.strip()[5:]  # Remove "data:" prefix
+                            data = eval(data_str)  # Use eval (safer than direct JSON for this case)
+
+                            server_id = data['replica']
+                            with self.lock:
+                                if server_id not in self.servers_data:
+                                    self.servers_data[server_id] = {
+                                        'spaceId': instance_id,
+                                        'owner': username,
+                                        'data': None,
+                                        'last_seen': 0,
+                                        'isOnline': False,
+                                    }
+
+                                self.servers_data[server_id]['data'] = data
+                                self.servers_data[server_id]['last_seen'] = time.time()
+                                self.servers_data[server_id]['isOnline'] = True
+
+                        except Exception as e:
+                            print(f"Error parsing data for {instance_id}: {e}")
+                            print(f"  Problematic line: {line.strip()}") # Debugging
+
+        except requests.exceptions.RequestException as e:
+            print(f"Connection failed ({username}/{instance_id}): {e}")
+            # Mark all replicas of this space as offline
+            with self.lock:
+                for server_id, server_info in list(self.servers_data.items()): # Iterate on a copy
+                    if server_info['spaceId'] == instance_id:
+                      server_info['isOnline'] = False
+                      # Optionally remove after a longer timeout:
+                      # if time.time() - server_info['last_seen'] > 30:  # 30 seconds
+                      #     del self.servers_data[server_id]
+
+
+    def check_timeouts(self):
+        while not self.stop_event.is_set():
+            with self.lock:
+                now = time.time()
+                for server_id, server_info in list(self.servers_data.items()): # Iterate on a copy
+                    if now - server_info['last_seen'] > 10:
+                        server_info['isOnline'] = False
+                        # Optionally remove after a longer timeout:
+                        # if now - server_info['last_seen'] > 30: # 30 seconds
+                        #     del self.servers_data[server_id]
+            time.sleep(2)
+
+    def start_monitoring(self):
+        instances = self.fetch_instances()
         for instance in instances:
-            thread = Thread(target=fetch_metrics, args=(instance['owner'], instance['id']))
-            threads.append(thread)
-            thread.start()
-
-        for thread in threads:
-            thread.join()  # 等待所有线程完成,防止线程过多
+            self.executor.submit(self.connect, instance['id'], instance['owner'])
+        self.timeout_thread = threading.Thread(target=self.check_timeouts, daemon=True)
+        self.timeout_thread.start()
 
-        time.sleep(300)  # 每 5 分钟刷新一次实例列表
+    def stop_monitoring(self):
+        self.stop_event.set()
+        self.executor.shutdown(wait=False)
+        if hasattr(self, 'timeout_thread'):
+            self.timeout_thread.join()
 
+    def get_data(self):
+        with self.lock:
+            return self.servers_data.copy()
 
-# 启动指标更新线程
-update_thread = Thread(target=update_metrics)
-update_thread.daemon = True  # 设置为守护线程，主程序退出时自动退出
-update_thread.start()
 
-def check_online_status():
-    """定期检查服务器在线状态"""
-    while True:
-        now = datetime.now()
-        with data_lock:
-            for space_id in list(last_fetch_time.keys()):  # 使用list防止迭代时字典大小改变
-                if now - last_fetch_time[space_id] > timedelta(seconds=MAX_OFFLINE_TIME):
-                    if space_id in servers_data:
-                        for replica_id in servers_data[space_id]:
-                            servers_data[space_id][replica_id]['isOnline'] = False
-        time.sleep(2) #每2秒检查一次
+metrics_manager = MetricsManager(username='yangtb24')
 
+@app.route("/")
+def home():
+    servers_data = metrics_manager.get_data()
+    return render_template_string(htmlTemplate, servers_data=servers_data)
 
-# 启动在线状态检查线程
-check_status_thread = Thread(target=check_online_status)
-check_status_thread.daemon = True
-check_status_thread.start()
+# Initialize and start monitoring *before* running the Flask app
+metrics_manager.start_monitoring()
 
+# Important:  Use a context manager to cleanly shut down monitoring
+#             when the Flask app stops.  This prevents orphaned threads.
+@app.teardown_appcontext
+def shutdown_session(exception=None):
+    metrics_manager.stop_monitoring()
 
-@app.route('/')
-def index():
-    with data_lock:
-        return render_template_string(htmlTemplate, servers_data=servers_data)
 
-@app.route('/update_data')
-def update_data():
-    with data_lock:
-        return {'servers_data': servers_data}
+if __name__ == "__main__":
+    app.run(debug=False, host="0.0.0.0", port=7860) # Use port 7860 for HF Spaces
 
-if __name__ == '__main__':
-    app.run(debug=True, host="0.0.0.0", port=7860) # HF Spaces 默认端口是 7860