Spaces:

Greg-House
/

Simple-piping

Paused

Update app.py

3532d53 verified almost 2 years ago

14.8 kB

import asyncio
import aiohttp
import time
from datetime import datetime
from collections import deque, defaultdict
import json
import secrets
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

    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
        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.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}"
        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):
        self.stored_data.append({
            'timestamp': timestamp,
            'data': data,
            'formatted_time': datetime.fromtimestamp(timestamp).strftime('%Y-%m-%d %H:%M:%S')
        })
        
        if time.time() - self.last_cleanup > 60:
            self.cleanup_old_data()
            self.last_cleanup = time.time()

    async def start_reading(self):
        while True:
            try:
                async with aiohttp.ClientSession() as session:
                    if self.show_output:
                        print(f"Connecting to {self.url}...")
                    async with session.get(self.url) as response:
                        if response.status == 200:
                            if self.show_output:
                                print("Connected! Reading stream...")
                            
                            while True:
                                chunk = await response.content.read(1024)
                                if not chunk:
                                    break
                                
                                current_time = time.time()
                                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:
                    print(f"Connection error: {e}")
            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:
    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 = await self.stream_manager.create_stream(retention_seconds)
            
            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)

    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)
            
        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')

    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;
                    white-space: pre-wrap;
                    word-break: break-all;
                }}
                .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):
        html = 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; 
                }}
                .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() {{
                    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>
                <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"""
                <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")
    
    # 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...")
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nStopping server...")

	import asyncio
	import aiohttp
	import time
	from datetime import datetime
	from collections import deque, defaultdict
	import json
	import secrets
	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

	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
	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.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}"
	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):
	self.stored_data.append({
	'timestamp': timestamp,
	'data': data,
	'formatted_time': datetime.fromtimestamp(timestamp).strftime('%Y-%m-%d %H:%M:%S')
	})

	if time.time() - self.last_cleanup > 60:
	self.cleanup_old_data()
	self.last_cleanup = time.time()

	async def start_reading(self):
	while True:
	try:
	async with aiohttp.ClientSession() as session:
	if self.show_output:
	print(f"Connecting to {self.url}...")
	async with session.get(self.url) as response:
	if response.status == 200:
	if self.show_output:
	print("Connected! Reading stream...")

	while True:
	chunk = await response.content.read(1024)
	if not chunk:
	break

	current_time = time.time()
	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:
	print(f"Connection error: {e}")
	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:
	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 = await self.stream_manager.create_stream(retention_seconds)

	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)

	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)

	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')

	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;
	white-space: pre-wrap;
	word-break: break-all;
	}}
	.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):
	html = 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;
	}}
	.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() {{
	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>
	<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"""
	<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")

	# 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...")
	try:
	asyncio.run(main())
	except KeyboardInterrupt:
	print("\nStopping server...")