Update app.py

app.py

@@ -1,174 +1,98 @@
-import logging
-import aiohttp
-import asyncio
 import gradio as gr
+import asyncio, aiohttp
+import threading, logging
 from datetime import datetime
-import pytz
-import threading
-from typing import Dict, Tuple
-
-# 配置日志系统
-logging.basicConfig(
-    level=logging.INFO,
-    format='%(asctime)s - %(levelname)s - %(message)s',
-    handlers=[
-        logging.StreamHandler()
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+URLS = [
+        'https://asfag654-g.hf.space',
+        'https://asfag654-libretranslate.hf.space',
+        'https://asfag654-kk.hf.space',
+        'https://asfag654-t.hf.space',
+        'https://asfag654-b.hf.space',
+        'https://imluochen-sleepy.hf.space',
+        'https://7d820-ss.hf.space',
+        'https://7d820-kk.hf.space'
     ]
-)
-logger = logging.getLogger("URLMonitor")
 
-class URLMonitor:
-    def __init__(self, urls: list, check_interval: int = 3600):
+class Monitor:
+    def __init__(self, urls):
         self.urls = urls
-        self.check_interval = check_interval
-        self.timezone = pytz.timezone('Asia/Shanghai')
-        self.status: Dict[str, Dict] = {url: {
-            'status_code': None,
-            'message': '等待首次检查',
-            'last_checked': None
-        } for url in urls}
-        self.lock = threading.Lock()
-        self._running = False
-
-    def get_current_time(self) -> str:
-        """获取当前格式化时间"""
-        return datetime.now(self.timezone).strftime('%m-%d %H:%M')
-    
-    async def check_single_url(self, session: aiohttp.ClientSession, url: str):
-        """检查单个URL状态"""
+        self.status = {url: [] for url in urls}
+        self.check_interval = 3600
+        self._stop_event = threading.Event()
+
+    def record(self, url, success):
+        now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+        self.status[url].append((now, success))
+        if len(self.status[url]) > 100:
+            self.status[url] = self.status[url][-100:]
+
+    def get_current(self):
+        return [[url, *([f"{t} ✅" if s else f"{t} ❌"][-1:])] for url, logs in self.status.items() for t, s in logs[-1:]]
+
+    def get_stats(self, selected=None):
+        table = []
+        for url, logs in self.status.items():
+            if selected and selected not in url: continue
+            total = len(logs)
+            ok = sum(1 for _, s in logs if s)
+            pct = f"{ok / total:.1%}" if total else "N/A"
+            table.append([url, total, ok, pct])
+        return sorted(table, key=lambda x: -x[2])
+
+    async def check_url(self, url):
         try:
-            start_time = datetime.now()
-            async with session.get(url, timeout=10) as response:
-                status_code = response.status
-                message = "服务正常" if status_code == 200 else f"服务异常 (HTTP {status_code})"
-                latency = (datetime.now() - start_time).total_seconds()
-                logger.info(f"{url} 状态: {status_code} 响应时间: {latency:.2f}s")
-                
-        except aiohttp.ClientError as e:
-            status_code = None
-            message = f"网络错误: {type(e).__name__}"
-            logger.warning(f"{url} 网络错误: {str(e)}")
-        except asyncio.TimeoutError:
-            status_code = None
-            message = "请求超时 (10秒)"
-            logger.warning(f"{url} 请求超时")
-        except Exception as e:
-            status_code = None
-            message = f"未知错误: {type(e).__name__}"
-            logger.error(f"{url} 未知错误: {str(e)}", exc_info=True)
-        
-        timestamp = self.get_current_time()
-        with self.lock:
-            self.status[url] = {
-                'status_code': status_code,
-                'message': f"{message} - 最后检查: {timestamp}",
-                'last_checked': timestamp
-            }
-
-    async def run_checks(self):
-        """执行所有URL检查"""
-        async with aiohttp.ClientSession() as session:
-            while self._running:
-                tasks = [self.check_single_url(session, url) for url in self.urls]
-                await asyncio.gather(*tasks)
-                logger.info(f"本轮检查完成，{self.check_interval}秒后重新检查")
-                await asyncio.sleep(self.check_interval)
+            async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(10)) as session:
+                async with session.get(url):
+                    self.record(url, True)
+        except:
+            self.record(url, False)
+
+    async def run_check(self):
+        await asyncio.gather(*(self.check_url(url) for url in self.urls))
 
     def start(self):
