Spaces:

Greg-House
/

Simple-piping

Running

App Files Files Community

Greg-House commited on Nov 14, 2024

Commit

d34c396

verified ·

1 Parent(s): 5027087

Update app.py

Browse files

Files changed (1) hide show

app.py +281 -98

app.py CHANGED Viewed

@@ -2,28 +2,56 @@ import asyncio
 import aiohttp
 import time
 from datetime import datetime
-from collections import deque
 import json
 from aiohttp import web
 class EphemeralStreamReader:
-    def __init__(self, piping_server_url="https://ppng.io", path="test123", retention_hours=1, show_output=False):
         self.url = f"{piping_server_url}/{path}"
         self.path = path
         self.reconnect_delay = 1
         self.show_output = show_output
-        self.retention_hours = retention_hours
         self.stored_data = deque()
         self.last_cleanup = time.time()
     def cleanup_old_data(self):
         current_time = time.time()
-        retention_seconds = self.retention_hours * 3600
-        while self.stored_data and (current_time - self.stored_data[0]['timestamp']) > retention_seconds:
             self.stored_data.popleft()
     def get_stored_data(self):
         self.cleanup_old_data()
         return list(self.stored_data)
     def store_chunk(self, data, timestamp):
@@ -57,14 +85,8 @@ class EphemeralStreamReader:
                                 try:
                                     text = chunk.decode('utf-8')
                                     self.store_chunk(text, current_time)
-                                    if self.show_output:
-                                        timestamp = datetime.fromtimestamp(current_time).strftime('%Y-%m-%d %H:%M:%S')
-                                        print(f"[{timestamp}] Received: {text}", end='', flush=True)
                                 except UnicodeDecodeError:
                                     self.store_chunk(str(chunk), current_time)
-                                    if self.show_output:
-                                        print(f"Received raw bytes: {chunk}")
             except aiohttp.ClientError as e:
                 if self.show_output:
@@ -73,105 +95,266 @@ class EphemeralStreamReader:
                 if self.show_output:
                     print(f"Error: {e}")
-            if self.show_output:
-                print(f"Connection closed. Reconnecting in {self.reconnect_delay} seconds...")
             await asyncio.sleep(self.reconnect_delay)
-    async def start_http_server(self, host='0.0.0.0', port=7860):
-        async def handle_get(request):
-            data = self.get_stored_data()
-            html = f"""
-            <!DOCTYPE html>
-            <html>
-            <head>
-                <title>Ephemeral Stream Data</title>
-                <style>
-                    body {{
-                        font-family: Arial, sans-serif;
-                        margin: 20px;
-                    }}
-                    .data-container {{
-                        margin: 20px 0;
-                    }}
-                    .timestamp {{
-                        color: #666;
-                        font-size: 0.9em;
-                    }}
-                    .data-item {{
-                        margin: 10px 0;
-                        padding: 10px;
-                        background: #f5f5f5;
-                        border-radius: 4px;
-                    }}
-                    .refresh-btn {{
-                        padding: 10px 20px;
-                        background: #4CAF50;
-                        color: white;
-                        border: none;
-                        border-radius: 4px;
-                        cursor: pointer;
-                    }}
-                    .info {{
-                        margin-bottom: 20px;
-                        color: #666;
-                    }}
-                </style>
-                <script>
-                    function refreshData() {{
-                        window.location.reload();
-                    }}
-                    // Auto refresh every 5 seconds
-                    setInterval(refreshData, 5000);
-                </script>
-            </head>
-            <body>
-                <h1>Ephemeral Stream Data</h1>
-                <div class="info">
-                    <p>Stream Path: {self.path}</p>
-                    <p>Retention Time: {self.retention_hours} hours</p>
-                    <p>To write to this stream: <code>seq inf | curl -T- {self.url}</code></p>
-                </div>
-                <button class="refresh-btn" onclick="refreshData()">Refresh Data</button>
-                <div class="data-container">
-            """
-            for item in reversed(data):
-                html += f"""
-                    <div class="data-item">
-                        <div class="timestamp">{item['formatted_time']}</div>
-                        <pre>{item['data']}</pre>
-                    </div>
-                """
-            html += """
-                </div>
-            </body>
-            </html>
-            """
-            return web.Response(text=html, content_type='text/html')
-        app = web.Application()
-        app.router.add_get('/', handle_get)
-        runner = web.AppRunner(app)
-        await runner.setup()
-        site = web.TCPSite(runner, host, port)
-        await site.start()
-        print(f"\nWeb interface available at http://{host}:{port}")
-        print(f"To write to this stream: seq inf | curl -T- {self.url}")
 async def main():
