這個需求快十年前弄過,那時是用 webpack proxy server,可以在習慣的桌機開發網頁,接著對於 CGI 查詢就透過 proxy 導向到 device CGI
然後,現在弄一個 Python ,方便後續在老 server 上運行,畢竟肥肥的程式碼搭配檔案系統大小寫問題,還是交給 server 檔案系統處理吧,如此,未來有部分 html/js/css code 要快速測試,就只需要到指定機器上,運行
$ python3 proxy-server.py --port 12345 --proxy-target 'http://192.168.123.234:80' --proxy-paths '/cgi-bin/' -d project/path/html/code/Serving files from directory: project/path/html/code/Serving HTTP on 0.0.0.0 port 12345 (multi-threaded) ...Proxying paths ['/cgi-bin/'] to http://192.168.123.234:80^CServer stopped.
如此對於 CGI request 就會自動導向到 'http://192.168.123.234:80'
程式碼純筆記,這都是 AI 產的:
#!/usr/bin/env python3import http.serverimport socketserverimport http.clientimport argparseimport osimport sysimport socketfrom urllib.parse import urlparse# 使用 ThreadingMixIn 來處理並發請求class ThreadedHTTPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):daemon_threads = Trueallow_reuse_address = Trueclass ProxyHTTPRequestHandler(http.server.SimpleHTTPRequestHandler):# 代理目標地址proxy_target = None# 需要代理的路徑前綴proxy_paths = []# 連接池(簡單實現)conn_pool = {}def __init__(self, *args, **kwargs):# 確保可以正確處理目錄參數super().__init__(*args, **kwargs)def do_GET(self):# 檢查是否需要代理這個請求for path_prefix in self.proxy_paths:if self.path.startswith(path_prefix):self.proxy_request()return# 如果不需要代理,則使用默認處理方式super().do_GET()def do_POST(self):# 為 POST 請求也提供代理功能for path_prefix in self.proxy_paths:if self.path.startswith(path_prefix):self.proxy_request()returnsuper().do_POST()def get_connection(self, host, is_ssl=False):"""從連接池獲取連接,或建立新連接"""key = (host, is_ssl)if key not in self.conn_pool:if is_ssl:self.conn_pool[key] = http.client.HTTPSConnection(host)else:self.conn_pool[key] = http.client.HTTPConnection(host)# 檢查連接是否仍然有效conn = self.conn_pool[key]try:# 嘗試使用一個非阻塞的方式檢查連接狀態old_timeout = conn.sock.gettimeout()conn.sock.settimeout(0.01)conn.sock.recv(1, socket.MSG_PEEK)conn.sock.settimeout(old_timeout)except (socket.error, AttributeError):# 連接已關閉或存在問題,創建新連接if is_ssl:self.conn_pool[key] = http.client.HTTPSConnection(host)else:self.conn_pool[key] = http.client.HTTPConnection(host)except Exception:# 其他類型的錯誤,可能連接仍然有效passreturn self.conn_pool[key]def proxy_request(self):"""使用 http.client 處理代理請求,這在 Python 3.2 中更高效"""target_url = urlparse(self.proxy_target + self.path)# 獲取請求頭部headers = {}for key, value in self.headers.items():# 排除一些特定的頭部if key.lower() not in ('host', 'content-length'):headers[key] = value# 設置正確的 Host 頭部headers['Host'] = target_url.netloc# 讀取請求體(如果有)content_length = int(self.headers.get('Content-Length', 0))body = self.rfile.read(content_length) if content_length > 0 else Nonetry:# 確定是否使用 HTTPSis_ssl = target_url.scheme == 'https'# 獲取或創建連接conn = self.get_connection(target_url.netloc, is_ssl)# 構建請求路徑request_path = target_url.pathif target_url.query:request_path += '?' + target_url.query# 發送請求conn.request(method=self.command,url=request_path,body=body,headers=headers)# 獲取響應response = conn.getresponse()# 設置響應狀態碼self.send_response(response.status)# 設置響應頭部for header in response.getheaders():key, value = headerif key.lower() != 'transfer-encoding': # 排除特定頭部self.send_header(key, value)self.end_headers()# 發送響應體self.wfile.write(response.read())# 不關閉連接,將其保留在連接池中except http.client.HTTPException as e:# 處理 HTTP 錯誤self.send_response(500)self.send_header('Content-Type', 'text/plain; charset=utf-8')self.end_headers()self.wfile.write("HTTP Error: {}".format(str(e)).encode('utf-8'))except Exception as e:# 處理其他錯誤self.send_response(500)self.send_header('Content-Type', 'text/plain; charset=utf-8')self.end_headers()self.wfile.write("Proxy error: {}".format(str(e)).encode('utf-8'))def run_server(port=8000, proxy_target="http://localhost:8080", proxy_paths=["/cgi-bin/"], directory=None):# 設置代理目標和路徑ProxyHTTPRequestHandler.proxy_target = proxy_targetProxyHTTPRequestHandler.proxy_paths = proxy_paths# 確保代理目標是有效的 URLif not ProxyHTTPRequestHandler.proxy_target.startswith(('http://', 'https://')):ProxyHTTPRequestHandler.proxy_target = "http://{}".format(ProxyHTTPRequestHandler.proxy_target)# 移除代理目標末尾的斜線(如果有)if ProxyHTTPRequestHandler.proxy_target.endswith('/'):ProxyHTTPRequestHandler.proxy_target = ProxyHTTPRequestHandler.proxy_target[:-1]# 設置工作目錄if directory:os.chdir(directory)print("Serving files from directory: {}".format(directory))else:print("Serving files from current directory: {}".format(os.getcwd()))# 提高 socket 超時時間以處理慢速連接socket.setdefaulttimeout(60)# 建立服務器(使用線程化服務器)server_address = ("", port)httpd = ThreadedHTTPServer(server_address, ProxyHTTPRequestHandler)print("Serving HTTP on 0.0.0.0 port {} (multi-threaded) ...".format(port))print("Proxying paths {} to {}".format(proxy_paths, proxy_target))try:httpd.serve_forever()except KeyboardInterrupt:print("\nServer stopped.")httpd.server_close()if __name__ == "__main__":# 解析命令行參數parser = argparse.ArgumentParser(description='HTTP Server with proxy capabilities (optimized for Python 3.2)')parser.add_argument('--port', type=int, default=8000, help='Port to listen on (default: 8000)')parser.add_argument('--proxy-target', type=str, default="http://localhost:8080",help='Target server to proxy to (default: http://localhost:8080)')parser.add_argument('--proxy-paths', type=str, default="/cgi-bin/",help='Comma-separated list of path prefixes to proxy (default: /cgi-bin/)')parser.add_argument('--directory', '-d', type=str, default=None,help='Specify directory to serve files from (default: current directory)')args = parser.parse_args()proxy_paths = [path.strip() for path in args.proxy_paths.split(',')]run_server(args.port, args.proxy_target, proxy_paths, args.directory)
沒有留言:
張貼留言