-        """启动监控后台任务"""
-        if not self._running:
-            self._running = True
-            threading.Thread(
-                target=self._run_async_tasks, 
-                daemon=True,
-                name="MonitorThread"
-            ).start()
-            logger.info("URL监控服务已启动")
-
-    def stop(self):
-        """停止监控服务"""
-        self._running = False
-        logger.info("URL监控服务已停止")
-
-    def _run_async_tasks(self):
-        """在新事件循环中运行异步任务"""
-        loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(loop)
-        loop.run_until_complete(self.run_checks())
-
-    def get_status(self) -> Dict[str, str]:
-        """获取当前状态信息"""
-        with self.lock:
-            return {url: info['message'] for url, info in self.status.items()}
-
-
-def create_gradio_interface(monitor: URLMonitor):
-    """创建Gradio监控界面"""
-    theme = gr.themes.Soft(
-        primary_hue="blue",
-        secondary_hue="gray"
-    )
-    
-    with gr.Blocks(theme=theme, title="URL状态监控面板") as demo:
-        gr.Markdown("# 🖥️ URL服务状态监控")
-        gr.Markdown("> 实时监控多个服务的可用性状态")
-        
-        with gr.Row():
-            refresh_btn = gr.Button("🔄 刷新状态", variant="primary")
-            interval_slider = gr.Slider(
-                minimum=60, maximum=86400, value=3600, step=300,
-                label="检查频率 (秒)", interactive=True
-            )
-        
-        status_table = gr.Dataframe(
-            headers=["URL", "状态信息"],
-            datatype=["str", "str"],
-            interactive=False,
-            label="服务状态"
-        )
-        
-        # 组件交互
-        refresh_btn.click(
-            fn=lambda: [[url, info] for url, info in monitor.get_status().items()],
-            outputs=status_table
-        )
-        
-        interval_slider.change(
-            fn=lambda x: setattr(monitor, "check_interval", x),
-            inputs=interval_slider
-        )
-        
-        # 初始加载数据
-        demo.load(
-            fn=lambda: [[url, info] for url, info in monitor.get_status().items()],
-            outputs=status_table
-        )
-    
-    return demo
-
-
-if __name__ == "__main__":
-    # 配置监控URL列表
-    monitored_urls = [
-        'https://asfag654-g.hf.space',
-        'https://asfag654-t.hf.space',
-        'https://imluochen-sleepy.hf.space'
-    ]
-    
-    # 初始化监控器
-    monitor = URLMonitor(monitored_urls, check_interval=3600)
-    monitor.start()
-    
-    try:
-        # 创建并启动Gradio界面
-        interface = create_gradio_interface(monitor)
-        interface.launch(
-            server_name="0.0.0.0",
-            server_port=7860,
-            show_error=True,
-            favicon_path="https://www.gstatic.com/devrel-devsite/prod/vc8928e4803ba00823d2deb39c327db4919db629f0ecb648b587b35b95297bc3f/tensorflow/images/favicon.png"
-        )
-    except KeyboardInterrupt:
-        monitor.stop()
-    finally:
-        monitor.stop()
+        def loop():
+            while not self._stop_event.is_set():
+                asyncio.run(self.run_check())
+                self._stop_event.wait(self.check_interval)
+        threading.Thread(target=loop, daemon=True).start()
+
+monitor = Monitor(URLS)
+monitor.start()
+
+def refresh_all():
+    logger.info("刷新状态")
+    return monitor.get_current(), monitor.get_stats()
+
+def get_filtered_stats(sel):
+    return monitor.get_stats(sel)
+
+with gr.Blocks(title="服务状态监控") as demo:
+    gr.Markdown("## ✅ 状态监控\n每小时自动检测服务状态。")
+
+    status_table = gr.Dataframe(headers=["URL", "最近状态"], interactive=False)
+    stats_table = gr.Dataframe(headers=["URL", "检测数", "成功", "成功率"], interactive=False)
+
+    with gr.Row():
+        sel = gr.Dropdown(choices=[u.split('//')[1].split('.')[0] for u in URLS], label="筛选")
+        btn = gr.Button("🔄 刷新状态")
+
+    sel.change(get_filtered_stats, inputs=sel, outputs=stats_table)
+    btn.click(refresh_all, outputs=[status_table, stats_table])
+
+    demo.load(refresh_all, outputs=[status_table, stats_table])
+
+    gr.HTML("""
+    <script>
+      setInterval(() => {
+        const btn = [...document.querySelectorAll('button')].find(b => b.innerText.includes('刷新'));
+        if (btn) btn.click();
+      }, 30000);
+    </script>
+    """)
+
+demo.launch()