Update app.py

app.py

@@ -5,34 +5,58 @@ from datetime import datetime
 from collections import deque, defaultdict
 import json
 import secrets
-import re
 from aiohttp import web
 
 class StreamManager:
     def __init__(self):
         self.streams = {}  # Dictionary to store all stream instances
         self.tokens = {}   # Dictionary to store stream tokens
+        self.tasks = {}    # Dictionary to store stream tasks
 
-    def create_stream(self, retention_seconds):
+    async def create_stream(self, retention_seconds):
         # Generate unique stream ID and token
         stream_id = secrets.token_urlsafe(8)
         access_token = secrets.token_urlsafe(16)
         
         # Create new stream instance
-        self.streams[stream_id] = EphemeralStreamReader(
+        stream = EphemeralStreamReader(
             path=stream_id, 
             retention_seconds=retention_seconds,
             show_output=True
         )
+        
+        self.streams[stream_id] = stream
         self.tokens[stream_id] = access_token
         
+        # Start the stream reader task
+        self.tasks[stream_id] = asyncio.create_task(stream.start_reading())
+        
         return stream_id, access_token
 
     def get_stream(self, stream_id, token):
-        if stream_id in self.streams and self.tokens[stream_id] == token:
+        if stream_id in self.streams and self.tokens.get(stream_id) == token:
             return self.streams[stream_id]
         return None
 
+    async def cleanup_inactive_streams(self):
+        while True:
+            current_time = time.time()
+            inactive_streams = []
+            
+            for stream_id, stream in self.streams.items():
+                # Clean up streams that haven't been accessed in 1 hour
+                if current_time - stream.last_access > 3600:
+                    inactive_streams.append(stream_id)
+            
+            for stream_id in inactive_streams:
+                if stream_id in self.tasks:
+                    self.tasks[stream_id].cancel()
+                    del self.tasks[stream_id]
+                del self.streams[stream_id]
+                del self.tokens[stream_id]
+            
+            await asyncio.sleep(300)  # Check every 5 minutes
+
 class EphemeralStreamReader:
     def __init__(self, piping_server_url="https://ppng.io", path="test123", retention_seconds=3600, show_output=False):
         self.url = f"{piping_server_url}/{path}"
@@ -94,7 +118,8 @@ class EphemeralStreamReader:
             except Exception as e:
                 if self.show_output:
                     print(f"Error: {e}")
-                
+            
+            # Always wait a bit before reconnecting
             await asyncio.sleep(self.reconnect_delay)
 
 class WebServer:
@@ -107,9 +132,10 @@ class WebServer:
             retention_minutes = float(data.get('retention_minutes', '60'))
             retention_seconds = min(3600, max(0, retention_minutes * 60))
             
-            stream_id, token = self.stream_manager.create_stream(retention_seconds)
+            stream_id, token = await self.stream_manager.create_stream(retention_seconds)
             
-            return web.Response(text=self.create_stream_page(stream_id, token), content_type='text/html')
+            html = self.create_stream_page(stream_id, token)
+            return web.Response(text=html, content_type='text/html')
             
         except Exception as e:
             return web.Response(text=f"Error creating stream: {str(e)}", status=400)
@@ -122,7 +148,8 @@ class WebServer:
         if not stream:
             return web.Response(text="Invalid stream ID or token", status=403)
             
-        return web.Response(text=self.create_viewer_page(stream, token), content_type='text/html')
+        html = self.create_viewer_page(stream, token)
+        return web.Response(text=html, content_type='text/html')
 
     async def handle_home(self, request):
         return web.Response(text=self.create_home_page(), content_type='text/html')
@@ -224,6 +251,8 @@ class WebServer:
                     padding: 15px;
                     border-radius: 4px;
                     font-family: monospace;
+                    white-space: pre-wrap;
+                    word-break: break-all;
                 }}
                 .button {{
                     display: inline-block;
@@ -263,7 +292,7 @@ class WebServer:
         """
 
     def create_viewer_page(self, stream, token):
-        return f"""
+        html = f"""
         <!DOCTYPE html>
         <html>
         <head>
@@ -298,6 +327,16 @@ class WebServer:
                     margin-bottom: 20px; 
                     color: #666; 
                 }}
+                .code-box {{
+                    background: #2b2b2b;
+                    color: #ffffff;
+                    padding: 15px;
+                    border-radius: 4px;
+                    font-family: monospace;
+                    margin: 10px 0;
+                    white-space: pre-wrap;
+                    word-break: break-all;
+                }}
             </style>
             <script>
                 function refreshData() {{
@@ -313,11 +352,12 @@ class WebServer:
             <div class="info">
                 <p>Stream ID: {stream.path}</p>
                 <p>Retention Time: {stream.retention_seconds / 60:.1f} minutes</p>
-                <p>To write to this stream: <code>seq inf | curl -T- {stream.url}</code></p>
+                <div class="code-box">To write to this stream: seq inf | curl -T- {stream.url}</div>
             </div>
             <button class="refresh-btn" onclick="refreshData()">Refresh Data</button>
             <div class="data-container">
         """
+        
         data = stream.get_stored_data()
         for item in reversed(data):
             html += f"""
@@ -350,8 +390,13 @@ async def main():
     
     print("\nWeb interface available at http://0.0.0.0:7860")
     
-    while True:
-        await asyncio.sleep(3600)  # Keep the server running
+    # Start the cleanup task
+    cleanup_task = asyncio.create_task(stream_manager.cleanup_inactive_streams())
+    
+    try:
+        await asyncio.Event().wait()  # Keep the server running indefinitely
+    finally:
+        cleanup_task.cancel()
 
 if __name__ == "__main__":
     print("Starting Ephemeral Stream Server...")