2025年4月22日 星期二

Python 開發筆記 - 使用 http.server 建置嵌入式產品網頁開發環境與 Proxy 機制

這個需求快十年前弄過,那時是用 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
^C
Server stopped.

如此對於 CGI request 就會自動導向到 'http://192.168.123.234:80' 

程式碼純筆記,這都是 AI 產的:

#!/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)

沒有留言:

張貼留言