目錄
Python實例題
題目
要求：
解題思路：
代碼實現：
Python實例題
題目
簡單的 Web 服務器
要求：
- 使用 Python 的?
socket?模塊實現一個簡單的 HTTP 服務器。 - 支持以下功能：
- 處理 GET 和 POST 請求
- 靜態文件服務（HTML、CSS、JavaScript、圖片等）
- 簡單的路由系統
- 錯誤處理（404、500 等）
- 添加多線程支持，以處理并發請求。
解題思路：
- 使用?
socket?建立網絡連接。 - 解析 HTTP 請求頭，處理不同類型的請求。
- 使用線程池處理并發連接。
代碼實現：
import socket
import threading
import os
import mimetypes
from urllib.parse import urlparse, parse_qsclass WebServer:def __init__(self, host='localhost', port=8080, document_root='www'):self.host = hostself.port = portself.document_root = document_rootself.routes = {'GET': {},'POST': {}}self.server_socket = Noneself.running = Falsedef route(self, method, path, handler):"""注冊路由處理函數"""self.routes[method][path] = handlerdef start(self):"""啟動服務器"""self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)self.server_socket.bind((self.host, self.port))self.server_socket.listen(5)self.running = Trueprint(f"服務器運行中: http://{self.host}:{self.port}")try:while self.running:client_socket, client_address = self.server_socket.accept()print(f"接受來自 {client_address} 的連接")# 創建線程處理請求client_handler = threading.Thread(target=self.handle_client,args=(client_socket,))client_handler.start()except KeyboardInterrupt:self.stop()def stop(self):"""停止服務器"""self.running = Falseif self.server_socket:self.server_socket.close()print("服務器已停止")def handle_client(self, client_socket):"""處理客戶端請求"""try:# 接收請求數據request_data = client_socket.recv(1024).decode('utf-8')if not request_data:return# 解析請求request_lines = request_data.splitlines()request_line = request_lines[0]method, path, _ = request_line.split(' ', 2)# 解析查詢參數parsed_url = urlparse(path)path = parsed_url.pathquery_params = parse_qs(parsed_url.query)# 處理 POST 請求體post_data = {}if method == 'POST':# 找到請求頭和請求體的分隔線body_index = request_data.find('\r\n\r\n')if body_index != -1:body = request_data[body_index + 4:]post_data = parse_qs(body)# 處理路由if path in self.routes[method]:response = self.routes[method][path](query_params, post_data)else:# 靜態文件服務response = self.serve_static_file(path)# 發送響應client_socket.sendall(response)except Exception as e:print(f"處理請求時出錯: {e}")error_response = self.generate_error_response(500)client_socket.sendall(error_response)finally:# 關閉連接client_socket.close()def serve_static_file(self, path):"""提供靜態文件服務"""# 處理根路徑if path == '/':path = '/index.html'# 構建文件路徑file_path = os.path.join(self.document_root, path[1:])# 檢查文件是否存在if not os.path.exists(file_path):return self.generate_error_response(404)# 檢查是否是目錄if os.path.isdir(file_path):return self.generate_error_response(403)try:# 讀取文件內容with open(file_path, 'rb') as file:content = file.read()# 確定 MIME 類型content_type, _ = mimetypes.guess_type(file_path)if not content_type:content_type = 'application/octet-stream'# 生成響應頭response_headers = (f'HTTP/1.1 200 OK\r\n'f'Content-Type: {content_type}\r\n'f'Content-Length: {len(content)}\r\n'f'\r\n').encode('utf-8')return response_headers + contentexcept Exception as e:print(f"讀取文件時出錯: {e}")return self.generate_error_response(500)def generate_error_response(self, status_code):"""生成錯誤響應"""status_messages = {404: 'Not Found',403: 'Forbidden',500: 'Internal Server Error'}status_message = status_messages.get(status_code, 'Error')# 生成錯誤頁面error_page = (f'<html><body><h1>{status_code} {status_message}</h1></body></html>').encode('utf-8')# 生成響應頭response_headers = (f'HTTP/1.1 {status_code} {status_message}\r\n'f'Content-Type: text/html\r\n'f'Content-Length: {len(error_page)}\r\n'f'\r\n').encode('utf-8')return response_headers + error_pagedef main():# 創建服務器實例server = WebServer(document_root='www')# 注冊動態路由def home_handler(query_params, post_data):response_body = '<html><body><h1>歡迎來到我的網站!</h1></body></html>'.encode('utf-8')response_headers = ('HTTP/1.1 200 OK\r\n''Content-Type: text/html\r\n'f'Content-Length: {len(response_body)}\r\n''\r\n').encode('utf-8')return response_headers + response_bodydef echo_handler(query_params, post_data):message = query_params.get('message', [''])[0]response_body = f'<html><body><p>你發送的消息是: {message}</p></body></html>'.encode('utf-8')response_headers = ('HTTP/1.1 200 OK\r\n''Content-Type: text/html\r\n'f'Content-Length: {len(response_body)}\r\n''\r\n').encode('utf-8')return response_headers + response_bodyserver.route('GET', '/', home_handler)server.route('GET', '/echo', echo_handler)# 啟動服務器try:server.start()except KeyboardInterrupt:passif __name__ == "__main__":main()
