134 lines
3.9 KiB
Python
134 lines
3.9 KiB
Python
import http.server
|
|
import random
|
|
import socketserver
|
|
import sys
|
|
import threading
|
|
import time
|
|
import typing
|
|
import webbrowser
|
|
|
|
# language=html
|
|
inject_script = """
|
|
<script>
|
|
// Send periodic heartbeat
|
|
let heartbeatInterval = setInterval(() => {
|
|
fetch('/heartbeat', {method: 'POST'})
|
|
.catch(() => {}); // Ignore errors
|
|
}, 1000);
|
|
|
|
// Multiple event handlers for different close scenarios
|
|
function sendShutdown() {
|
|
clearInterval(heartbeatInterval);
|
|
// Synchronous request to ensure it gets sent
|
|
const xhr = new XMLHttpRequest();
|
|
xhr.open('POST', '/shutdown', false); // false makes it synchronous
|
|
try {
|
|
xhr.send();
|
|
} catch (e) {}
|
|
}
|
|
|
|
window.addEventListener('beforeunload', sendShutdown);
|
|
window.addEventListener('unload', sendShutdown);
|
|
// setTimeout(sendShutdown, 5000);
|
|
|
|
// Handle visibility change (tab hidden/visible)
|
|
document.addEventListener('visibilitychange', () => {
|
|
if (document.visibilityState === 'hidden') {
|
|
sendShutdown();
|
|
}
|
|
});
|
|
</script>
|
|
"""
|
|
|
|
|
|
def handler_for_html(html: str) -> typing.Type[http.server.SimpleHTTPRequestHandler]:
|
|
class Handler(http.server.SimpleHTTPRequestHandler):
|
|
def do_GET(self):
|
|
if self.path == "/":
|
|
self.send_response(200)
|
|
self.send_header("Content-type", "text/html")
|
|
self.end_headers()
|
|
self.wfile.write(html.encode())
|
|
else:
|
|
self.send_error(404)
|
|
|
|
def do_POST(self):
|
|
if self.path == "/shutdown":
|
|
self.send_response(200)
|
|
self.end_headers()
|
|
self.server.last_heartbeat = 0 # Force shutdown
|
|
self.server.shutdown()
|
|
elif self.path == "/heartbeat":
|
|
self.send_response(200)
|
|
self.end_headers()
|
|
self.server.last_heartbeat = time.time()
|
|
else:
|
|
self.send_error(404)
|
|
|
|
def log_message(self, format, *args):
|
|
# Suppress logging
|
|
pass
|
|
|
|
return Handler
|
|
|
|
|
|
class ServerWithHeartbeat(socketserver.TCPServer):
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self.last_heartbeat = time.time()
|
|
|
|
|
|
def monitor_heartbeat(server: ServerWithHeartbeat, timeout: int = 2):
|
|
while True:
|
|
time.sleep(0.5)
|
|
if time.time() - server.last_heartbeat > timeout:
|
|
print("No heartbeat received, shutting down...")
|
|
server.shutdown()
|
|
break
|
|
|
|
|
|
def run_server(handler: http.server.SimpleHTTPRequestHandler, port: int = 8000):
|
|
with ServerWithHeartbeat(("", port), handler) as httpd:
|
|
print(f"Serving at port {port}")
|
|
|
|
# Start heartbeat monitor in separate thread
|
|
monitor_thread = threading.Thread(target=monitor_heartbeat, args=(httpd,))
|
|
monitor_thread.daemon = True
|
|
monitor_thread.start()
|
|
|
|
httpd.serve_forever()
|
|
|
|
|
|
def main():
|
|
html = sys.stdin.read()
|
|
if "<body>" in html:
|
|
html_to_display = html.replace("</body>", f"{inject_script}</body>")
|
|
else:
|
|
html_to_display = f"<html><body>{html}{inject_script}</body></html>"
|
|
|
|
handler = handler_for_html(html_to_display)
|
|
|
|
# Start the server in a separate thread
|
|
port = random.randint(20000, 65535)
|
|
server_thread = threading.Thread(target=run_server, kwargs=dict(handler=handler, port=port))
|
|
server_thread.daemon = True
|
|
server_thread.start()
|
|
|
|
# Wait a moment for the server to start
|
|
time.sleep(0.5)
|
|
|
|
# Open the web browser
|
|
webbrowser.open(f"http://localhost:{port}")
|
|
|
|
# Wait for the server thread to finish
|
|
server_thread.join()
|
|
print("Server stopped")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except KeyboardInterrupt:
|
|
print("\nReceived keyboard interrupt, exiting...")
|
|
raise SystemExit(0)
|