diff --git a/ruoyi-fastapi-backend/.env.dev b/ruoyi-fastapi-backend/.env.dev index a90df81..206597d 100644 --- a/ruoyi-fastapi-backend/.env.dev +++ b/ruoyi-fastapi-backend/.env.dev @@ -72,38 +72,51 @@ REDIS_DATABASE = 3 #------------------海康平台配置------------------ -HAIKANG_URL = '' +HAIKANG_URL = 'https://192.168.89.230' HAIKANG_PORT = 443 -HAIKANG_AK = '' -HAIKANG_SK = '' -# 获取access_token + +HAIKANG_AK = '29862915' +HAIKANG_SK = '3QlsmcyadBqu4OpVoxsJ' HAIKANG_ACCESS_TOKEN_URL = '/api/v1/oauth/token' -# 查看门禁状态(门的状态) -HAIKANG_DOOR_STATES_URL = '/api/v1/door/states' -# 门禁点反控 +# HAIKANG_DOOR_STATES_URL = '/api/v1/door/states' +HAIKANG_DOOR_STATES_URL = '/api/acs/v1/door/states' HAIKANG_DOOR_DOCONTROL_URL = '/api/acs/v1/door/doControl' -# 门禁点事件v2 HAIKANG_DOOR_ENVENTS_URL = '/api/acs/v2/door/events' -# 门禁点列表v2 HAIKANG_DOOR_SEARCH = '/api/resource/v2/door/search' -# 查看门禁点在线状态 HAIKANG_DOOR_ONLINE_STATUS = '/api/nms/v1/online/acs_device/get' - - -# 人脸分组1vN检索 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_FACECAPATURE_SEARCH = '/api/frs/v1/event/face_capture/search' - -# 查询访客预约记录 HAIKANG_VISITOR_RECORD_SEARCH = '/api/visitor/v2/appointment/records' +HAIKANG_VISITOR_RECORD_PICTURES = '/api/visitor/v1/record/pictures' + +HAIKANG_PERSON_LIST = '/api/resource/v2/person/personList' +HAIKANG_GET_PERSON_PICTURE = '/api/resource/v1/person/picture' + +HAIKANG_FACE_GROUP_ADDITION = '/api/frs/v1/face/group/single/addition' + +HAIKANG_FACE_SIGLE_ADDITION = '/api/frs/v1/face/single/addition' +HAIKANG_FACE_DELETE = '/api/frs/v1/face/deletion' + +HAIKANG_FACE_SAVE_PATH = "./face_images" + + +# # 人脸分组1vN检索 +# HAIKANG_APPLICATION_ONETOMANY_URL = '/api/frs/v1/application/oneToMany' + +# # 人脸评分 +# HAIKANG_PICTURE_CHECK_URL = '/api/frs/v1/face/picture/check' + +# # 按条件查询人脸分组 +# HAIKANG_FACE_GROUP_URL = '/api/frs/v1/face/group' + +# # 查询人脸识别事件 +# HAIKANG_FACECAPATURE_SEARCH = '/api/frs/v1/event/face_capture/search' + +# # 查询访客预约记录 +# HAIKANG_VISITOR_RECORD_SEARCH = '/api/visitor/v2/appointment/records' diff --git a/ruoyi-fastapi-backend/config/env.py b/ruoyi-fastapi-backend/config/env.py index b0a00c6..ba76f4f 100644 --- a/ruoyi-fastapi-backend/config/env.py +++ b/ruoyi-fastapi-backend/config/env.py @@ -154,13 +154,14 @@ class HaiKangSettings: """ 海康平台配置 """ - HAIKANG_URL = 'http://10.0.0.202' - # HAIKANG_PORT = 443 - HAIKANG_PORT = 9909 - HAIKANG_AK = '' - HAIKANG_SK = '' + HAIKANG_URL = 'https://192.168.89.230' + HAIKANG_PORT = 443 + + HAIKANG_AK = '29862915' + HAIKANG_SK = '3QlsmcyadBqu4OpVoxsJ' HAIKANG_ACCESS_TOKEN_URL = '/api/v1/oauth/token' - HAIKANG_DOOR_STATES_URL = '/api/v1/door/states' + # HAIKANG_DOOR_STATES_URL = '/api/v1/door/states' + HAIKANG_DOOR_STATES_URL = '/api/acs/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' @@ -169,7 +170,19 @@ class HaiKangSettings: 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' + HAIKANG_VISITOR_RECORD_PICTURES = '/api/visitor/v1/record/pictures' + + HAIKANG_PERSON_LIST = '/api/resource/v2/person/personList' + HAIKANG_GET_PERSON_PICTURE = '/api/resource/v1/person/picture' + + HAIKANG_FACE_GROUP_ADDITION = '/api/frs/v1/face/group/single/addition' + + HAIKANG_FACE_SIGLE_ADDITION = '/api/frs/v1/face/single/addition' + HAIKANG_FACE_DELETE = '/api/frs/v1/face/deletion' + + HAIKANG_FACE_SAVE_PATH = "./face_images" diff --git a/ruoyi-fastapi-backend/utils/haikang_util.py b/ruoyi-fastapi-backend/utils/haikang_util.py index 338d6bb..69502ef 100644 --- a/ruoyi-fastapi-backend/utils/haikang_util.py +++ b/ruoyi-fastapi-backend/utils/haikang_util.py @@ -4,6 +4,7 @@ import json import base64 import requests import httpx +import os from fastapi import HTTPException from datetime import datetime, timezone from email.utils import format_datetime @@ -19,6 +20,604 @@ 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, json=True): + + + + async with httpx.AsyncClient(verify=False) as client: + response = await client.request( + method, url, headers=headers, content=body, timeout=60 + ) + if response.status_code in (301, 302, 303, 307, 308): + location = response.headers.get("Location") + # 可能需要处理成 GET 请求或保留原方法 + response = await client.get(location, headers=headers) + 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) + if json: + return response.json() + return response + + # 获取access_token + @classmethod + async def get_access_token(cls): + """获取access_token + + Returns: + _type_: json + """ + + url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}/artemis{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}/artemis{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}/artemis{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}/artemis{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}/artemis{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}/artemis{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}/artemis{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}/artemis{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}/artemis{HaiKangConfig.HAIKANG_FACE_GROUP_URL}" + + headers = cls.build_signed_headers( + "POST", url, json.dumps({}), HaiKangConfig.HAIKANG_AK, HaiKangConfig.HAIKANG_SK + ) + + back = await cls.send_request("POST", url, headers, json.dumps({})) + + 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}/artemis{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"]] + + # 获取访客记录中的图片 + @classmethod + async def query_visitor_record_pictures(cls, svrIndexCode: str, picUri: str, save_path): + """获取访客记录中的图片 + + Args: + visitorRecordIndexCode (str): 访客记录索引码 + + Returns: + _type_: _description_ + """ + url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}/artemis{HaiKangConfig.HAIKANG_VISITOR_RECORD_PICTURES}" + + body_dict = { + "svrIndexCode": svrIndexCode, + "picUri": picUri, + } + + 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, False) + + os.makedirs(save_path, exist_ok=True) + save_path = os.path.join(save_path, picUri.split('/')[-1]+".jpg") + + with open(save_path, 'wb') as f: + for chunk in back.iter_bytes(chunk_size=8192): + if chunk: # 忽略 keep-alive 的空 chunk + f.write(chunk) + + + + # 获取人员列表 + @classmethod + async def get_person_list(cls, pageNo: int = 1, pageSize: int = 10): + url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}/artemis{HaiKangConfig.HAIKANG_PERSON_LIST}" + body_dict = { + "pageNo": pageNo, + "pageSize": pageSize, + } + 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": + return [True, back["data"]] + else: + return [False, back["code"], back["msg"]] + + # 提取人员图片 + @classmethod + async def get_person_picture(cls, serverIndexCode, picUri, save_path): + + url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}/artemis{HaiKangConfig.HAIKANG_GET_PERSON_PICTURE}" + body_dict = { + "serverIndexCode": serverIndexCode, + "picUri": picUri, + } + + 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, False) + + os.makedirs(save_path, exist_ok=True) + save_path = os.path.join(save_path, picUri.split('/')[-1]+".jpg") + + with open(save_path, 'wb') as f: + for chunk in back.iter_bytes(chunk_size=8192): + if chunk: # 忽略 keep-alive 的空 chunk + f.write(chunk) + + + # 单个添加人脸分组 + @classmethod + async def face_group_addition(cls, name, description): + url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}/artemis{HaiKangConfig.HAIKANG_FACE_GROUP_ADDITION}" + body_dict = { + "name": name, + "description": description, + } + 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": + return [True, back["data"]] + else: + return [False, back["code"], back["msg"]] + + # 向分组中单个添加人脸图片 + @classmethod + async def face_single_addition(cls, faceGroupIndexCode: str, faceInfo: dict, facePic: dict): + url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}/artemis{HaiKangConfig.HAIKANG_FACE_SIGLE_ADDITION}" + + + body_dict = { + "faceGroupIndexCode": faceGroupIndexCode, + "faceInfo": { + "name": "测1" + }, + "facePic": { + #"facePicBinaryData": facePic["facePicBinaryData"] + "facePicUrl":"https://pic-image.yesky.com/uploadImages/newPic/2023/271/38/A3A130JG06X2.png" + } + } + body_json = json.dumps(body_dict, separators=(",", ":")) + + # print(body_json) + + 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": + return [True, back["data"]] + else: + return [False, back["code"], back["msg"]] + + # 批量删除人脸 + @classmethod + async def face_delete(cls, faceGroupIndexCode: str, indexCodes: list): + url = f"{HaiKangConfig.HAIKANG_URL}:{HaiKangConfig.HAIKANG_PORT}/artemis{HaiKangConfig.HAIKANG_FACE_DELETE}" + + body_dict = { + "faceGroupIndexCode": faceGroupIndexCode, + "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": + return [True, back["data"]] + else: + return [False, back["code"], back["msg"]] + """ + 海康平台工具类 + """ + # def __init__(self): # self.HAIKANG_URL = HaiKangConfig.HAIKANG_URL # self.HAIKANG_PORT = HaiKangConfig.HAIKANG_PORT