haikang_api/001测试海康api.py

from util.haikang_util import HaikangUtil
import asyncio
import base64
import json
import os





# 查询门禁点列表
async def get_door_list_service(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]))
    
# 查询门禁状态
async def get_door_status_service(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]))
    
# 门禁控制
async def door_do_control_service(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]))

# 查询门禁点事件
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)
    with open("query_door_events_service.json", "w", encoding="utf-8") as f:
        f.write(json.dumps(result[1]))
    
# 查看门禁点在线状态
async def door_online_status_service(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]))
    
# 按条件查询人脸分组, 很重要
async def get_face_group_service():
    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搜索
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')

    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=1,
        pageSize=10,
        searchNum=99,
        minSimilarity=50,
        faceGroupIndexCodes=['70cd3fcdcdb444a0b92e75ff66b549e6']
    )
    print(result)
    with open("face_group_1vN_search_service.json", "w", encoding="utf-8") as f:
        f.write(json.dumps(result[1]))
    
# 人脸评分
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')
    
    # 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]))
    
# 查询访客预约记录
async def query_visitor_record():
    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]))


# 查询访客预约记录图片
async def query_visitor_record_pictures_service(svrIndexCode, picUri):
    await HaikangUtil.query_visitor_record_pictures(svrIndexCode, picUri)


async def get_all_visitor_pictures(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']}.jpg")
                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")



async def get_all_employee_pictures(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:
        if employee["visitorId"] not in person_ids:
            person_ids_add.append(employee["visitorId"])
            # 获取图片
            try:
                await HaikangUtil.get_person_picture(employee["serverIndexCode"], employee["picUri"], f"./employee/face_images/{employee['personName']}.jpg")
                await asyncio.sleep(1)
            except Exception as e:
                print(e, employee["visitorId"])

    with open("./employee/personIds.txt", "a", encoding="utf-8") as f:
        for person_id in person_ids_add:
            f.write(person_id + "\n")


# 添加单个人脸分组
async def face_group_addition_service(name, description):
    result = await HaikangUtil.face_group_addition(name, description)
    print(result)
    with open("face_group_addition_service.json", "w", encoding="utf-8") as f:
        f.write(json.dumps(result[1]))



if __name__ == '__main__':
    #asyncio.run(get_door_list_service())
    # print("*"*100)
    # asyncio.run(get_door_status_service(['d4a610b6554d4544bc35dd3138f128b0']))
    # print("*"*100)
    
    # asyncio.run(door_do_control_service(['D01'], 1))
    
    # asyncio.run(query_door_events_service('d4a610b6554d4544bc35dd3138f128b0',1,10,'20025-09-10T17:30:08+08:00','2004-09-15T17:30:08+08:00'))
    
    #asyncio.run(get_face_group_service())
    
    # image_path = "./测试图片1.jpg"
    # asyncio.run(face_group_1vN_search_service(image_path))
    #asyncio.run(face_picture_check(image_path))
    
    #asyncio.run(query_visitor_record())
    # asyncio.run(query_visitor_record_pictures_service("80890167-b8d4-4e33-af35-94db87c90816"
	# , "/pic?0de900=71f0ip-4eo881-342*fo1=9=8522*4l5373169147*1t1=0*0ps==418b*=6127*38da35a12-8338cd-9*l168od0f1f1=002"))
    #asyncio.run(door_online_status_service(["d4a610b6554d4544bc35dd3138f128b0"]))
    asyncio.run(get_all_visitor_pictures())