Spaces:
Runtime error
Runtime error
| import http.server | |
| import socketserver | |
| import urllib.request | |
| import urllib.parse | |
| import re | |
| import gzip | |
| import time | |
| import json | |
| import threading | |
| PORT = 5173 | |
| CACHE = {} | |
| OPENER = urllib.request.build_opener() | |
| OPENER.addheaders = [('User-Agent', 'Mozilla/5.0'), ('Accept-Encoding', 'gzip')] | |
| class ProxyHandler(http.server.BaseHTTPRequestHandler): | |
| protocol_version = 'HTTP/1.1' | |
| def do_GET(self): | |
| parsed = urllib.parse.urlparse(self.path) | |
| if parsed.path == '/api/internet_speed': | |
| try: | |
| test_url = "https://google.com" | |
| start = time.perf_counter() | |
| with OPENER.open(test_url, timeout=5) as res: | |
| data = res.read() | |
| duration = time.perf_counter() - start | |
| mbps = ((len(data) * 8) / duration) / 1_000_000 | |
| self._send_json({"mbps": round(mbps, 2)}) | |
| except Exception as e: | |
| self._send_json({"error": str(e)}, 500) | |
| return | |
| query = urllib.parse.parse_qs(parsed.query) | |
| target = query.get('proxy', [None])[0] | |
| if not target: | |
| self._send_resp(b"Usage: /site?proxy=URL or /api/internet_speed", "text/plain") | |
| return | |
| if target in CACHE: | |
| self._send_resp(CACHE[target][0], CACHE[target][1]) | |
| return | |
| try: | |
| with OPENER.open(target, timeout=7) as res: | |
| ctype = res.headers.get('Content-Type', '') | |
| content = res.read() | |
| if res.headers.get('Content-Encoding') == 'gzip': | |
| content = gzip.decompress(content) | |
| if any(t in ctype for t in ['html', 'css', 'javascript']): | |
| try: | |
| text = content.decode('utf-8', errors='ignore') | |
| pattern = r'(href|src|action)=["\'](.*?)["\']|url\(["\']?(.*?)["\']?\)' | |
| def rewrite(m): | |
| is_url_func = m.group(0).startswith('url') | |
| attr_val = m.group(3) if is_url_func else m.group(2) | |
| prefix = m.group(1) if not is_url_func else None | |
| if not attr_val or attr_val.startswith(('data:', 'javascript:', '#')): | |
| return m.group(0) | |
| full_url = urllib.parse.urljoin(target, attr_val) | |
| proxy_url = f'/site?proxy={urllib.parse.quote(full_url)}' | |
| return f'url("{proxy_url}")' if is_url_func else f'{prefix}="{proxy_url}"' | |
| text = re.sub(pattern, rewrite, text) | |
| if 'html' in ctype: | |
| text = re.sub(r'(<head.*?>)', r'\1<base href="'+target+'">', text, count=1, flags=re.I) | |
| content = text.encode('utf-8') | |
| except: | |
| pass | |
| if len(CACHE) < 200: | |
| CACHE[target] = (content, ctype) | |
| self._send_resp(content, ctype) | |
| except Exception as e: | |
| self._send_resp(str(e).encode(), "text/plain", 500) | |
| def _send_resp(self, body, ctype, code=200): | |
| self.send_response(code) | |
| self.send_header('Content-Type', ctype) | |
| self.send_header('Content-Length', len(body)) | |
| self.send_header('Access-Control-Allow-Origin', '*') | |
| self.end_headers() | |
| self.wfile.write(body) | |
| def _send_json(self, data, code=200): | |
| self._send_resp(json.dumps(data).encode(), "application/json", code) | |
| class ThreadedServer(socketserver.ThreadingMixIn, socketserver.TCPServer): | |
| daemon_threads = True | |
| allow_reuse_address = True | |
| if __name__ == "__main__": | |
| with ThreadedServer(("0.0.0.0", PORT), ProxyHandler) as server: | |
| print(f"Server online at http://0.0.0.0:{PORT}") | |
| server.serve_forever() |