-    reader = EphemeralStreamReader(retention_hours=1, show_output=True)
-    await asyncio.gather(
-        reader.start_reading(),
-        reader.start_http_server(host='0.0.0.0', port=7860)
-    )
 if __name__ == "__main__":
-    print("Starting Ephemeral Stream Reader...")
     try:
         asyncio.run(main())
     except KeyboardInterrupt:

 import aiohttp
 import time
 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
+    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(
+            path=stream_id,
+            retention_seconds=retention_seconds,
+            show_output=True
+        )
+        self.tokens[stream_id] = access_token
+        return stream_id, access_token
+    def get_stream(self, stream_id, token):
+        if stream_id in self.streams and self.tokens[stream_id] == token:
+            return self.streams[stream_id]
+        return None
 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}"
         self.path = path
         self.reconnect_delay = 1
         self.show_output = show_output
+        self.retention_seconds = min(3600, max(0, retention_seconds))  # Limit between 0 and 3600 seconds
         self.stored_data = deque()
         self.last_cleanup = time.time()
+        self.last_access = time.time()
     def cleanup_old_data(self):
         current_time = time.time()
+        while self.stored_data and (current_time - self.stored_data[0]['timestamp']) > self.retention_seconds:
             self.stored_data.popleft()
     def get_stored_data(self):
         self.cleanup_old_data()
+        self.last_access = time.time()
         return list(self.stored_data)
     def store_chunk(self, data, timestamp):
                                 try:
                                     text = chunk.decode('utf-8')
                                     self.store_chunk(text, current_time)
                                 except UnicodeDecodeError:
                                     self.store_chunk(str(chunk), current_time)
             except aiohttp.ClientError as e:
                 if self.show_output:
                 if self.show_output:
                     print(f"Error: {e}")
             await asyncio.sleep(self.reconnect_delay)
