109 lines
4.1 KiB
Python
109 lines
4.1 KiB
Python
from config.env import ComprefaceConfig
|
|
from compreface import CompreFace
|
|
from compreface.service import RecognitionService
|
|
from compreface.collections import FaceCollection
|
|
from compreface.common.typed_dict import AllOptionsDict, check_fields_by_name
|
|
|
|
|
|
import httpx
|
|
import json
|
|
import os
|
|
|
|
|
|
|
|
RECOGNITION_ROOT_API: str = '/api/v1/recognition'
|
|
|
|
RECOGNIZE_API: str = RECOGNITION_ROOT_API + '/recognize'
|
|
RECOGNIZE_CRUD_API: str = RECOGNITION_ROOT_API + '/faces'
|
|
SUBJECTS_CRUD_API: str = RECOGNITION_ROOT_API + '/subjects'
|
|
|
|
DETECTION_API: str = '/api/v1/detection/detect'
|
|
|
|
VERIFICATION_API: str = '/api/v1/verification'
|
|
|
|
|
|
|
|
|
|
compre_face: CompreFace = CompreFace(ComprefaceConfig.COMPREFACE_BASE_URL, ComprefaceConfig.COMPERFACE_BASE_PORT)
|
|
# recognition: RecognitionService = compre_face.init_face_recognition(ComprefaceConfig.COMPREFACE_API_KEY)
|
|
# face_collection: FaceCollection = recognition.get_face_collection()
|
|
|
|
|
|
|
|
class ComprefaceUtil:
|
|
|
|
api_key: str = ComprefaceConfig.COMPREFACE_API_KEY
|
|
@classmethod
|
|
async def send_request(cls, method: str, url: str, headers: dict, body: str | None=None, data=None, files=None ,json=True):
|
|
|
|
async with httpx.AsyncClient(verify=False) as client:
|
|
response = await client.request(
|
|
method, url, headers=headers, content=body, data=data, files=files, timeout=5 # timeout 单位秒
|
|
)
|
|
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 json:
|
|
return response.json()
|
|
return response
|
|
|
|
# 添加人脸到指定分组下, subject 为分组名称
|
|
@classmethod
|
|
async def face_addition(cls, image: str = '' or bytes, subject: str = '') -> dict:
|
|
client_url: str = RECOGNIZE_CRUD_API
|
|
url: str = ComprefaceConfig.COMPREFACE_BASE_URL + ':' + ComprefaceConfig.COMPERFACE_BASE_PORT + client_url + '?subject=' + subject
|
|
# Validation loop and adding fields to the url.
|
|
m = await multipart_constructor(image)
|
|
result = await cls.send_request("post"
|
|
, url
|
|
, headers={'x-api-key': cls.api_key}
|
|
, files=m)
|
|
return result
|
|
# 人脸识别
|
|
@classmethod
|
|
async def face_recognition(cls, image: str = '' or bytes, options: AllOptionsDict = {}) -> dict:
|
|
|
|
client_url: str = RECOGNIZE_API
|
|
url: str = ComprefaceConfig.COMPREFACE_BASE_URL + ':' + ComprefaceConfig.COMPERFACE_BASE_PORT + client_url+"?"
|
|
|
|
for key in options.keys():
|
|
# Checks fields with necessary rules.
|
|
# key - key field by options.
|
|
check_fields_by_name(key, options[key])
|
|
url += '&' + key + "=" + str(options[key])
|
|
|
|
m = await multipart_constructor(image)
|
|
result = await cls.send_request("post"
|
|
, url
|
|
, headers={'x-api-key': cls.api_key}
|
|
, files=m)
|
|
return result
|
|
|
|
|
|
async def get_file(image: str = '' or bytes):
|
|
if not os.path.isfile(image):
|
|
if type(image) != bytes:
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.get(image, timeout=5)
|
|
file = response.content
|
|
else:
|
|
file = image
|
|
file = ('image.jpg', file)
|
|
else:
|
|
name_img: str = os.path.basename(image)
|
|
file = (name_img, open(image, 'rb'))
|
|
return file
|
|
|
|
|
|
async def multipart_constructor(image: str = '' or bytes):
|
|
|
|
return {"file": await get_file(image)}
|
|
|
|
|
|
# def multipart_constructor_with_two_images(source_image: str = '' or bytes, target_image: str = '' or bytes):
|
|
# return MultipartEncoder(
|
|
# fields={'source_image': get_file(
|
|
# source_image), 'target_image': get_file(target_image)}
|
|
# ) |