haotian/kangda_robot_backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修改获取机器人角色接口, 现讲解风格和机器人角色相绑定, 返回机器人角色和当前角色的风格

Browse Source
This commit is contained in:
haotian 2025-08-25 16:21:39 +08:00
parent 8be0d4349c
commit 85641967f0
13 changed files with 525 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
ruoyi-fastapi-backend/.env.dev
View File

@ -55,7 +55,7 @@ DB_POOL_TIMEOUT = 30
# -------- Redis配置 --------
# Redis主机
REDIS_HOST = '127.0.0.1'
REDIS_HOST = '10.0.0.58'
# Redis端口
REDIS_PORT = 6379
# Redis用户名
@ -63,4 +63,31 @@ REDIS_USERNAME = ''
# Redis密码
REDIS_PASSWORD = ''
# Redis数据库
REDIS_DATABASE = 2
REDIS_DATABASE = 3
#------------------海康平台配置------------------
HAIKANG_URL = ''
HAIKANG_PORT = 443
HAIKANG_AK = ''
HAIKANG_SK = ''
# 获取access_token
HAIKANG_ACCESS_TOKEN_URL = '/api/v1/oauth/token'
# 查看门禁状态
HAIKANG_DOOR_STATES_URL = '/api/v1/door/states'
# 门禁点反控
HAIKANG_DOOR_DOCONTROL_URL = '/api/acs/v1/door/doControl'
# 门禁点事件v2
HAIKANG_DOOR_ENVENTS_URL = '/api/acs/v2/door/events'
# 人脸分组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'

27
ruoyi-fastapi-backend/config/env.py
View File

