537 lines
23 KiB
Python
537 lines
23 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
|
||
from http.server import HTTPServer, BaseHTTPRequestHandler
|
||
import json
|
||
import urllib.parse
|
||
import logging
|
||
|
||
# 配置日志
|
||
logging.basicConfig(level=logging.INFO)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class CustomHTTPRequestHandler(BaseHTTPRequestHandler):
|
||
"""自定义HTTP请求处理器"""
|
||
|
||
def do_GET(self):
|
||
"""处理GET请求"""
|
||
# 解析URL和查询参数
|
||
parsed_path = urllib.parse.urlparse(self.path)
|
||
query_params = urllib.parse.parse_qs(parsed_path.query)
|
||
|
||
logger.info(f"GET 请求: {self.path}")
|
||
|
||
# 根据路径返回不同的响应
|
||
if parsed_path.path == "/":
|
||
self._send_response(200, "text/html", self._get_home_page())
|
||
elif parsed_path.path == "/api/user":
|
||
self._handle_user_api(query_params)
|
||
elif parsed_path.path == "/api/status":
|
||
self._send_json_response(200, {"status": "ok", "message": "服务运行正常"})
|
||
elif parsed_path.path == "/api/time":
|
||
import datetime
|
||
|
||
current_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
self._send_json_response(200, {"time": current_time})
|
||
else:
|
||
self._send_json_response(404, {"error": "页面未找到"})
|
||
|
||
def do_POST(self):
|
||
"""处理POST请求"""
|
||
content_length = int(self.headers.get("Content-Length", 0))
|
||
post_data = self.rfile.read(content_length)
|
||
|
||
logger.info(f"POST 请求: {self.path}")
|
||
|
||
try:
|
||
# 尝试解析JSON数据
|
||
if content_length == 0:
|
||
logger.info("POST 请求数据为空")
|
||
elif self.headers.get("Content-Type") == "application/json":
|
||
data = json.loads(post_data.decode("utf-8"))
|
||
logger.info(f"接收到JSON数据: {data}")
|
||
else:
|
||
# 解析表单数据
|
||
data = urllib.parse.parse_qs(post_data.decode("utf-8"))
|
||
logger.info(f"接收到表单数据: {data}")
|
||
|
||
# print("x"*100)
|
||
# 根据路径处理不同的POST请求
|
||
# if self.path == '/api/login':
|
||
# self._handle_login(data)
|
||
# elif self.path == '/api/echo':
|
||
# self._send_json_response(200, {'echo': data, 'message': '数据已接收'})
|
||
# else:
|
||
# self._send_json_response(404, {'error': '接口未找到'})
|
||
if self.path == "/api/resource/v2/door/search":
|
||
self._send_json_response(
|
||
200,
|
||
{
|
||
"code": "0",
|
||
"msg": "SUCCESS",
|
||
"data": {
|
||
"total": 3,
|
||
"pageNo": 1,
|
||
"pageSize": 1,
|
||
"list": [
|
||
{
|
||
"indexCode": "df8w8cr800283c24c",
|
||
"resourceType": "door",
|
||
"name": "资源 1",
|
||
"doorNo": "123",
|
||
"channelNo": "1",
|
||
"parentIndexCode": "80d9099q9e991231",
|
||
"controlOneId": "11111111",
|
||
"controlTwoId": "2222222222",
|
||
"readerInId": "ac789y2c0019c",
|
||
"readerOutId": "arcew78c710",
|
||
"doorSerial": 1,
|
||
"treatyType": "hiksdk_net",
|
||
"regionIndexCode": "d8a5476e-25c0-4aa2-b7e3-db3788ba1f77",
|
||
"regionPath": "@root000000@",
|
||
"createTime": "2018-11-28T16:47:27:358+08:00",
|
||
"updateTime": "2018-11-28T16:48:34:011+08:00",
|
||
"description": "Test",
|
||
"channelType": "door",
|
||
"regionName": "acs_setUp_42054",
|
||
"regionPathName": "@root000000@9ca1eef0-4579-4e7e-a601-caf486442d54@",
|
||
"installLocation": "位置 1",
|
||
}
|
||
],
|
||
},
|
||
},
|
||
)
|
||
elif self.path == "/api/v1/door/states":
|
||
self._send_json_response(
|
||
200,
|
||
{
|
||
"code": "0",
|
||
"msg": "success",
|
||
"data": {
|
||
"authDoorList": [
|
||
{
|
||
"doorIndexCode": "e8e3ef5c149243abb4341124ab38fcfc",
|
||
"doorState": 0,
|
||
}
|
||
],
|
||
"noAuthDoorIndexCodeList": [
|
||
"e8e3ef5c149243abb4341124ab38fcfc"
|
||
],
|
||
},
|
||
},
|
||
)
|
||
elif self.path == "/api/acs/v1/door/doControl":
|
||
self._send_json_response(
|
||
200,
|
||
{
|
||
"code": "0",
|
||
"msg": "success",
|
||
"data": [
|
||
{
|
||
"doorIndexCode": "2c95c028a809448f962a969e3ab34f",
|
||
"controlResultCode": 0, # 0表示反控成功, 其他表示失败
|
||
"controlResultDesc": "success",
|
||
}
|
||
],
|
||
},
|
||
)
|
||
elif self.path == "/api/acs/v2/door/events":
|
||
self._send_json_response(
|
||
200,
|
||
{
|
||
"code": "0",
|
||
"msg": "success",
|
||
"data": {
|
||
"total": 1,
|
||
"totalPage": 1,
|
||
"pageNo": 1,
|
||
"pageSize": 100,
|
||
"list": [
|
||
{
|
||
"eventId": "207dd3b1-37a7-4d6c-8e4d-c8bfd343051b",
|
||
"eventName": "acs.acs.eventType.successCard",
|
||
"eventTime": "2019-11-16T15:44:33+08:00",
|
||
"personId": "216e2ba145824269a1cbb423cdc85cb1",
|
||
"cardNo": "3891192334",
|
||
"personName": "sdk 人员 1zzzcb",
|
||
"orgIndexCode": "root000000",
|
||
"orgName": "默认组织",
|
||
"doorName": "10.40.239.69new_test2_门_1",
|
||
"doorIndexCode": "f0b50050d3434f15b4e34f885d5dacfe",
|
||
"doorRegionIndexCode": "fd2df06b-1afb-4c9b-b058-5740c2c00076",
|
||
"picUri": "no-pcnvr",
|
||
"svrIndexCode": "/pic?=d62i7f6e*6a7i125-c838b9--a8c67dea96e65icb1*=sd*=5dpi*=1dpi*m2i1t=4ed35444bb4s=-39",
|
||
"eventType": 198914,
|
||
"inAndOutType": 1,
|
||
"readerDevIndexCode": "378e563bf3e84d5ba6ef5742bbaa8933",
|
||
"readerDevName": "读卡器_1",
|
||
"devIndexCode": "dcff422aad9c4d60a47b8b2fe2757b71",
|
||
"devName": "10.40.239.69new_test2",
|
||
"identityCardUri": "/pic?=d62i7f6e*6a7i125-c838b9--a8c67dea96e65icb1*=sd*=5dpi*=1dpi*m2i1t=4ed35444bb4s=-39z422d3",
|
||
"receiveTime": "2019-11-16T15:45:13.525+08:00",
|
||
"jobNo": "23333",
|
||
"studentId": "201900001",
|
||
"certNo": "320826199012110005",
|
||
}
|
||
],
|
||
},
|
||
},
|
||
)
|
||
elif self.path == "/api/frs/v1/face/group":
|
||
self._send_json_response(
|
||
200,
|
||
{
|
||
"code": "0",
|
||
"msg": "Success",
|
||
"data": [
|
||
{
|
||
"indexCode": "5dc82633-a4cb-4107-b55e-f21bf952f9",
|
||
"name": "仓库值守人员",
|
||
"description": "仓库值守人员是指守着仓库的人",
|
||
}
|
||
],
|
||
},
|
||
)
|
||
elif self.path == "/api/frs/v1/application/oneToMany":
|
||
self._send_json_response(
|
||
200,
|
||
{
|
||
"code": "0",
|
||
"msg": "Success.",
|
||
"data": {
|
||
"total": 500,
|
||
"pageNo": 1,
|
||
"pageSize": 10,
|
||
"list": [
|
||
{
|
||
"similarity": 80,
|
||
"indexCode": "7cc0adb2-a3c3-48fd-b432-718103e85c28",
|
||
"faceInfo": {
|
||
"name": "张三",
|
||
"sex": "1",
|
||
"certificateType": "111",
|
||
"certificateNum": "420204199605121656",
|
||
},
|
||
"facePic": {
|
||
"faceUrl": "http://10.166.165.121:8080/frs/facepicturetemp/test.jpg"
|
||
},
|
||
}
|
||
],
|
||
},
|
||
},
|
||
)
|
||
elif self.path == "/api/frs/v1/face/picture/check":
|
||
self._send_json_response(
|
||
200,
|
||
{
|
||
"code": "0",
|
||
"msg": "Success",
|
||
"data": {
|
||
"checkResult": True,
|
||
"faceScore": 90,
|
||
"facePicAnalysisResult": {
|
||
"id": 5566,
|
||
"age": 16,
|
||
"ageRange": 1,
|
||
"ageGroup": "TEENAGER",
|
||
"": "male",
|
||
"glasses": "NO",
|
||
"smile": "NO",
|
||
"facePose": {
|
||
"pitch": 45,
|
||
"yaw": 25,
|
||
"roll": 10,
|
||
"clearityScore": 0.5,
|
||
"colorConfidence": 0.5,
|
||
"eyeDistance": 300,
|
||
"grayMean": 120,
|
||
"visibleScore": 0.5,
|
||
},
|
||
"targetModelData": "DD",
|
||
"faceRect": {
|
||
"height": 12.1,
|
||
"width": 16,
|
||
"x": 15,
|
||
"y": 3,
|
||
},
|
||
"recommendFaceRect": {
|
||
"height": 4,
|
||
"width": 6,
|
||
"x": 2,
|
||
"y": 1,
|
||
},
|
||
"faceMark": {
|
||
"leftEye": {"x": 33, "y": 22},
|
||
"rightEye": {"x": 44, "y": 33},
|
||
"noseTip": {"x": 43, "y": 12},
|
||
"leftMouth": {"x": 32, "y": 54},
|
||
"rightMouth": {"x": 67, "y": 12},
|
||
},
|
||
"mask": "NO",
|
||
"faceScore": 90,
|
||
},
|
||
},
|
||
},
|
||
)
|
||
elif self.path == "/api/visitor/v2/appointment/records":
|
||
self._send_json_response(
|
||
200,
|
||
{
|
||
"code": "0",
|
||
"msg": "success",
|
||
"data": {
|
||
"total": 1,
|
||
"pageNo": 1,
|
||
"pageSize": 20,
|
||
"list": [
|
||
{
|
||
"appointRecordId": "321654987",
|
||
"receptionistId": "3124126241412",
|
||
"receptionistName": "王五",
|
||
"receptionistCode": "323JH234KJH23",
|
||
"visitStartTime": "2018-07-26T15:00:00 + 08:00",
|
||
"visitEndTime": "2018-07-26T19:00:00 + 08:00",
|
||
"visitPurpose": "参考",
|
||
"visitorName": "张三",
|
||
"visitorId": "ASDF454SDAF565613JHU7712332",
|
||
"verificationCode": "1234",
|
||
"QRCode": "2015468421",
|
||
"": 1,
|
||
"phoneNo": "13576361254",
|
||
"plateNo": "浙 A12345",
|
||
"certificateType": 111,
|
||
"certificateNo": "311256196602145692",
|
||
"picUri": "/pic?adsdqwe21-asafdd-12sfsdfsdf",
|
||
"svrIndexCode": "sadsa123-asd21edsfhgsd-23rfdvsr",
|
||
"visitorStatus": 1,
|
||
"certAddr": "杭州滨江",
|
||
"certIssuer": "滨江分局",
|
||
"nation": 1,
|
||
"birthplace": "杭州",
|
||
"visitorWorkUnit": "中国工商银行",
|
||
"visitorAddress": "杭州滨江",
|
||
"orderId": "d089ady8a0dud87018d0y90ay9d901",
|
||
"designatedResources": {
|
||
"paramKey": "1",
|
||
"paramValues": ["52v72v35762587n75b26"],
|
||
},
|
||
"privilegeGroupNames": ["one"],
|
||
"identityUri": "/pic?123-scccdf334-3216516516516",
|
||
"identitySvrCode": "12ddf53ggg56sss6554",
|
||
}
|
||
],
|
||
},
|
||
},
|
||
)
|
||
elif self.path == "/api/nms/v1/online/acs_device/get":
|
||
self._send_json_response(
|
||
200,
|
||
{
|
||
"code": "0",
|
||
"msg": "success",
|
||
"data": {
|
||
"pageNo": 1,
|
||
"pageSize": 10,
|
||
"totalPage": 0,
|
||
"total": 1,
|
||
"list": [
|
||
{
|
||
"deviceType": "HIK%2FDS-9116HW-ST%2F-AF-DVR",
|
||
"deviceIndexCode": "null",
|
||
"regionIndexCode": "ce91c758-5af4-4539-845a",
|
||
"collectTime": "2018-12-28T10:21:40.000+08:00",
|
||
"regionName": "NMS 自动化",
|
||
"indexCode": "82896441ced946d5a51c6d6ca8e65851",
|
||
"cn": "Onvif-IPC(10.67.172.13 )",
|
||
"treatyType": "onvif_net",
|
||
"manufacturer": "hikvision",
|
||
"ip": "10.67.172.13",
|
||
"port": 80,
|
||
"online": 1,
|
||
}
|
||
],
|
||
},
|
||
},
|
||
)
|
||
except json.JSONDecodeError:
|
||
self._send_json_response(400, {"error": "无效的JSON数据"})
|
||
except Exception as e:
|
||
self._send_json_response(500, {"code": 0, "error": f"服务器错误: {str(e)}"})
|
||
|
||
def do_PUT(self):
|
||
"""处理PUT请求"""
|
||
content_length = int(self.headers.get("Content-Length", 0))
|
||
put_data = self.rfile.read(content_length)
|
||
|
||
logger.info(f"PUT 请求: {self.path}")
|
||
|
||
try:
|
||
data = json.loads(put_data.decode("utf-8"))
|
||
self._send_json_response(200, {"message": "数据已更新", "data": data})
|
||
except json.JSONDecodeError:
|
||
self._send_json_response(400, {"error": "无效的JSON数据"})
|
||
|
||
def do_DELETE(self):
|
||
"""处理DELETE请求"""
|
||
logger.info(f"DELETE 请求: {self.path}")
|
||
self._send_json_response(200, {"message": "删除成功", "path": self.path})
|
||
|
||
def _handle_user_api(self, query_params):
|
||
"""处理用户API请求"""
|
||
user_id = query_params.get("id", [""])[0]
|
||
if user_id:
|
||
user_data = {
|
||
"id": user_id,
|
||
"name": f"用户{user_id}",
|
||
"email": f"user{user_id}@example.com",
|
||
"status": "active",
|
||
}
|
||
self._send_json_response(200, user_data)
|
||
else:
|
||
# 返回用户列表
|
||
users = [
|
||
{"id": "1", "name": "张三", "email": "zhangsan@example.com"},
|
||
{"id": "2", "name": "李四", "email": "lisi@example.com"},
|
||
{"id": "3", "name": "王五", "email": "wangwu@example.com"},
|
||
]
|
||
self._send_json_response(200, {"users": users})
|
||
|
||
def _handle_login(self, data):
|
||
"""处理登录请求"""
|
||
username = (
|
||
data.get("username", [""])[0]
|
||
if isinstance(data, dict)
|
||
else data.get("username", "")
|
||
)
|
||
password = (
|
||
data.get("password", [""])[0]
|
||
if isinstance(data, dict)
|
||
else data.get("password", "")
|
||
)
|
||
|
||
# 简单的用户验证（仅做演示）
|
||
if username == "admin" and password == "123456":
|
||
response_data = {
|
||
"success": True,
|
||
"message": "登录成功",
|
||
"token": "fake_jwt_token_here",
|
||
"user": {"username": username, "role": "admin"},
|
||
}
|
||
self._send_json_response(200, response_data)
|
||
else:
|
||
self._send_json_response(
|
||
401, {"success": False, "message": "用户名或密码错误"}
|
||
)
|
||
|
||
def _send_response(self, status_code, content_type, content):
|
||
"""发送HTTP响应"""
|
||
self.send_response(status_code)
|
||
self.send_header("Content-Type", f"{content_type}; charset=utf-8")
|
||
self.send_header("Access-Control-Allow-Origin", "*") # 允许跨域
|
||
self.send_header(
|
||
"Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS"
|
||
)
|
||
self.send_header("Access-Control-Allow-Headers", "Content-Type")
|
||
self.end_headers()
|
||
self.wfile.write(content.encode("utf-8"))
|
||
|
||
def _send_json_response(self, status_code, data):
|
||
"""发送JSON响应"""
|
||
json_data = json.dumps(data, ensure_ascii=False, indent=2)
|
||
self._send_response(status_code, "application/json", json_data)
|
||
|
||
def _get_home_page(self):
|
||
"""获取首页HTML"""
|
||
return """
|
||
<!DOCTYPE html>
|
||
<html lang="zh-CN">
|
||
<head>
|
||
<meta charset="UTF-8">
|
||
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
||
<title>Python HTTP 服务端</title>
|
||
<style>
|
||
body { font-family: Arial, sans-serif; margin: 40px; }
|
||
.api-list { background: #f5f5f5; padding: 20px; border-radius: 5px; }
|
||
.api-item { margin: 10px 0; padding: 10px; background: white; border-radius: 3px; }
|
||
.method { font-weight: bold; color: #007cba; }
|
||
</style>
|
||
</head>
|
||
<body>
|
||
<h1>🚀 Python HTTP 服务端</h1>
|
||
<p>服务运行成功！以下是可用的API接口：</p>
|
||
|
||
<div class="api-list">
|
||
<div class="api-item">
|
||
<span class="method">GET</span> /api/status - 获取服务状态
|
||
</div>
|
||
<div class="api-item">
|
||
<span class="method">GET</span> /api/time - 获取当前时间
|
||
</div>
|
||
<div class="api-item">
|
||
<span class="method">GET</span> /api/user - 获取用户列表
|
||
</div>
|
||
<div class="api-item">
|
||
<span class="method">GET</span> /api/user?id=1 - 获取指定用户
|
||
</div>
|
||
<div class="api-item">
|
||
<span class="method">POST</span> /api/login - 用户登录 (username: admin, password: 123456)
|
||
</div>
|
||
<div class="api-item">
|
||
<span class="method">POST</span> /api/echo - 回显接收到的数据
|
||
</div>
|
||
</div>
|
||
|
||
<h3>测试示例：</h3>
|
||
<pre>
|
||
# 获取状态
|
||
curl http://localhost:8080/api/status
|
||
|
||
# 用户登录
|
||
curl -X POST -H "Content-Type: application/json" \\
|
||
-d '{"username":"admin","password":"123456"}' \\
|
||
http://localhost:8080/api/login
|
||
|
||
# 获取用户信息
|
||
curl http://localhost:8080/api/user?id=1
|
||
</pre>
|
||
</body>
|
||
</html>
|
||
"""
|
||
|
||
def do_OPTIONS(self):
|
||
"""处理预检请求（CORS）"""
|
||
self.send_response(200)
|
||
self.send_header("Access-Control-Allow-Origin", "*")
|
||
self.send_header(
|
||
"Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS"
|
||
)
|
||
self.send_header("Access-Control-Allow-Headers", "Content-Type")
|
||
self.end_headers()
|
||
|
||
def log_message(self, format, *args):
|
||
"""自定义日志格式"""
|
||
logger.info(f"{self.address_string()} - {format % args}")
|
||
|
||
|
||
def run_server(host="localhost", port=8080):
|
||
"""启动HTTP服务器"""
|
||
server_address = (host, port)
|
||
httpd = HTTPServer(server_address, CustomHTTPRequestHandler)
|
||
|
||
print(f"🌟 HTTP服务器启动成功!")
|
||
print(f"📍 地址: http://{host}:{port}")
|
||
print(f"🔗 在浏览器中访问: http://{host}:{port}")
|
||
print(f"⏹️ 按 Ctrl+C 停止服务器\n")
|
||
|
||
try:
|
||
httpd.serve_forever()
|
||
except KeyboardInterrupt:
|
||
print("\n🛑 服务器已停止")
|
||
httpd.server_close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 启动服务器
|
||
run_server(host="localhost", port=9919)
|