初始化海康api测试库
This commit is contained in:
commit
ff4ccc8fa7
96
001测试海康api.py
Normal file
96
001测试海康api.py
Normal file
@ -0,0 +1,96 @@
|
||||
from util.haikang_util import HaikangUtil
|
||||
import asyncio
|
||||
import base64
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
# 查询门禁点列表
|
||||
async def get_door_list_service(pageNo: int = 1, pageSize: int = 10):
|
||||
result = await HaikangUtil.get_door_list_v2(pageNo, pageSize)
|
||||
print(result)
|
||||
|
||||
# 查询门禁状态
|
||||
async def get_door_status_service(door_index_codes):
|
||||
result = await HaikangUtil.get_door_status(door_index_codes)
|
||||
print(result)
|
||||
|
||||
# 门禁控制
|
||||
async def door_do_control_service(door_index_codes, control_type):
|
||||
result = await HaikangUtil.door_do_control(door_index_codes, control_type)
|
||||
print(result)
|
||||
|
||||
# 查询门禁点事件
|
||||
async def query_door_events_service(door_index_code,pageNo, pageSize, startTime, endTime):
|
||||
result = await HaikangUtil.query_door_events_v2(door_index_code, pageNo=pageNo, pageSize=pageSize ,startTime=startTime, endTime=endTime)
|
||||
print(result)
|
||||
|
||||
# 查看门禁点在线状态
|
||||
async def door_online_status_service(door_index_codes):
|
||||
result = await HaikangUtil.door_online_status(door_index_codes)
|
||||
print(result)
|
||||
|
||||
# 按条件查询人脸分组, 很重要
|
||||
async def get_face_group_service():
|
||||
result = await HaikangUtil.get_face_group()
|
||||
print(result)
|
||||
|
||||
# 人脸分组1vN搜索
|
||||
async def face_group_1vN_search_service(image_path):
|
||||
|
||||
with open(image_path, 'rb') as f:
|
||||
image_data = f.read()
|
||||
|
||||
encoded_image = base64.b64encode(image_data).decode('utf-8')
|
||||
|
||||
result = await HaikangUtil.face_group_1vN_search(
|
||||
facePicBinaryData=encoded_image,
|
||||
pageNo=1,
|
||||
pageSize=10,
|
||||
searchNum=99,
|
||||
minSimilarity=50,
|
||||
faceGroupIndexCodes=['5dc82633-a4cb-4107-b55e-f21bf952f9']
|
||||
)
|
||||
print(result)
|
||||
|
||||
# 人脸评分
|
||||
async def face_picture_check(image_path):
|
||||
|
||||
with open(image_path, 'rb') as f:
|
||||
image_data = f.read()
|
||||
|
||||
encoded_image = base64.b64encode(image_data).decode('utf-8')
|
||||
|
||||
|
||||
result = await HaikangUtil.face_picture_check(
|
||||
facePicBinaryData=encoded_image
|
||||
)
|
||||
print(result)
|
||||
|
||||
# 查询访客预约记录
|
||||
async def query_visitor_record():
|
||||
result = await HaikangUtil.query_visitor_record()
|
||||
print(result)
|
||||
|
||||
if __name__ == '__main__':
|
||||
# asyncio.run(get_door_list_service())
|
||||
# print("*"*100)
|
||||
# asyncio.run(get_door_status_service(['D01']))
|
||||
# print("*"*100)
|
||||
|
||||
# asyncio.run(door_do_control_service(['D01'], 1))
|
||||
|
||||
# asyncio.run(query_door_events_service('D01',1,10,1640995200,1640995200))
|
||||
|
||||
# asyncio.run(get_face_group_service())
|
||||
|
||||
# image_path = "75c03e462769c81b6a8513d90ff2a27d.jpg"
|
||||
# asyncio.run(face_group_1vN_search_service(image_path))
|
||||
# asyncio.run(face_picture_check(image_path))
|
||||
|
||||
# asyncio.run(query_visitor_record())
|
||||
|
||||
asyncio.run(door_online_status_service(["xxxxxxxx"]))
|
||||
|
||||
536
002简单http服务.py
Normal file
536
002简单http服务.py
Normal file
@ -0,0 +1,536 @@
|
||||
#!/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)
|
||||
0
config/__init__.py
Normal file
0
config/__init__.py
Normal file
BIN
config/__pycache__/__init__.cpython-310.pyc
Normal file
BIN
config/__pycache__/__init__.cpython-310.pyc
Normal file
Binary file not shown.
BIN
config/__pycache__/env.cpython-310.pyc
Normal file
BIN
config/__pycache__/env.cpython-310.pyc
Normal file
Binary file not shown.
86
config/env.py
Normal file
86
config/env.py
Normal file
@ -0,0 +1,86 @@
|
||||
import argparse
|
||||
import os
|
||||
import sys
|
||||
from dotenv import load_dotenv
|
||||
from functools import lru_cache
|
||||
from pydantic import computed_field
|
||||
from pydantic_settings import BaseSettings
|
||||
from typing import Literal
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
class HaiKangSettings:
|
||||
"""
|
||||
海康平台配置
|
||||
"""
|
||||
HAIKANG_URL = 'http://localhost'
|
||||
# HAIKANG_PORT = 443
|
||||
HAIKANG_PORT = 9919
|
||||
HAIKANG_AK = ''
|
||||
HAIKANG_SK = ''
|
||||
HAIKANG_ACCESS_TOKEN_URL = '/api/v1/oauth/token'
|
||||
HAIKANG_DOOR_STATES_URL = '/api/v1/door/states'
|
||||
HAIKANG_DOOR_DOCONTROL_URL = '/api/acs/v1/door/doControl'
|
||||
HAIKANG_DOOR_ENVENTS_URL = '/api/acs/v2/door/events'
|
||||
HAIKANG_DOOR_SEARCH = '/api/resource/v2/door/search'
|
||||
HAIKANG_DOOR_ONLINE_STATUS = '/api/nms/v1/online/acs_device/get'
|
||||
HAIKANG_APPLICATION_ONETOMANY_URL = '/api/frs/v1/application/oneToMany'
|
||||
HAIKANG_PICTURE_CHECK_URL = '/api/frs/v1/face/picture/check'
|
||||
HAIKANG_FACECAPATURE_SEARCH = '/api/frs/v1/event/face_capture/search'
|
||||
HAIKANG_FACE_GROUP_URL = '/api/frs/v1/face/group'
|
||||
HAIKANG_VISITOR_RECORD_SEARCH = '/api/visitor/v2/appointment/records'
|
||||
|
||||
|
||||
|
||||
|
||||
class GetConfig:
|
||||
"""
|
||||
获取配置
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.parse_cli_args()
|
||||
|
||||
#
|
||||
@lru_cache()
|
||||
def get_haikang_config(self):
|
||||
"""
|
||||
获取海康平台配置
|
||||
"""
|
||||
return HaiKangSettings()
|
||||
|
||||
|
||||
@staticmethod
|
||||
def parse_cli_args():
|
||||
"""
|
||||
解析命令行参数
|
||||
"""
|
||||
if 'uvicorn' in sys.argv[0]:
|
||||
# 使用uvicorn启动时，命令行参数需要按照uvicorn的文档进行配置，无法自定义参数
|
||||
pass
|
||||
else:
|
||||
# 使用argparse定义命令行参数
|
||||
parser = argparse.ArgumentParser(description='命令行参数')
|
||||
parser.add_argument('--env', type=str, default='dev', help='运行环境')
|
||||
# 解析命令行参数
|
||||
args = parser.parse_args()
|
||||
# 设置环境变量，如果未设置命令行参数，默认APP_ENV为dev
|
||||
os.environ['APP_ENV'] = args.env if args.env else 'dev'
|
||||
# 读取运行环境
|
||||
run_env = os.environ.get('APP_ENV', '')
|
||||
# 运行环境未指定时默认加载.env.dev
|
||||
env_file = '.env.dev'
|
||||
# 运行环境不为空时按命令行参数加载对应.env文件
|
||||
if run_env != '':
|
||||
env_file = f'.env.{run_env}'
|
||||
# 加载配置
|
||||
load_dotenv(env_file)
|
||||
|
||||
|
||||
# 实例化获取配置类
|
||||
get_config = GetConfig()
|
||||
# 海康平台配置
|
||||
HaiKangConfig = get_config.get_haikang_config()
|
||||
|
||||
0
util/__init__.py
Normal file
0
util/__init__.py
Normal file
BIN
util/__pycache__/__init__.cpython-310.pyc
Normal file
BIN
util/__pycache__/__init__.cpython-310.pyc
Normal file
Binary file not shown.
BIN
util/__pycache__/haikang_util.cpython-310.pyc
Normal file
BIN
util/__pycache__/haikang_util.cpython-310.pyc
Normal file
Binary file not shown.
453
util/haikang_util.py
Normal file
453
util/haikang_util.py
Normal file
@ -0,0 +1,453 @@
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import base64
|
||||
import requests
|
||||
import httpx
|
||||
from fastapi import HTTPException
|
||||
from datetime import datetime, timezone
|
||||
from email.utils import format_datetime
|
||||
from urllib.parse import urlparse
|
||||
from dateutil.relativedelta import relativedelta
|
||||
|
||||
from config.env import HaiKangConfig
|
||||
# from utils.log_util import logger
|
||||
|
||||
|
||||
class HaikangUtil:
|
||||
"""
|
||||
海康平台工具类
|
||||
"""
|
||||
|
||||
# def __init__(self):
|
||||
# self.HAIKANG_URL = HaiKangConfig.HAIKANG_URL
|
||||
# self.HAIKANG_PORT = HaiKangConfig.HAIKANG_PORT
|
||||
# self.HAIKANG_AK = HaiKangConfig.HAIKANG_AK
|
||||
# self.HAIKANG_SK = HaiKangConfig.HAIKANG_SK
|
||||
# 通用请求头
|
||||
@classmethod
|
||||
def build_signed_headers(cls, method, url, body, app_key, app_secret):
|
||||
"""
|
||||
返回一个 dict，包含所有签名请求所需的 header。
|
||||
|
||||
参数：
|
||||
- method: HTTP 方法，例如 "POST"
|
||||
- url: URL完整url(代码中提取相对位置, 保留 path 和 query）
|
||||
- body: 请求主体字符串（如 JSON），可为空。
|
||||
- app_key: AK
|
||||
- app_secret: SK
|
||||
"""
|
||||
# 1. 基本 headers
|
||||
accept = "*/*"
|
||||
content_type = "application/json"
|
||||
|
||||
# 2. 计算 MD5（可选，如果 body 存在）
|
||||
content_md5 = ""
|
||||
if body:
|
||||
|
||||
# .digest()返回原始二进制md5
|
||||
md5_digest = hashlib.md5(body.encode("utf-8")).digest()
|
||||
content_md5 = base64.b64encode(md5_digest).decode("utf-8")
|
||||
|
||||
# 3. 生成 Date header（HTTP 规范格式）
|
||||
now = datetime.now(timezone.utc)
|
||||
date = format_datetime(now, usegmt=True)
|
||||
|
||||
# 4. 构造 httpHeaders 部分
|
||||
http_headers_str = "\n".join(
|
||||
[method.upper(), accept, content_md5, content_type, date, ""]
|
||||
)
|
||||
|
||||
# 5. 自定义 headers 部分
|
||||
custom_headers_str = f"x-ca-key:{app_key}\n"
|
||||
|
||||
# 6. 拼接 path + query
|
||||
parsed = urlparse(url)
|
||||
path_and_query = parsed.path
|
||||
if parsed.query:
|
||||
path_and_query += "?" + parsed.query
|
||||
|
||||
# 7. 构造签名字符串
|
||||
string_to_sign = http_headers_str + custom_headers_str + path_and_query
|
||||
|
||||
# 8. 使用 HmacSHA256 + Base64 签名
|
||||
h = hmac.new(
|
||||
app_secret.encode("utf-8"), string_to_sign.encode("utf-8"), hashlib.sha256
|
||||
)
|
||||
signature = base64.b64encode(h.digest()).decode("utf-8")
|
||||
|
||||
# 9. 返回完整 headers
|
||||
headers = {
|
||||
"Accept": accept,
|
||||
"Content-MD5": content_md5,
|
||||
"Content-Type": content_type,
|
||||
"Date": date,
|
||||
"X-Ca-Key": app_key,
|
||||
"X-Ca-Signature": signature,
|
||||
"X-Ca-Signature-Headers": "x-ca-key",
|
||||
}
|
||||
return headers
|
||||
|
||||
# 发送请求
|
||||
@classmethod
|
||||
async def send_request(cls, method: str, url: str, headers: dict, body: str | None):
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.request(
|
||||
method, url, headers=headers, content=body, timeout=60
|
||||
)
|
||||
response.raise_for_status()
|
||||
# if response.status_code >= 400:
|
||||
# logger.error(f"发送请求失败: {url} , {response.status_code} {response.text}")
|
||||
# raise HTTPException(status_code=response.status_code, detail=response.text)
|
||||
return response.json()
|
||||
|
||||
# 获取access_token
|
||||
@classmethod
|
||||
async def get_access_token(cls):
|
||||
"""获取access_token
|
||||
|
||||
Returns:
|
||||
_type_: json
|
||||
"""
|
||||
|
||||
url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}{HaiKangConfig.HAIKANG_ACCESS_TOKEN_URL}"
|
||||
headers = cls.build_signed_headers(
|
||||
"POST", url, None, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK
|
||||
)
|
||||
back = await cls.send_request("POST", url, headers, None)
|
||||
|
||||
if back["code"] == "0":
|
||||
# logger.info("获取access_token成功")
|
||||
return [True, back]
|
||||
# return back["data"]["access_token"],back["data"]["token_type"] ,back["data"]["expires_in"]
|
||||
else:
|
||||
# logger.error("获取access_token失败", back["code"], back["msg"])
|
||||
return [False, back]
|
||||
|
||||
# 查询门禁点列表v2
|
||||
@classmethod
|
||||
async def get_door_list_v2(cls, pageNo: int = 1, pageSize: int = 10):
|
||||
"""获取门禁点列表v2
|
||||
|
||||
Args:
|
||||
pageNo (int, optional): 页码. Defaults to 1.
|
||||
pageSize (int, optional): 每页个数. Defaults to 10.
|
||||
|
||||
Returns:
|
||||
_type_: _description_
|
||||
"""
|
||||
url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}{HaiKangConfig.HAIKANG_DOOR_SEARCH}"
|
||||
body_dict = {
|
||||
"pageNo": max(pageNo, 1),
|
||||
"pageSize": min(pageSize, 999),
|
||||
}
|
||||
body_json = json.dumps(body_dict, separators=(",", ":"))
|
||||
|
||||
headers = cls.build_signed_headers(
|
||||
"POST", url, body_json, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK
|
||||
)
|
||||
|
||||
back = await cls.send_request("POST", url, headers, body_json)
|
||||
|
||||
if back["code"] == "0":
|
||||
# logger.info("查询门禁点列表成功")
|
||||
return [True, back["data"]]
|
||||
else:
|
||||
# logger.error("查询门禁点列表失败 ", back["code"], back["msg"])
|
||||
return [False, back["code"], back["msg"]]
|
||||
|
||||
# 查询门禁状态
|
||||
@classmethod
|
||||
async def get_door_status(cls, door_index_codes: list):
|
||||
"""查询门禁状态
|
||||
|
||||
Args:
|
||||
door_index_codes (list): 门禁点唯一标识
|
||||
"""
|
||||
url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}{HaiKangConfig.HAIKANG_DOOR_STATES_URL}"
|
||||
|
||||
body_dict = {"doorIndexCodes": door_index_codes}
|
||||
body_json = json.dumps(body_dict, separators=(",", ":"))
|
||||
|
||||
headers = cls.build_signed_headers(
|
||||
"POST", url, body_json, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK
|
||||
)
|
||||
|
||||
back = await cls.send_request("POST", url, headers, body_json)
|
||||
|
||||
if back["code"] == "0":
|
||||
# logger.info("查询门禁状态成功")
|
||||
return [
|
||||
True,
|
||||
back["data"]["authDoorList"],
|
||||
back["data"]["noAuthDoorIndexCodeList"],
|
||||
]
|
||||
else:
|
||||
# logger.error("查询门禁状态失败 ", back["code"], back["msg"])
|
||||
return [False, back["code"], back["msg"]]
|
||||
|
||||
# 门禁点反控
|
||||
@classmethod
|
||||
async def door_do_control(cls, door_index_code: list, control_type: int):
|
||||
"""门禁点反控
|
||||
|
||||
Args:
|
||||
door_index_code (list): 门禁点唯一标识
|
||||
control_type (int): 操作类型 0-常开, 1-门闭, 2-门开, 3-常闭, 不允许长闭
|
||||
"""
|
||||
if control_type not in [0, 1, 2]:
|
||||
# logger.error("control_type参数错误 ", control_type)
|
||||
return [False, 400, "control_type参数错误"]
|
||||
|
||||
url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}{HaiKangConfig.HAIKANG_DOOR_DOCONTROL_URL}"
|
||||
body_dict = {"doorIndexCodes": door_index_code, "controlType": control_type}
|
||||
body_json = json.dumps(body_dict, separators=(",", ":"))
|
||||
|
||||
headers = cls.build_signed_headers(
|
||||
"POST", url, body_json, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK
|
||||
)
|
||||
|
||||
back = await cls.send_request("POST", url, headers, body_json)
|
||||
|
||||
"""
|
||||
{
|
||||
"code": "0",
|
||||
"msg": "success",
|
||||
"data": [
|
||||
{
|
||||
"doorIndexCode": "2c95c028a809448f962a969e3ab34f",
|
||||
"controlResultCode": 0, # 0表示反控成功, 其他表示失败, 门禁控制是否成功主要看这个值
|
||||
"controlResultDesc": "success",
|
||||
}
|
||||
],
|
||||
},
|
||||
"""
|
||||
|
||||
if back["code"] == "0":
|
||||
# logger.info(
|
||||
# f"执行门禁控制接口成功 door_index_code:{door_index_code} control_type:{control_type} "
|
||||
# )
|
||||
return [True, back["data"]]
|
||||
else:
|
||||
# logger.error(f"执行门禁控制失败 ", back["code"], back["msg"])
|
||||
return [False, back["code"], back["msg"]]
|
||||
|
||||
@classmethod
|
||||
# 查询门禁点事件v2
|
||||
async def query_door_events_v2(cls, door_index_code: list, **args):
|
||||
"""查询门禁点事件v2
|
||||
|
||||
Args:
|
||||
door_index_code (str): 门禁唯一标识
|
||||
"""
|
||||
|
||||
url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}{HaiKangConfig.HAIKANG_DOOR_ENVENTS_URL}"
|
||||
|
||||
# current_time = datetime.now(timezone.utc)
|
||||
# three_months_ago = current_time - relativedelta(months=3)
|
||||
|
||||
# # ISO8601 时间格式
|
||||
# current_time_iso = current_time.isoformat()
|
||||
# three_months_ago_iso = three_months_ago.isoformat()
|
||||
|
||||
body_dict = {
|
||||
"pageNo": max(args.get("pageNo", 1), 1),
|
||||
"pageSize": min(args.get("pageSize", 10), 999),
|
||||
"doorIndexCode": door_index_code,
|
||||
# 排序字段
|
||||
"sort": "eventTime",
|
||||
# 倒序返回
|
||||
"order": "desc",
|
||||
}
|
||||
|
||||
if args.get("startTime") and args.get('startTime') is not None:
|
||||
body_dict["startTime"] = args.get("startTime")
|
||||
if args.get("endTime") and args.get('endTime') is not None:
|
||||
body_dict["endTime"] = args.get("endTime")
|
||||
if args.get("eventType") and args.get('eventType') is not None:
|
||||
body_dict["eventType"] = args.get("eventType")
|
||||
if args.get("personName") and args.get('personName') is not None:
|
||||
body_dict["personName"] = args.get("personName")
|
||||
|
||||
body_json = json.dumps(body_dict, separators=(",", ":"))
|
||||
|
||||
headers = cls.build_signed_headers(
|
||||
"POST", url, body_json, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK
|
||||
)
|
||||
|
||||
back = await cls.send_request("POST", url, headers, body_json)
|
||||
|
||||
if back["code"] == "0":
|
||||
# logger.info(f"获取门禁事件成功")
|
||||
return [True, back["data"]]
|
||||
else:
|
||||
# logger.error(f"获取门禁事件失败 ", back["code"], back["msg"])
|
||||
return [False, back["code"], back["msg"]]
|
||||
|
||||
# 查看门禁在线状态
|
||||
@classmethod
|
||||
async def door_online_status(cls, indexCodes: list):
|
||||
url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}{HaiKangConfig.HAIKANG_DOOR_ONLINE_STATUS}"
|
||||
|
||||
body_dict = {
|
||||
"indexCodes": indexCodes
|
||||
}
|
||||
|
||||
body_json = json.dumps(body_dict, separators=(",", ":"))
|
||||
|
||||
headers = cls.build_signed_headers(
|
||||
"POST", url, body_json, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK
|
||||
)
|
||||
|
||||
back = await cls.send_request("POST", url, headers, body_json)
|
||||
|
||||
if back["code"] == "0":
|
||||
# logger.info(f"获取门禁在线状态成功")
|
||||
return [True, back["data"]]
|
||||
else:
|
||||
# logger.error(f"获取门禁在线状态失败 ", back["code"], back["msg"])
|
||||
return [False, back["code"], back["msg"]]
|
||||
|
||||
# 人脸分组1vN检索
|
||||
@classmethod
|
||||
async def face_group_1vN_search(
|
||||
cls,
|
||||
facePicBinaryData: str, # base64编码后的字符串
|
||||
pageNo: int = 1,
|
||||
pageSize: int = 20,
|
||||
searchNum: int = 99,
|
||||
minSimilarity: int = 50,
|
||||
faceGroupIndexCodes: list[str] = None,
|
||||
):
|
||||
"""人脸分组1vN检索
|
||||
|
||||
Args:
|
||||
facePicBinaryData (str): 图片二值化后,base64编码的字符串
|
||||
pageSize (int, optional): 每页个数 Defaults to 20.
|
||||
searchNum (int, optional): 最大搜索返回数. Defaults to 99.
|
||||
minSimilarity (int, optional): 最小相似度. Defaults to 50.
|
||||
faceGroupIndexCodes (list[str], optional): 查询人脸分组. Defaults to None.
|
||||
|
||||
Returns:
|
||||
_type_: _description_
|
||||
"""
|
||||
url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}{HaiKangConfig.HAIKANG_APPLICATION_ONETOMANY_URL}"
|
||||
|
||||
body_dict = {
|
||||
"facePicBinaryData": facePicBinaryData,
|
||||
"pageNo": pageNo,
|
||||
"pageSize": pageSize,
|
||||
"searchNum": searchNum,
|
||||
"minSimilarity": minSimilarity,
|
||||
"faceGroupIndexCodes": faceGroupIndexCodes,
|
||||
}
|
||||
|
||||
body_json = json.dumps(body_dict, separators=(",", ":"))
|
||||
|
||||
headers = cls.build_signed_headers(
|
||||
"POST", url, body_json, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK
|
||||
)
|
||||
|
||||
back = await cls.send_request("POST", url, headers, body_json)
|
||||
|
||||
if back["code"] == "0":
|
||||
# logger.info(f"获取人脸分组检索成功")
|
||||
return [True, back["data"]]
|
||||
else:
|
||||
# logger.error(f"获取人脸分组检索失败 ", back["code"], back["msg"])
|
||||
return [False, back["code"], back["msg"]]
|
||||
|
||||
# 人脸评分
|
||||
@classmethod
|
||||
async def face_picture_check(
|
||||
cls,
|
||||
facePicBinaryData: str, #
|
||||
):
|
||||
"""人脸评分
|
||||
|
||||
Args:
|
||||
facePicBinaryData (str): 人脸图的二进制数据经过Base64编码后的字符串
|
||||
|
||||
Returns:
|
||||
_type_: _description_
|
||||
"""
|
||||
url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}{HaiKangConfig.HAIKANG_PICTURE_CHECK_URL}"
|
||||
|
||||
body_dict = {"facePicBinaryData": facePicBinaryData}
|
||||
|
||||
body_json = json.dumps(body_dict, separators=(",", ":"))
|
||||
|
||||
headers = cls.build_signed_headers(
|
||||
"POST", url, body_json, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK
|
||||
)
|
||||
|
||||
back = await cls.send_request("POST", url, headers, body_json)
|
||||
|
||||
if back["code"] == "0":
|
||||
# logger.info(f"获取人脸评分成功")
|
||||
return [True, back["data"]]
|
||||
else:
|
||||
# logger.error(f"获取人脸评分失败 ", back["code"], back["msg"])
|
||||
return [False, back["code"], back["msg"]]
|
||||
|
||||
# 按条件查询人员识别事件
|
||||
@classmethod
|
||||
async def get_face_capture_list_event(cls, startTime, endTime):
|
||||
"""按条件查询人员识别事件, 暂无实现.
|
||||
|
||||
Args:
|
||||
startTime (_type_): _description_
|
||||
endTime (_type_): _description_
|
||||
"""
|
||||
pass
|
||||
|
||||
# 按条件查询人脸分组
|
||||
@classmethod
|
||||
async def get_face_group(cls):
|
||||
url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}{HaiKangConfig.HAIKANG_FACE_GROUP_URL}"
|
||||
|
||||
headers = cls.build_signed_headers(
|
||||
"POST", url, None, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK
|
||||
)
|
||||
|
||||
back = await cls.send_request("POST", url, headers, None)
|
||||
|
||||
if back["code"] == "0":
|
||||
# logger.info(f"按条件查询人脸成功")
|
||||
return [True, back["data"]]
|
||||
else:
|
||||
# logger.error(f"按条件查询人脸失败 ", back["code"], back["msg"])
|
||||
return [False, back["code"], back["msg"]]
|
||||
|
||||
# 查询访客预约记录, visitorStatus: 0 待审核, 1 正常, 2 迟到, 3 失效, 4 审核退回, 9 审核失效, 10 邀约中, 11 邀约失效
|
||||
@classmethod
|
||||
async def query_visitor_record(
|
||||
cls, **args
|
||||
):
|
||||
"""查询访客预约记录
|
||||
|
||||
Args:
|
||||
pageNo (int, optional): 页码. Defaults to 1.
|
||||
pageSize (int, optional): 每页个数. Defaults to 10.
|
||||
visitorStatus (int, optional): 访客状态. Defaults to 1.
|
||||
|
||||
Returns:
|
||||
_type_: _description_
|
||||
"""
|
||||
url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}{HaiKangConfig.HAIKANG_VISITOR_RECORD_SEARCH}"
|
||||
body_dict = args
|
||||
|
||||
body_json = json.dumps(body_dict, separators=(",", ":"))
|
||||
|
||||
headers = cls.build_signed_headers(
|
||||
"POST", url, body_json, HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK
|
||||
)
|
||||
|
||||
back = await cls.send_request("POST", url, headers, body_json)
|
||||
|
||||
if back["code"] == "0":
|
||||
# logger.info(f"查询访客预约记录成功")
|
||||
return [True, back["data"]]
|
||||
else:
|
||||
# logger.error(f"查询访客预约记录失败 ", back["code"], back["msg"])
|
||||
return [False, back["code"], back["msg"]]
|
||||
Loading…
Reference in New Issue
Block a user