#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
@File : mockHTTPServer_Imporved_PT.py
@Create Time: 2025-08-14 9:22
@Description: Mock Server 服务 (Command Line Interface Version)
- 使用 logging 模块替换 print
- 接口返回值包含动态信息 (路径, 方法, 时间戳, UUID)
- 显式添加 Content-Length 响应头
- 在日志中打印接收到的 POST 数据内容
- 支持通过命令行参数或环境变量配置 Host 和 Port
- 默认监听所有接口 (0.0.0.0) 的 9012 端口
"""
import json
import logging
import argparse
import os
from http.server import HTTPServer, BaseHTTPRequestHandler
from socketserver import ThreadingMixIn
from datetime import datetime
import uuid
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class ThreadedHTTPServer(ThreadingMixIn, HTTPServer):
"""Handle requests in a separate thread."""
class RequestHandler(BaseHTTPRequestHandler):
"""处理 HTTP 请求"""
def _send_dynamic_response(self, method, path, received_data=None):
"""发送包含动态信息的 JSON 响应，ID使用UUID"""
# 生成唯一的请求ID
request_id = str(uuid.uuid4())
data = {
'status_code': 200,
'message': 'Request processed successfully',
'timestamp': datetime.now().isoformat(),
'id': request_id, # 使用动态生成的UUID
'request': {
'method': method,
'path': path
}
}
if received_data is not None:
# 将接收到的原始数据作为字符串存入响应
decoded_data = received_data.decode('utf-8') if isinstance(received_data, bytes) else received_data
data['request']['received_data'] = decoded_data
# 序列化并编码响应
response_bytes = json.dumps(data, ensure_ascii=False).encode('utf-8')
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.send_header('Content-Length', str(len(response_bytes)))
# 为了方便前端调用，可以添加 CORS 头 (根据评价建议)
# 注意：* 是不安全的，生产环境应指定具体域名
self.send_header('Access-Control-Allow-Origin', '*')
self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS')
self.send_header('Access-Control-Allow-Headers', 'Content-Type')
self.end_headers()
self.wfile.write(response_bytes)
# 记录响应发送
logger.info(f"Sent dynamic response for {method} {path} with ID: {request_id}")
def do_GET(self):
"""处理 GET 请求"""
logger.info(f"Received GET request for path: {self.path}")
try:
self._send_dynamic_response('GET', self.path)
except Exception as e:
logger.error(f"Error in GET request: {e}")
self.send_error(500, f"Internal Server Error: {e}")
def do_POST(self):
"""处理 POST 请求"""
logger.info(f"Received POST request for path: {self.path}")
try:
content_length = int(self.headers.get('Content-Length', 0))
post_data = self.rfile.read(content_length) if content_length > 0 else b''
# --- 打印 POST 数据内容 ---
if post_data:
try:
# 尝试将其作为 JSON 解析并美化打印
post_data_str = post_data.decode('utf-8')
post_data_json = json.loads(post_data_str)
logger.info(f"POST data received (JSON):\n{json.dumps(post_data_json, indent=2, ensure_ascii=False)}")
except json.JSONDecodeError:
# 如果不是 JSON，直接打印原始字符串
logger.info(f"POST data received (Raw): {post_data.decode('utf-8')}")
else:
logger.info("POST request received with no data.")
self._send_dynamic_response('POST', self.path, post_data)
except Exception as e:
logger.error(f"Error in POST request: {e}")
self.send_error(500, f"Internal Server Error: {e}")
def do_OPTIONS(self):
# Handle preflight CORS requests
self.send_response(200)
self.send_header('Access-Control-Allow-Origin', '*')
self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS')
self.send_header('Access-Control-Allow-Headers', 'Content-Type')
self.end_headers()
# 禁用 BaseHTTPRequestHandler 默认的日志记录格式
def log_message(self, format, *args):
pass # 使用我们自己的 logger
def parse_arguments():
"""解析命令行参数"""
parser = argparse.ArgumentParser(description="A simple mock HTTP server.")
parser.add_argument(
"--host", "-H",
type=str,
default=os.environ.get("MOCK_SERVER_HOST", "0.0.0.0"), # 默认从环境变量或 '0.0.0.0' 获取
help="Host to bind the server to (default: 0.0.0.0 or value of MOCK_SERVER_HOST env var)"
)
parser.add_argument(
"--port", "-p",
type=int,
default=int(os.environ.get("MOCK_SERVER_PORT", 9012)), # 默认从环境变量或 9012 获取
help="Port to bind the server to (default: 9012 or value of MOCK_SERVER_PORT env var)"
)
return parser.parse_args()
if __name__ == '__main__':
args = parse_arguments()
host = args.host
port = args.port
# 使用 ('', port) 也可以绑定到所有接口，但显式使用 '0.0.0.0' 更清晰
# 如果 host 是 '0.0.0.0' 或 ''，则监听所有接口
server_address = (host, port)
server = ThreadedHTTPServer(server_address, RequestHandler)
logger.info("====== Mock Server with CLI Main Begin ======")
logger.info("Starting server, listening at http://%s:%s", host if host else '0.0.0.0', port)
logger.info("Server will run persistently. Press Ctrl+C to stop.")
try:
server.serve_forever()
except KeyboardInterrupt:
logger.info("Received interrupt signal. Shutting down server gracefully...")
server.shutdown()
server.server_close()
logger.info("Server stopped.")
