添加海康服务类方法
This commit is contained in:
parent
10ad997949
commit
cf6a4baa6b
@ -1,12 +1,241 @@
|
||||
from utils.haikang_util import HaikangUtil
|
||||
from module_admin.entity.vo.haikang_vo import VisitorReservationQueryModel
|
||||
import json
|
||||
import base64
|
||||
import asyncio
|
||||
|
||||
class HaiKangService:
|
||||
"""
|
||||
康达服务
|
||||
海康服务
|
||||
"""
|
||||
# 获取门禁列表
|
||||
@classmethod
|
||||
async def get_door_list_service(cls, pageNo: int = 1, pageSize: int = 10):
|
||||
result = await HaikangUtil.get_door_list_v2(pageNo, pageSize)
|
||||
print(result)
|
||||
with open("get_door_list_service.json", "w", encoding="utf-8") as f:
|
||||
f.write(json.dumps(result[1]))
|
||||
|
||||
# 查询访客列表
|
||||
# 查询门禁状态
|
||||
@classmethod
|
||||
async def get_door_status_service(cls, door_index_codes):
|
||||
result = await HaikangUtil.get_door_status(door_index_codes)
|
||||
print(result)
|
||||
with open("get_door_status_service.json", "w", encoding="utf-8") as f:
|
||||
f.write(json.dumps(result[1]))
|
||||
|
||||
# 门禁控制
|
||||
@classmethod
|
||||
async def door_do_control_service(cls, door_index_codes, control_type):
|
||||
result = await HaikangUtil.door_do_control(door_index_codes, control_type)
|
||||
print(result)
|
||||
with open("door_do_control_service.json", "w", encoding="utf-8") as f:
|
||||
f.write(json.dumps(result[1]))
|
||||
|
||||
# 查询门禁点事件
|
||||
@classmethod
|
||||
async def query_door_events_service(cls, 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)
|
||||
with open("query_door_events_service.json", "w", encoding="utf-8") as f:
|
||||
f.write(json.dumps(result[1]))
|
||||
|
||||
# 查看门禁点在线状态
|
||||
@classmethod
|
||||
async def door_online_status_service(cls, door_index_codes):
|
||||
result = await HaikangUtil.door_online_status(door_index_codes)
|
||||
print(result)
|
||||
with open("door_online_status_service.json", "w", encoding="utf-8") as f:
|
||||
f.write(json.dumps(result[1]))
|
||||
|
||||
# 按条件查询人脸分组, 很重要
|
||||
@classmethod
|
||||
async def get_face_group_service(cls):
|
||||
result = await HaikangUtil.get_face_group()
|
||||
print(result)
|
||||
with open("get_face_group_service.json", "w", encoding="utf-8") as f:
|
||||
f.write(json.dumps(result[1]))
|
||||
|
||||
# 人脸分组1vN搜索
|
||||
@classmethod
|
||||
async def face_group_1vN_search_service(cls, image_path,faceGroupIndexCodes:list, pageNo=1, pageSize=5, searchNum:int=5 , minSimilarity=50):
|
||||
|
||||
with open(image_path, 'rb') as f:
|
||||
image_data = f.read()
|
||||
|
||||
encoded_image = base64.b64encode(image_data).decode('utf-8')
|
||||
|
||||
# with open("image_base64.txt", "w", encoding="utf-8") as f:
|
||||
# f.write(encoded_image)
|
||||
|
||||
result = await HaikangUtil.face_group_1vN_search(
|
||||
facePicBinaryData=encoded_image,
|
||||
pageNo=pageNo,
|
||||
pageSize=pageSize,
|
||||
searchNum=searchNum,
|
||||
minSimilarity=minSimilarity,
|
||||
faceGroupIndexCodes=faceGroupIndexCodes
|
||||
)
|
||||
print(result)
|
||||
with open("face_group_1vN_search_service.json", "w", encoding="utf-8") as f:
|
||||
f.write(json.dumps(result[1]))
|
||||
|
||||
# 人脸评分
|
||||
@classmethod
|
||||
async def face_picture_check(cls, image_path):
|
||||
|
||||
with open(image_path, 'rb') as f:
|
||||
image_data = f.read()
|
||||
|
||||
encoded_image = base64.b64encode(image_data).decode('utf-8')
|
||||
|
||||
# print(encoded_image)
|
||||
result = await HaikangUtil.face_picture_check(
|
||||
facePicBinaryData=encoded_image
|
||||
)
|
||||
print(result)
|
||||
with open("face_picture_check.json", "w", encoding="utf-8") as f:
|
||||
f.write(json.dumps(result[1]))
|
||||
|
||||
# 查询访客预约记录
|
||||
@classmethod
|
||||
async def query_visitor_record(cls):
|
||||
result = await HaikangUtil.query_visitor_record(pageNo=1, pageSize=10, visitorStatus=1)
|
||||
print(result)
|
||||
with open("query_visitor_record.json", "w", encoding="utf-8") as f:
|
||||
f.write(json.dumps(result[1]))
|
||||
|
||||
|
||||
# 查询访客预约记录图片
|
||||
@classmethod
|
||||
async def query_visitor_record_pictures_service(cls, svrIndexCode, picUri):
|
||||
await HaikangUtil.query_visitor_record_pictures(svrIndexCode, picUri)
|
||||
|
||||
|
||||
# 获取所有访客图片
|
||||
@classmethod
|
||||
async def get_all_visitor_pictures(cls, pageNo=1, pageSize=10, visitorStatus=1):
|
||||
|
||||
|
||||
result = await HaikangUtil.query_visitor_record(pageNo=pageNo, pageSize=pageSize, visitorStatus=visitorStatus)
|
||||
|
||||
visitor_list = list()
|
||||
if result[0]:
|
||||
total = result[1]['total']
|
||||
visitor_list += result[1]['list']
|
||||
while pageNo*pageSize < total:
|
||||
pageNo += 1
|
||||
result = await HaikangUtil.query_visitor_record(pageNo=pageNo+1, pageSize=pageSize, visitorStatus=visitorStatus)
|
||||
if result[0]:
|
||||
visitor_list += result[1]['list']
|
||||
else:
|
||||
break
|
||||
#
|
||||
# print(visitor_list)
|
||||
|
||||
with open("./visitor/visitorIds.txt", "r", encoding="utf-8") as f:
|
||||
visitor_ids = [line.rstrip('\n') for line in f]
|
||||
# print(visitor_ids)
|
||||
visitor_ids_add = list()
|
||||
for visitor in visitor_list:
|
||||
if visitor["visitorId"] not in visitor_ids:
|
||||
visitor_ids_add.append(visitor["visitorId"])
|
||||
# 获取图片
|
||||
try:
|
||||
await HaikangUtil.query_visitor_record_pictures(visitor["svrIndexCode"], visitor["picUri"], f"./visitor/face_images/{visitor['visitorName']}")
|
||||
await asyncio.sleep(1)
|
||||
except Exception as e:
|
||||
print(e, visitor["visitorId"])
|
||||
|
||||
with open("./visitor/visitorIds.txt", "a", encoding="utf-8") as f:
|
||||
for visitor_id in visitor_ids_add:
|
||||
f.write(visitor_id + "\n")
|
||||
|
||||
|
||||
|
||||
# 获取所有员工图片
|
||||
@classmethod
|
||||
async def get_all_employee_pictures(cls, pageNo=1, pageSize=500):
|
||||
|
||||
result = await HaikangUtil.get_person_list(pageNo=pageNo, pageSize=pageSize)
|
||||
|
||||
employee_list = list()
|
||||
if result[0]:
|
||||
total = result[1]['total']
|
||||
employee_list += result[1]['list']
|
||||
while pageNo * pageSize < total:
|
||||
pageNo += 1
|
||||
result = await HaikangUtil.get_person_list(pageNo=pageNo + 1, pageSize=pageSize, )
|
||||
if result[0]:
|
||||
employee_list += result[1]['list']
|
||||
else:
|
||||
break
|
||||
|
||||
|
||||
with open("./employee/personIds.txt", "r", encoding="utf-8") as f:
|
||||
person_ids = [line.rstrip('\n') for line in f]
|
||||
# print(visitor_ids)
|
||||
person_ids_add = list()
|
||||
for employee in employee_list:
|
||||
# print(employee)
|
||||
if employee["personId"] not in person_ids:
|
||||
person_ids_add.append(employee["personId"])
|
||||
# 获取图片
|
||||
try:
|
||||
await HaikangUtil.get_person_picture(employee.get("personPhoto")[0]["serverIndexCode"], employee.get("personPhoto")[0]["picUri"], f"./employee/face_images/{employee['personName']}")
|
||||
await asyncio.sleep(0.5)
|
||||
except Exception as e:
|
||||
print(f"error: {e}, personId: {employee['personId']}")
|
||||
|
||||
|
||||
with open("./employee/personIds.txt", "a", encoding="utf-8") as f:
|
||||
for person_id in person_ids_add:
|
||||
f.write(person_id + "\n")
|
||||
|
||||
|
||||
# 添加单个人脸分组
|
||||
@classmethod
|
||||
async def face_group_addition_service(cls, name, description):
|
||||
result = await HaikangUtil.face_group_addition(name, description)
|
||||
print(result)
|
||||
with open("face_group_addition_service.json", "a", encoding="utf-8") as f:
|
||||
f.write(json.dumps(result[1]))
|
||||
|
||||
# 向分组中添加单个人脸
|
||||
@classmethod
|
||||
async def face_single_addition_service(cls, faceGroupIndexCode, image_path, name):
|
||||
|
||||
with open(image_path, 'rb') as f:
|
||||
image_data = f.read()
|
||||
|
||||
encoded_image = base64.b64encode(image_data).decode('utf-8')
|
||||
|
||||
faceInfo = {
|
||||
"name": name,
|
||||
}
|
||||
facePic = {
|
||||
"facePicBinaryData": encoded_image,
|
||||
"facePicUrl":""
|
||||
}
|
||||
|
||||
|
||||
result = await HaikangUtil.face_single_addition(faceGroupIndexCode, faceInfo, facePic)
|
||||
print(result)
|
||||
with open("face_single_addition_service.json", "a", encoding="utf-8") as f:
|
||||
f.write(json.dumps(result[1]))
|
||||
|
||||
# 批量删除人脸
|
||||
@classmethod
|
||||
async def face_delete_service(cls, faceGroupIndexCode: str, indexCodes: list):
|
||||
result = await HaikangUtil.face_delete(faceGroupIndexCode, indexCodes)
|
||||
print(result)
|
||||
with open("face_delete_service.json", "a", encoding="utf-8") as f:
|
||||
f.write(json.dumps(result[1]))
|
||||
|
||||
|
||||
|
||||
|
||||
# # 查询访客列表
|
||||
@classmethod
|
||||
async def get_visitor_list_service(cls, visitor_query: VisitorReservationQueryModel):
|
||||
"""查询访客列表
|
||||
@ -17,43 +246,43 @@ class HaiKangService:
|
||||
return result
|
||||
|
||||
|
||||
# 查询门禁点列表v2
|
||||
@classmethod
|
||||
async def get_door_list_service(cls, pageNo: int = 1, pageSize: int = 10):
|
||||
"""查询门禁点列表v2
|
||||
"""
|
||||
# # 查询门禁点列表v2
|
||||
# @classmethod
|
||||
# async def get_door_list_service(cls, pageNo: int = 1, pageSize: int = 10):
|
||||
# """查询门禁点列表v2
|
||||
# """
|
||||
|
||||
result = await HaikangUtil.get_door_list_v2(pageNo=pageNo, pageSize=pageSize)
|
||||
# result = await HaikangUtil.get_door_list_v2(pageNo=pageNo, pageSize=pageSize)
|
||||
|
||||
return result
|
||||
# return result
|
||||
|
||||
# 查询门禁状态
|
||||
@classmethod
|
||||
async def get_door_status_service(cls, door_index_codes):
|
||||
"""查询门禁状态
|
||||
# # 查询门禁状态
|
||||
# @classmethod
|
||||
# async def get_door_status_service(cls, door_index_codes):
|
||||
# """查询门禁状态
|
||||
|
||||
Args:
|
||||
door_index_codes (list): 门禁点唯一标识
|
||||
"""
|
||||
# Args:
|
||||
# door_index_codes (list): 门禁点唯一标识
|
||||
# """
|
||||
|
||||
result = await HaikangUtil.get_door_status(door_index_codes)
|
||||
# result = await HaikangUtil.get_door_status(door_index_codes)
|
||||
|
||||
if result[0]:
|
||||
noAuthDoorIndexCodeList = result[1]
|
||||
if door_index_codes in noAuthDoorIndexCodeList:
|
||||
return False, 500 ,'无门禁权限'
|
||||
else:
|
||||
return True, result[1]
|
||||
else:
|
||||
return result
|
||||
# if result[0]:
|
||||
# noAuthDoorIndexCodeList = result[1]
|
||||
# if door_index_codes in noAuthDoorIndexCodeList:
|
||||
# return False, 500 ,'无门禁权限'
|
||||
# else:
|
||||
# return True, result[1]
|
||||
# else:
|
||||
# return result
|
||||
|
||||
@classmethod
|
||||
def parse_dict(cls, d):
|
||||
t_d = dict()
|
||||
for k, v in d.items():
|
||||
if v:
|
||||
t_d[k] = v
|
||||
return t_d
|
||||
# @classmethod
|
||||
# def parse_dict(cls, d):
|
||||
# t_d = dict()
|
||||
# for k, v in d.items():
|
||||
# if v:
|
||||
# t_d[k] = v
|
||||
# return t_d
|
||||
|
||||
# 门禁点反控
|
||||
# @classmethod
|
||||
|
||||
Loading…
Reference in New Issue
Block a user