From ff4ccc8fa70fd5da8fc2a7c9f357ecb9fd536e09 Mon Sep 17 00:00:00 2001 From: haotianmingyue <2421912570@qq.com> Date: Mon, 15 Sep 2025 11:06:32 +0800 Subject: [PATCH] =?UTF-8?q?=E5=88=9D=E5=A7=8B=E5=8C=96=E6=B5=B7=E5=BA=B7ap?= =?UTF-8?q?i=E6=B5=8B=E8=AF=95=E5=BA=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- 001测试海康api.py | 96 ++++ 002简单http服务.py | 536 ++++++++++++++++++ config/__init__.py | 0 config/__pycache__/__init__.cpython-310.pyc | Bin 0 -> 139 bytes config/__pycache__/env.cpython-310.pyc | Bin 0 -> 3055 bytes config/env.py | 86 +++ util/__init__.py | 0 util/__pycache__/__init__.cpython-310.pyc | Bin 0 -> 137 bytes util/__pycache__/haikang_util.cpython-310.pyc | Bin 0 -> 9815 bytes util/haikang_util.py | 453 +++++++++++++++ 10 files changed, 1171 insertions(+) create mode 100644 001测试海康api.py create mode 100644 002简单http服务.py create mode 100644 config/__init__.py create mode 100644 config/__pycache__/__init__.cpython-310.pyc create mode 100644 config/__pycache__/env.cpython-310.pyc create mode 100644 config/env.py create mode 100644 util/__init__.py create mode 100644 util/__pycache__/__init__.cpython-310.pyc create mode 100644 util/__pycache__/haikang_util.cpython-310.pyc create mode 100644 util/haikang_util.py diff --git a/001测试海康api.py b/001测试海康api.py new file mode 100644 index 0000000..234cae4 --- /dev/null +++ b/001测试海康api.py @@ -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"])) + diff --git a/002简单http服务.py b/002简单http服务.py new file mode 100644 index 0000000..1a54782 --- /dev/null +++ b/002简单http服务.py @@ -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 """ + + + + + + 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) diff --git a/config/__init__.py b/config/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/config/__pycache__/__init__.cpython-310.pyc b/config/__pycache__/__init__.cpython-310.pyc new file mode 100644 index 0000000000000000000000000000000000000000..77cde147946652810f52dcc8baaa0e56d83caade GIT binary patch literal 139 zcmd1j<>g`kf`2u~GeGoX5P=LBfgA@QE@lA|DGb33nv8xc8Hzx{2;!HXt5r-!Vtz?x zVqQ!^Wl2VUo=0M4c4A(-V?kz2a(-S~W_nC~d}dx|NqoFsLFFwDo80`A(wtN~kU7On IK!Sw<03MYei~s-t literal 0 HcmV?d00001 diff --git a/config/__pycache__/env.cpython-310.pyc b/config/__pycache__/env.cpython-310.pyc new file mode 100644 index 0000000000000000000000000000000000000000..668b9a54ea3d80c57d07110abbbc24eae0b087e4 GIT binary patch literal 3055 zcmZ`5O>f-Bb@;JdE|R1=h5+Y@y-bO^-6LAf&3ef=50ILI5B|2a_ zU^Cv#9>TNU?7dU_2+u)(8v5sm&_j*dSKuA88bV$TB8S){O1y9j&`Lm4+jZQ52T0wG zhNCp`h`sK60Rf8smJ@qDFG>7xBL=*@;wK(;0+*Sa{scg-L5o)ac!(_>mMvh3h*%~P zE)f|k(8{EQRifYuDPxVOxJoKmCmODiDy|b9&yX6PC3Sp?%-}gP3$CBaXRbT`vJ-Ah z{75gr9NJ6pJ^I^&gU=oueEQdeyZ68P#hou7{+W}r*+G(wt~Q%N~zx^`OU4@nvs(xgJu%l^1>q`i6YvJ6DRTF?0GKfxG|IF3=*}XFriV9oju|$6g+Rs z3zImzz!}twqm;T{b8F0vJ%_r3>=NgOLvAt(13&bdFo8|HEgz&eykso0PPvFP8U?=V zBz_b&qtNR|t4_F`UF25ABI^z)j(j&ssn>)26O(q40JmOMgXU%J-gpoHZ87ycEVZg zbh>u0Z}iO`XFPw5vDj&KtbVt%!g+`0xRPnTV_N+}>+~_Lo@sPj*Yo41I@U_tGWk?D zdikY^g0Z%?(ry|3cE_?imf7#D8rHjfw(}E_wRWq2qifo&>t<`2vG5BM*4swQY#D0? z=Z~j7GtIP@x}6(q+}?{5`aA7jyAP}GnypTE(Js_JLG%U42oC<+Ft6SmI8ov|;my%@ zGKfORo?DPT#>l_vM&Y`@aT8L1ezZ-?p!7Rv@yh^IBnz@6ivXYcs{)sQcH*A{BTr9! zNa9jH(>0df4x*nHFH1dG;?{S-ASfLiSziY^;NY%Y$Hbvpcjh zbD^_n`BvZj8QwHr$9*eUU2I=j9a;^AR=^v|DmSe%paa0OY_gTkkLTvmH09~+C%TF_#) z>K$|s?IU)jfJK4b0QHeWW2gs#Z##4&{zS;-*bCOD*3OpA65J2{#I_$YQ_Q9=WOIiO z%}>}*OwU|k(|@f1NYEaiIQdNYAW5pDD(&^qbZ9281>Pz1GIMD)Kw z>}%$tQpzqIx%i~uGo7WcfObPjUxn_3o8u6j+VQc6ECinI;W%2y9{z{PGUF%xJ1*ve z#Nj8yi0vUi%N-aHAwio05znJ7MAh(OB5q1VBJv$!7tVPT;wmW-h$o`nfrkd1yt0pW zg&kqDOfNAQ3PefN%?hc!iCBM6V~>e_aaSbOU1>)o+AWE`LG&H*_vjDeu9Q@Fq&)#* zrv|Z8Prm~$9PQ1YK6>=w2M51;_^;1De!Tz7gS+oN`t|)oE-t5Vr$scREE>`o04GEt z5Ua$CUFwh61(%gBUxu=jOK=&cEN&lv@d23q^8LRY{Bf@#<#Mt;^71nA)}1s+a(Und zBV2|GY(p)lEHg88p4Bzb&q{n`HS}DHx8q!fFyG2mfIRZ^xmto-3D9j}YEXu;IA?FI0qAcj={@QMn&6VXe* z32Og?7PEw|mqbyQ<^)wdBi2P(R7FkD#q8Ws2p4`#a@A^RPpfS7Ww1ph2EN4ryme5n zvxkiD4n4R72%mP!6@Fjv>UI%i{|+sl1Hh`AD$7661X1oT{iBZX%rWg*V*7G6{N zgji##TjVxhQeGRs0gGzLSBPul%pB7W@4#Z6MGg<;Irq9RpL1^w_xKd;Z# N&(>At$7)>{{|_UDG#UT^ literal 0 HcmV?d00001 diff --git a/config/env.py b/config/env.py new file mode 100644 index 0000000..6e5050c --- /dev/null +++ b/config/env.py @@ -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() + diff --git a/util/__init__.py b/util/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/util/__pycache__/__init__.cpython-310.pyc b/util/__pycache__/__init__.cpython-310.pyc new file mode 100644 index 0000000000000000000000000000000000000000..393ec5bae05717ed45ee587f52fa4f12b3ba58cb GIT binary patch literal 137 zcmd1j<>g`kg3mR_GeGoX5P=LBfgA@QE@lA|DGb33nv8xc8Hzx{2;!HHt5r-!Vtz?x zVqQ!^Wl2VUo=0M4c4A(-V?kz2X-Q^IOniK1US>&ryk0@&Ee@O9{FKt1R6CF<#Y{kg Gg#iH6UmnK* literal 0 HcmV?d00001 diff --git a/util/__pycache__/haikang_util.cpython-310.pyc b/util/__pycache__/haikang_util.cpython-310.pyc new file mode 100644 index 0000000000000000000000000000000000000000..21cad5b348d7bbc14c605edb2f3a5015ade5405a GIT binary patch literal 9815 zcmc&)>2n+BdEX0*#mPf7b@+mu!-6g9vgO*W8=ImiiIxm1BxT1NwU`xlMMA>C?k;4D z1I|!~uUJm!yFT+=#;<(>aq4p9{cpA5L28qt;e1=oqC$I9X zET{r|imJ$-k}9#MPxav`WaO-0_471F%qUq!Rk8s!z{*l4m<_2RmiJ}C*@zlpc{vl! zHmOZ4@6R-6Thtbw+sWyQ9yrPAffIt-s`CR}JUE9nafvwZM0O1h?(KOgWsKPAd@e3H zimus)oz5DLLg{EeXP{ft!MvH(>|~`8RZ79kjA*7s%@NbcXmT$GBfdQcBt@{g@3t^nb}h3COpj3Pv@*;{XxF1|ZGH9kG{%b1={*`MD$RXX)@ z>D;f%uaB26y*~5dXQgweXRlr>|N1y;<}QuTTsRpUHZYJ_IUaHN1xxk zP_5~RWes~cuXn~sl49kzJ}Lj^Jf@%i?2XcU$72ud?Hd?;pr$vZrZX1X*T1K9w;k({ru+ZHB%nQ>qk3d?&_zfZcKmr($Lir(?&*;hYifqwY#C#GE$~tcg6)34^9gVWwKzb;r*#*-WZLcqP>|%Rn#oHYJux1Ts z(g##At8aB=BbUnShAQjnA;YpA`G95_k3H%H4m|d#+vLc)L0wD&Zb!`JUo<7;RIy;E zOdr)rhw|y1<137e7^X>b=18U7gJET34gA?4*{{lWgT`6a&(^p`@KpTA-}M532#nFcO21E=@n+Oah?W zn|eC;4{Lclt>yNQ90l*^JdElZN$)QJr_jnUD@lb7BS)RK1BG-(XLB2RWkvU3F6&7o zQbghlzEwtqD~UqP}3#~lEBaZ)E-2a5z${;z$6E042k8-_GzPq>!u`yrKcStDmMWHAbC7suW z(>}LtOfJfVNjy++LYm~B=LC$A@H-*8J$P<_+w9F`ZG4gE(U<35z^$VtE+*!P2yu}u zPxX77t@=s(<_mV*=lE!IUs6L|*3n$5JClZ_sD9JH-dVQVY-(xCNJ2kj+p%F5*c8GT z_nWIQ?P6cA$|-ja#Wbvue9ki1zPd>agPgEsfH%B1>_?5&j6@i36(l~$hrhzkNSpp^ zbX6_wYlozOz153^h)B%HdjJ_C?;#!vf}X-f4ykyNdq*1Mi`)eCf05Tk)cJ7BlfE$l zwSKBC!ZMhYKvZ!THzpQEU0TiQzQcmqq02>aNEm?1ai4LeumdleJ#!6kN-||x;O)ak zZXU?{4Zugv>a3z2;UJT@WM_;BSZ%n?cRCWGc*2ng9~^$u;EuRrl46B=?&{s$mDrgC zpgZA8VQ*jmpt`D3>gw+985l?o_U-OTP(O1mjSNPNShtN z9mIyqsD*WbTDj(V)UeJsyRVl5NzjLl8{4jOtxY`bw# zD`addX6Iv@H&j`Kc&9?Hg@emi&tg0*4_FQjQ*7EWznVL8WKuV*l$mBLN><-%eixr< zZbmXUgI$UX<|9<`D3T9^+x&Xg{sFHBEd$7}*?H44A4FrqTu1%lNCq7V3PjuCh*@hW z9%fu^K18h#BT-jWxO978Uw?9-r>nnv7rJUMAxY=VZ=;Xn(+k-Vi*dENo@J0cu+SNHPi{q5tDT%Cg&c9@;$91k{f-pYE; z8|)hF8DQv1JZJs|C4{4m9OtRME^1=%`8~?PKljxCVhq+qaL7pGYS0Pvw#YmIfB6;S zt152^Vh|co`a`RK^zKGLt(n1TVug7IbC^kI<}*yiEDfm_@YZ9b>G&hO85lwo;4=Jm za1iNB(u%eEQe-;Og;nCeT37{Fd9YgHKZ4c3LRbxOHNd+4Yk~EjvA}s|IeWT%=KcQz z!!48F2v5lls?dvJD!8=#$6- zLanTBdg^rPrFFst_UuG2U7=1*s@US*I9 zpxuT(u4RrzlBe=@;;bxg!GR zP4cyv*|M|>sgWy3(qJPlTd>-gJMF3xMkWi^%75UE)+Qv-CV}Q)GkLJMpJ zTIG(yy{)rh>U`V9rgnl%!!UJRCxUer%qw{Jk|OxWSuk*pnFWK0r|t(EcOvvBlbOy8 z^C0qQK?-|@8Rp@HYqBP27wYYYB{q41T~LpgeThB2iJs&D%!7RcU$d{CX~5pST&Qei z@{`YH(mBKPuVX+oi&O;mQ$XC}s|B{_eYKXs!U^UmEIXH?gv82eIjE(Kz3J4pbWSsm zBHpDzRt(Kd4JQg&CjvWrAe~KTG&5};F+=hB6(&NGa~4jvHoHZQDSn($&3vl`dcw$pF(suM8+#) z0$&^??r3#jnE`B!i`m58| zPnE`RmQKA@I(KHh8{C_@`4+r3L~^U)(5muZzQ) zxl8ygH~hT7{d?kx`q}Dsdhr=QI*aJ=^v#>4>+jXOl3N>Awg{WJ#@7#_0RFPhxpiHL z+Wn}XI)Njg{|7%cE}H{j$9ZjG4pMO+CC^b(j|a6rYAxz70ED#_1FCE9-aWnDU4y-S ziDX}*XRz%xD+Hni%Ho zXrKZ0*VdxPqO<);puW*r?Szyw7uAz;=EkCCUdpfIdtI)Yy?PRY{*8p&V@rtmg_98R z*;^M&uYC+jDxbbGbK~sntyj0PIKC_D5dAu7ugdB-j;%1J_=b$Ry26-yd%Fks_4g#Z zclC7dUIs&Yii*8^xqR2kIAsJe6`YDBQ^Q8;aGIE|-DSE6+}K9*qE1aM9A4+7QVbB& zxj|eRnPWUVNEQm0Q7E5!z5LE3Y|$!|l>X)?2q2VR`;}{>u8);39525;QM&T(%$bjB zz+_>FqG~0^*16$WSgVUizX*gqJMwG$CRjWPE+30W^y{zk^x4ca1jQW1Osd% z0!$(qT}&zw0-zS8if4cYJOhj(HcMDsZe+VA2H0NV;~ibyJ;|N@ef#z@CSFT()Cc^` z4OG`a)M%AF3>ME59laoRBTKGXl&Wl49U&|BEpsuC--(cLA*<0jxR}-G9Nfu8qH|PR ziO%^NY}?1b&bBQ)j4yK~VD`$b(v{0|mru@Ie{c56N2O2C-wFKw1X1UT0YlFtX)BG; z@BlkKT4;8--!j8ui}i*9BH({K!hS<=y|==Y&-D)U4k9+)-_zaKzugN_FV7R$L(IBd zfc1Pkk61VyOTn#upLp4W*eDKM8u@t<53YU!M3R`3O^yIOX_nth%lbYg{YVn=c1KAj zb8w53Nhg?0X7hR>L-}wL$G2L>Yw;)HccxGsN@X<5a!;Jyqb3n1AHw|54J;~ek~8Xv zIO8y%rkq3=XLciN?xf^dO8O{yo{}F>@&Y9)B|oHOUcy-}Ms;+A+Jf8qDAo7q;XM8>uz{VCv*UfWozB+TA$HyWn@oZqHHhdL{@b*{$Q^M5 i5wxhqDB~zkx_kCLaEp6zp>PUc`Gt8Ld%HiORr+7_Yf8@m literal 0 HcmV?d00001 diff --git a/util/haikang_util.py b/util/haikang_util.py new file mode 100644 index 0000000..f14e51b --- /dev/null +++ b/util/haikang_util.py @@ -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"]]