這個需求快十年前弄過，那時是用 webpack proxy server，可以在習慣的桌機開發網頁，接著對於 CGI 查詢就透過 proxy 導向到 device CGI
然後，現在弄一個 Python ，方便後續在老 server 上運行，畢竟肥肥的程式碼搭配檔案系統大小寫問題，還是交給 server 檔案系統處理吧，如此，未來有部分 html/js/css code 要快速測試，就只需要到指定機器上，運行
#!/usr/bin/env python3
import http.server
import socketserver
import http.client
import argparse
import os
import sys
import socket
from urllib.parse import urlparse
# 使用 ThreadingMixIn 來處理並發請求
class ThreadedHTTPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
daemon_threads = True
allow_reuse_address = True
class 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()
return
super().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:
# 其他類型的錯誤，可能連接仍然有效
pass
return 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 None
try:
# 確定是否使用 HTTPS
is_ssl = target_url.scheme == 'https'
# 獲取或創建連接
conn = self.get_connection(target_url.netloc, is_ssl)
# 構建請求路徑
request_path = target_url.path
if 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 = header
if 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_target
ProxyHTTPRequestHandler.proxy_paths = proxy_paths
# 確保代理目標是有效的 URL
if 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)
