1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
|
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
from pathlib import Path
from datetime import datetime
class RequestHandler(BaseHTTPRequestHandler):
"""自定义请求处理器"""
requests_log = []
def log_request_info(self):
"""记录请求信息"""
content_length = int(self.headers.get('Content-Length', 0))
body = self.rfile.read(content_length).decode('utf-8') if content_length > 0 else None
request_info = {
'timestamp': datetime.now().isoformat(),
'method': self.command,
'path': self.path,
'headers': dict(self.headers),
'body': body
}
RequestHandler.requests_log.append(request_info)
print(f"\n[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 收到请求:")
print(f" 方法: {self.command}")
print(f" 路径: {self.path}")
if body:
print(f" 数据: {body}")
return body
def send_response_json(self, data, status=200):
"""发送JSON响应"""
self.send_response(status)
self.send_header('Content-Type', 'application/json')
self.end_headers()
self.wfile.write(json.dumps(data, ensure_ascii=False).encode('utf-8'))
def do_POST(self):
"""处理POST请求"""
if self.path == '/push_parameters':
body = self.log_request_info()
if not body or 'parameters' not in body:
self.send_response_json({
'code': 400,
'message': 'Missing required parameter: parameters',
'status': 'error'
}, 400)
return
try:
data = json.loads(body)
parameters = data['parameters']
id = data['id']
token = self.get_token()
path = Path(r"C:\Users\Administrator\Documents")
target_file = f"{id}.csv"
file_path = path / target_file
self.upload_file(token, file_path)
self.send_response_json({
'code': 200,
'message': 'Parameters pushed successfully',
'status': 'success',
'data': {
'parameters': parameters,
'id': id
}
})
except ValueError as e:
self.send_response_json({
'code': 400,
'message': f'Invalid parameter format: {str(e)}',
'status': 'error'
}, 400)
except Exception as e:
self.send_response_json({
'code': 500,
'message': f'Internal server error: {str(e)}',
'status': 'error'
}, 500)
else:
self.send_response_json({
'code': 404,
'message': 'Not found',
'status': 'error'
}, 404)
@staticmethod
def get_token():
pass
@staticmethod
def upload_file(token, file_path):
pass
def run_server(port=8080):
"""启动服务器"""
server_address = ('', port)
httpd = HTTPServer(server_address, RequestHandler)
print(f"服务器已启动，监听端口: {port}")
print(f"访问地址: http://localhost:{port}")
print("\n可用端点:")
print(" POST /push_parameters- 接收参数")
print("\n按 Ctrl+C 停止服务\n")
try:
httpd.serve_forever()
except KeyboardInterrupt:
print("\n服务已停止")
httpd.server_close()
if __name__ == '__main__':
run_server(8777)
|