@ -143,6 +143,24 @@ class CachePathConfig:
PATH = os.path.join(os.path.abspath(os.getcwd()), 'caches')
PATHSTR = 'caches'
class HaiKangSettings:
"""
海康平台配置
"""
HAIKANG_URL = ''
HAIKANG_PORT = 443
HAIKANG_AK = ''
HAIKANG_SK = ''
HAIKANG_ACCESS_TOKEN_URL = '/api/v1/oauth/token'
HAIKANG_DOOR_STATES_URL = '/api/v1/door/states'
HAIKANG_DOOR_DOCONTROL_URL = '/api/acs/v1/door/doControl'
HAIKANG_DOOR_ENVENTS_URL = '/api/acs/v2/door/events'
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'
class GetConfig:
@ -200,6 +218,13 @@ class GetConfig:
"""
# 实例上传配置
return UploadSettings()
@lru_cache()
def get_haikang_config(self):
"""
获取海康平台配置
"""
return HaiKangSettings()
@staticmethod
def parse_cli_args():
@ -242,3 +267,5 @@ RedisConfig = get_config.get_redis_config()
GenConfig = get_config.get_gen_config()
# 上传配置
UploadConfig = get_config.get_upload_config()
# 海康平台配置
HaiKangConfig = get_config.get_haikang_config()

59
ruoyi-fastapi-backend/module_admin/controller/identify_record_controller.py
View File

@ -35,20 +35,20 @@ async def get_system_identify_record_list(
return ResponseUtil.success(model_content=identify_record_page_query_result)
@identify_recordController.post('', dependencies=[Depends(CheckUserInterfaceAuth('system:identify_record:add'))])
@ValidateFields(validate_model='add_identify_record')
@Log(title='识别记录', business_type=BusinessType.INSERT)
async def add_system_identify_record(
request: Request,
add_identify_record: Identify_recordModel,
query_db: AsyncSession = Depends(get_db),
current_user: CurrentUserModel = Depends(LoginService.get_current_user),
):
add_identify_record.create_time = datetime.now()
add_identify_record_result = await Identify_recordService.add_identify_record_services(query_db, add_identify_record)
logger.info(add_identify_record_result.message)
return ResponseUtil.success(msg=add_identify_record_result.message)
# @identify_recordController.post('', dependencies=[Depends(CheckUserInterfaceAuth('system:identify_record:add'))])
# @ValidateFields(validate_model='add_identify_record')
# @Log(title='识别记录', business_type=BusinessType.INSERT)
# async def add_system_identify_record(
# request: Request,
# add_identify_record: Identify_recordModel,
# query_db: AsyncSession = Depends(get_db),
# current_user: CurrentUserModel = Depends(LoginService.get_current_user),
# ):
# add_identify_record.create_time = datetime.now()
# add_identify_record_result = await Identify_recordService.add_identify_record_services(query_db, add_identify_record)
# logger.info(add_identify_record_result.message)
#
# return ResponseUtil.success(msg=add_identify_record_result.message)
# @identify_recordController.put('', dependencies=[Depends(CheckUserInterfaceAuth('system:identify_record:edit'))])
@ -88,16 +88,21 @@ async def add_system_identify_record(
# return ResponseUtil.success(data=identify_record_detail_result)
#
#
# @identify_recordController.post('/export', dependencies=[Depends(CheckUserInterfaceAuth('system:identify_record:export'))])
# @Log(title='识别记录', business_type=BusinessType.EXPORT)
# async def export_system_identify_record_list(
# request: Request,
# identify_record_page_query: Identify_recordPageQueryModel = Form(),
# query_db: AsyncSession = Depends(get_db),
# ):
# # 获取全量数据
# identify_record_query_result = await Identify_recordService.get_identify_record_list_services(query_db, identify_record_page_query, is_page=False)
# identify_record_export_result = await Identify_recordService.export_identify_record_list_services(identify_record_query_result)
# logger.info('导出成功')
#
# return ResponseUtil.streaming(data=bytes2file_response(identify_record_export_result))
@identify_recordController.post('/export'
# , dependencies=[Depends(CheckUserInterfaceAuth('system:identify_record:export'))]
)
@Log(title='识别记录', business_type=BusinessType.EXPORT)
async def export_system_identify_record_list(
request: Request,
identify_record_page_query: Identify_recordPageQueryModel = Form(),
query_db: AsyncSession = Depends(get_db),
):
# 获取全量数据
# identify_record_query_result = await Identify_recordService.get_identify_record_list_services(query_db, identify_record_page_query, is_page=False)
identify_record_query_result = await Identify_recordService.test_export_services(query_db)
identify_record_export_result = await Identify_recordService.export_identify_record_list_services(identify_record_query_result)
logger.info('导出成功')
return ResponseUtil.streaming(data=bytes2file_response(identify_record_export_result))

5
ruoyi-fastapi-backend/module_admin/controller/robot_role_pairing_controller.py
View File

@ -21,6 +21,7 @@ from datetime import datetime
pairingController = APIRouter(prefix='/system/robot_role', dependencies=[Depends(LoginService.get_current_user)])
# 获取机器人角色
@pairingController.get(
'/{robot_id}',
)
@ -30,9 +31,9 @@ async def get_system_role_pairing_list(
query_db: AsyncSession = Depends(get_db),
):
# 获取分页数据
pairing_list = await PairingService.get_role_pairing_list_services(query_db, robot_id)
pairing_list, style = await PairingService.get_role_pairing_list_services(query_db, robot_id)
return ResponseUtil.success(data=pairing_list)
return ResponseUtil.success(data={"role": pairing_list, "style":style})
@pairingController.post(
'/change',

16
ruoyi-fastapi-backend/module_admin/dao/explanation_style_robot_pair_dao.py
View File

@ -4,6 +4,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from module_admin.entity.do.explanation_style_do import ExplanationStyle
from module_admin.entity.do.explanation_style_robot_pair_do import ExplanationStyleRobotPairing
from module_admin.entity.do.explanation_style_do import ExplanationStyle
from module_admin.entity.vo.explanation_style_robot_pair_vo import Explanation_style_robot_pairModel, \
Explanation_style_robot_pairPageQueryModel, SwitchExplanationStyleModel
from utils.page_util import PageUtil
@ -13,6 +14,21 @@ class Explanation_style_robot_pairDao:
"""
讲解风格--机器人配对模块数据库操作层
"""
@classmethod
async def get_robot_role_style(cls, db: AsyncSession, robot_id: int, role_id: int):
stmt = (
select(ExplanationStyleRobotPairing.pairing_id, ExplanationStyleRobotPairing.explanation_style_id, ExplanationStyle.name, ExplanationStyle.detail, ExplanationStyleRobotPairing.status)
.select_from(ExplanationStyleRobotPairing)
.join(ExplanationStyle, ExplanationStyle.explanation_style_id == ExplanationStyleRobotPairing.explanation_style_id)
.where(ExplanationStyleRobotPairing.robot_id == robot_id, ExplanationStyleRobotPairing.robot_role_id == role_id)
)
result = (await db.execute(stmt)).mappings().all()
return result
@classmethod
async def switch_explanation_style(cls, db: AsyncSession, robot_id: int = None, pairing_id: int = None, status:str = '0', update_by: str = None):

8
ruoyi-fastapi-backend/module_admin/dao/identify_record_dao.py
View File

@ -11,6 +11,14 @@ class Identify_recordDao:
识别记录模块数据库操作层
"""
@classmethod
async def test_export(cls, db: AsyncSession):
return (
await db.execute(
select(IdentifyRecord)
)
).mappings().all()
@classmethod
async def get_identify_record_detail_by_id(cls, db: AsyncSession, record_id: int):
"""

2
ruoyi-fastapi-backend/module_admin/dao/robot_role_pairing_dao.py
View File

@ -34,7 +34,7 @@ class PairingDao:
async def get_role_pairing_list(cls, db: AsyncSession, robot_id: int):
query = (
select(RobotRolePairing.pairing_id, RobotRole.rolename, RobotRole.detail, RobotRolePairing.status)
select(RobotRolePairing.pairing_id, RobotRole.robot_role_id ,RobotRole.rolename, RobotRole.detail, RobotRolePairing.status)
.select_from(RobotRolePairing)
.join(RobotRole, RobotRole.robot_role_id == RobotRolePairing.robot_role_id)
.where(RobotRolePairing.robot_id == robot_id)

36
ruoyi-fastapi-backend/module_admin/service/haikang_service.py Normal file
View File

@ -0,0 +1,36 @@
from utils.haikang_util import HaikangUtil
class HaiKangService:
"""
康达服务
"""
# 查询门禁状态
@classmethod
async def get_door_status_service(cls, door_index_codes):
"""查询门禁状态
Args:
door_index_codes (list): 门禁点唯一标识
"""
result = await HaikangUtil.get_door_status(door_index_codes)
if result[0]:
noAuthDoorIndexCodeList = result[1]
if door_index_codes in noAuthDoorIndexCodeList:
return False, '无门禁权限'
else:
return True, result[1]
else:
return False, result[1]
# 门禁点反控
# @classmethod
# async def door_do_control_service(cls, door_index_code: list, control_type: int):
# """门禁点反控服务
# Args:
# door_index_code (list): _description_
# control_type (int): _description_
# """

