365 lines
14 KiB
Python
365 lines
14 KiB
Python
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(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)
|
||
|
||
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: str, **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"):
|
||
body_dict["startTime"] = args.get("startTime")
|
||
if args.get("endTime"):
|
||
body_dict["endTime"] = args.get("endTime")
|
||
if args.get("eventType"):
|
||
body_dict["eventType"] = args.get("eventType")
|
||
|
||
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"]]
|
||
|
||
# 人脸评分
|
||
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"]] |