#!/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 """
Python HTTP 服务端
🚀 Python HTTP 服务端
服务运行成功！以下是可用的API接口：
GET /api/status - 获取服务状态
GET /api/time - 获取当前时间
GET /api/user - 获取用户列表
GET /api/user?id=1 - 获取指定用户
POST /api/login - 用户登录 (username: admin, password: 123456)
POST /api/echo - 回显接收到的数据
测试示例：
# 获取状态
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
"""
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)