6
ruoyi-fastapi-backend/module_admin/service/identify_record_service.py
View File

@ -14,6 +14,12 @@ class Identify_recordService:
识别记录模块服务层
"""
# 测试方法
@classmethod
async def test_export_services(cls, query_db: AsyncSession):
result = await Identify_recordDao.test_export(query_db)
return result
@classmethod
async def get_identify_record_list_services(
cls, query_db: AsyncSession, query_object: Identify_recordPageQueryModel, is_page: bool = False

14
ruoyi-fastapi-backend/module_admin/service/robot_role_pairing_service.py
View File

@ -4,6 +4,7 @@ from config.constant import CommonConstant
from exceptions.exception import ServiceException
from module_admin.entity.vo.common_vo import CrudResponseModel
from module_admin.dao.robot_role_pairing_dao import PairingDao
from module_admin.dao.explanation_style_robot_pair_dao import Explanation_style_robot_pairDao
from module_admin.entity.vo.robot_role_pairing_vo import DeletePairingModel, PairingModel, PairingPageQueryModel, \
ChangeRobotRoleModel
from utils.common_util import CamelCaseUtil
@ -33,7 +34,18 @@ class PairingService:
@classmethod
async def get_role_pairing_list_services(cls, query_db: AsyncSession, robot_id: int):
pairing_list = await PairingDao.get_role_pairing_list(query_db, robot_id)
return pairing_list
style = None
for pairing in pairing_list:
if pairing["status"] == '1':
style = await Explanation_style_robot_pairDao.get_robot_role_style(query_db, robot_id, pairing["robot_role_id"])
return pairing_list, style
@classmethod
async def get_pairing_list_services(

16
ruoyi-fastapi-backend/utils/excel_util.py
View File

@ -34,8 +34,22 @@ class ExcelUtil:
:param mapping_dict: 映射字典
:return: list数据对应excel的二进制数据
"""
mapping_data = cls.__mapping_list(list_data, mapping_dict)
print("-" * 100, '\n', list_data)
new_t = list()
# list_data = dict(list_data)
for item in list_data:
t = dict(item)
new_t.append(t)
print(t)
# for k, v in item.items():
# print(k, v)
# print(item.get('recordId'))
# print(item.get())
# mapping_data = cls.__mapping_list(list_data, mapping_dict)
mapping_data = cls.__mapping_list(new_t, mapping_dict)
df = pd.DataFrame(mapping_data)
print("-"*100,'\n', df)
binary_data = io.BytesIO()
df.to_excel(binary_data, index=False, engine='openpyxl')
binary_data = binary_data.getvalue()

336
ruoyi-fastapi-backend/utils/haikang_util.py Normal file
View File

@ -0,0 +1,336 @@
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]
# 查询门禁状态
@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": min(args.get("pageNo", 1), 1),
"pageSize": max(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"]]

3
ruoyi-fastapi-backend/utils/response_util.py
View File

@ -264,3 +264,6 @@ class ResponseUtil:
return StreamingResponse(
status_code=status.HTTP_200_OK, content=data, headers=headers, media_type=media_type, background=background
)