+class WebServer:
+    def __init__(self, stream_manager):
+        self.stream_manager = stream_manager
+    async def handle_create_stream(self, request):
+        try:
+            data = await request.post()
+            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)
+            return web.Response(text=self.create_stream_page(stream_id, token), content_type='text/html')
+        except Exception as e:
+            return web.Response(text=f"Error creating stream: {str(e)}", status=400)
+    async def handle_view_stream(self, request):
+        stream_id = request.match_info.get('stream_id')
+        token = request.query.get('token')
+        stream = self.stream_manager.get_stream(stream_id, token)
+        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')
+    async def handle_home(self, request):
+        return web.Response(text=self.create_home_page(), content_type='text/html')
+    def create_home_page(self):
+        return """
+        <!DOCTYPE html>
+        <html>
+        <head>
+            <title>Create Ephemeral Stream</title>
+            <style>
+                body {
+                    font-family: Arial, sans-serif;
+                    margin: 20px;
+                    max-width: 800px;
+                    margin: 0 auto;
+                    padding: 20px;
+                }
+                .form-container {
+                    margin: 20px 0;
+                    padding: 20px;
+                    border: 1px solid #ddd;
+                    border-radius: 4px;
+                }
+                .input-group {
+                    margin: 10px 0;
+                }
+                input, button {
+                    padding: 8px;
+                    margin: 5px 0;
+                }
+                button {
+                    background: #4CAF50;
+                    color: white;
+                    border: none;
+                    border-radius: 4px;
+                    cursor: pointer;
+                }
+                .info {
+                    margin: 20px 0;
+                    padding: 15px;
+                    background: #f5f5f5;
+                    border-radius: 4px;
+                }
+            </style>
+        </head>
+        <body>
+            <h1>Create New Ephemeral Stream</h1>
+            <div class="form-container">
+                <form method="POST" action="/create">
+                    <div class="input-group">
+                        <label for="retention_minutes">Retention Time (minutes, max 60):</label><br>
+                        <input type="number" id="retention_minutes" name="retention_minutes"
+                               min="0" max="60" value="60" step="0.5" required>
+                    </div>
+                    <button type="submit">Create Stream</button>
+                </form>
+            </div>
+            <div class="info">
+                <h3>About Ephemeral Streams</h3>
+                <p>Create your private stream with custom retention time. Each stream:</p>
+                <ul>
+                    <li>Has a unique ID and access token</li>
+                    <li>Can only be viewed with the correct token</li>
+                    <li>Automatically deletes data older than the specified retention time</li>
+                    <li>Supports retention times from 0 to 60 minutes</li>
+                </ul>
+            </div>
+        </body>
+        </html>
+        """
+    def create_stream_page(self, stream_id, token):
+        view_url = f"/view/{stream_id}?token={token}"
+        write_url = f"https://ppng.io/{stream_id}"
+        return f"""
+        <!DOCTYPE html>
+        <html>
+        <head>
+            <title>Stream Created</title>
+            <style>
+                body {{
+                    font-family: Arial, sans-serif;
+                    margin: 20px;
+                    max-width: 800px;
+                    margin: 0 auto;
+                    padding: 20px;
+                }}
+                .info-box {{
+                    background: #f5f5f5;
+                    padding: 20px;
+                    border-radius: 4px;
+                    margin: 20px 0;
+                }}
+                .code-box {{
+                    background: #2b2b2b;
+                    color: #ffffff;
+                    padding: 15px;
+                    border-radius: 4px;
+                    font-family: monospace;
+                }}
+                .button {{
+                    display: inline-block;
+                    padding: 10px 20px;
+                    background: #4CAF50;
+                    color: white;
+                    text-decoration: none;
+                    border-radius: 4px;
+                    margin: 10px 0;
+                }}
+            </style>
+        </head>
+        <body>
+            <h1>Stream Created Successfully</h1>
+            <div class="info-box">
+                <h3>Your Stream Information</h3>
+                <p><strong>Stream ID:</strong> {stream_id}</p>
+                <p><strong>Access Token:</strong> {token}</p>
+                <p><strong>View URL:</strong> <a href="{view_url}">{view_url}</a></p>
+                <h3>How to Write to Your Stream</h3>
+                <p>Use curl to write to your stream:</p>
+                <div class="code-box">
+                    seq inf | curl -T- {write_url}
+                </div>
+                <h3>Important Notes</h3>
+                <ul>
+                    <li>Keep your access token secret</li>
+                    <li>Bookmark the view URL for easy access</li>
+                    <li>Data older than your specified retention time will be automatically deleted</li>
+                </ul>
+            </div>
+            <a href="{view_url}" class="button">View Your Stream</a>
+        </body>
+        </html>
+        """
+    def create_viewer_page(self, stream, token):
+        return f"""
+        <!DOCTYPE html>
+        <html>
+        <head>
+            <title>Stream Viewer</title>
+            <style>
+                body {{
+                    font-family: Arial, sans-serif;
+                    margin: 20px;
+                }}
+                .data-container {{
+                    margin: 20px 0;
+                }}
+                .timestamp {{
+                    color: #666;
+                    font-size: 0.9em;
+                }}
+                .data-item {{
+                    margin: 10px 0;
+                    padding: 10px;
+                    background: #f5f5f5;
+                    border-radius: 4px;
+                }}
+                .refresh-btn {{
+                    padding: 10px 20px;
+                    background: #4CAF50;
+                    color: white;
+                    border: none;
+                    border-radius: 4px;
+                    cursor: pointer;
+                }}
+                .info {{
+                    margin-bottom: 20px;
+                    color: #666;
+                }}
+            </style>
+            <script>
+                function refreshData() {{
+                    window.location.reload();
+                }}
+                // Auto refresh every 5 seconds
+                setInterval(refreshData, 5000);
+            </script>
+        </head>
+        <body>
+            <h1>Stream Viewer</h1>
+            <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>
+            <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"""
+                <div class="data-item">
+                    <div class="timestamp">{item['formatted_time']}</div>
+                    <pre>{item['data']}</pre>
+                </div>
+            """
+        html += """
+            </div>
+        </body>
+        </html>
+        """
+        return html
 async def main():
+    stream_manager = StreamManager()
+    web_server = WebServer(stream_manager)
+    app = web.Application()
+    app.router.add_get('/', web_server.handle_home)
+    app.router.add_post('/create', web_server.handle_create_stream)
+    app.router.add_get('/view/{stream_id}', web_server.handle_view_stream)
+    runner = web.AppRunner(app)
+    await runner.setup()
+    site = web.TCPSite(runner, '0.0.0.0', 7860)
+    await site.start()
+    print("\nWeb interface available at http://0.0.0.0:7860")
+    while True:
+        await asyncio.sleep(3600)  # Keep the server running
 if __name__ == "__main__":
+    print("Starting Ephemeral Stream Server...")
     try:
         asyncio.run(main())
     except KeyboardInterrupt: