完成人脸计算任务调用方法

This commit is contained in:
zqc
2025-12-20 20:08:52 +08:00
parent 66890ff094
commit 3a4bcab991
6 changed files with 481 additions and 3 deletions

View File

@@ -0,0 +1,315 @@
"""
人脸特征计算算法路由
提供人脸特征计算的HTTP接口
"""
import os
import logging
from datetime import datetime, timedelta
from typing import Dict, Any
from fastapi import APIRouter, HTTPException, BackgroundTasks
from sqlalchemy.orm import Session
from src.config import settings
from src.database.connection import db_manager
from src.models.face_feature import SurFaceFeature, FeatureStatus
from src.repositories.face_feature_repository import FaceFeatureRepository
from src.face_recognition_algorithm import FaceRecognitionAlgorithm
# 创建路由器
router = APIRouter(prefix="/algorithm", tags=["algorithm"])
# 初始化人脸识别算法
face_algorithm = FaceRecognitionAlgorithm(use_gpu=settings.FACE_USE_GPU, use_npu=settings.FACE_USE_NPU)
logger = logging.getLogger(__name__)
def process_feature_calculation(feature_id: int) -> bool:
"""
处理单个人脸特征计算
参数:
feature_id: 特征记录ID
返回:
是否成功处理
"""
try:
with db_manager.get_session() as session:
repository = FaceFeatureRepository(session)
# 获取特征记录
feature = repository.get_by_id(feature_id)
if not feature:
logger.error(f"特征记录不存在: {feature_id}")
return False
# 检查是否已经处理完成
if feature.status in [FeatureStatus.SUCCESS, FeatureStatus.FAILED]:
logger.info(f"特征记录已处理完成: {feature_id}, 状态: {feature.status_name}")
return True
# 检查是否超时
if feature.status == FeatureStatus.PROCESSING:
if feature.start_time:
timeout_duration = timedelta(hours=settings.FACE_CAL_FEATURE_TIMEOUT_HOURS)
if datetime.now() - feature.start_time > timeout_duration:
# 超时处理
feature.status = FeatureStatus.FAILED
feature.finish_time = datetime.now()
session.commit()
logger.warning(f"特征计算超时: {feature_id}")
return False
else:
# 没有开始时间,重置状态
feature.status = FeatureStatus.NOT_STARTED
session.commit()
# 处理未开始的计算
if feature.status == FeatureStatus.NOT_STARTED:
# 设置状态为计算中
feature.status = FeatureStatus.PROCESSING
feature.start_time = datetime.now()
session.commit()
logger.info(f"开始计算特征: {feature_id}")
# 构建图片路径
if not feature.pic_id:
logger.error(f"特征记录缺少图片ID: {feature_id}")
feature.status = FeatureStatus.FAILED
feature.finish_time = datetime.now()
session.commit()
return False
image_path = os.path.join(settings.FACE_REGISTER_IMAGE_RESOURCE_DIR, feature.pic_id)
# 检查图片文件是否存在
if not os.path.exists(image_path):
logger.error(f"图片文件不存在: {image_path}")
feature.status = FeatureStatus.FAILED
feature.finish_time = datetime.now()
session.commit()
return False
# 提取人脸特征
try:
feature_vector = face_algorithm.extract_face_feature(str(image_path))
if feature_vector is not None:
# 转换为二进制数据
feature_bytes = feature_vector.tobytes()
feature.feature_data = feature_bytes
feature.status = FeatureStatus.SUCCESS
feature.finish_time = datetime.now()
session.commit()
logger.info(f"特征计算成功: {feature_id}, 特征向量长度: {len(feature_vector)}")
return True
else:
logger.error(f"特征提取失败: {feature_id}")
feature.status = FeatureStatus.FAILED
feature.finish_time = datetime.now()
session.commit()
return False
except Exception as e:
logger.error(f"特征计算过程中出错: {feature_id}, 错误: {str(e)}")
feature.status = FeatureStatus.FAILED
feature.finish_time = datetime.now()
session.commit()
return False
return True
except Exception as e:
logger.error(f"处理特征计算时发生异常: {feature_id}, 错误: {str(e)}")
return False
def process_pending_features() -> Dict[str, Any]:
"""
处理所有待处理的人脸特征计算
返回:
处理结果统计
"""
try:
with db_manager.get_session() as session:
repository = FaceFeatureRepository(session)
# 查找需要处理的记录
# 条件: feature_type = FACE_MODEL_VERSION 且 status = 0 (未开始)
pending_features = repository.get_features_by_type_and_status(
feature_type=settings.FACE_MODEL_VERSION,
status=FeatureStatus.NOT_STARTED
)
# 查找可能超时的记录 (status = 1 且超时)
timeout_features = []
processing_features = repository.get_features_by_type_and_status(
feature_type=settings.FACE_MODEL_VERSION,
status=FeatureStatus.PROCESSING
)
for feature in processing_features:
if feature.start_time:
timeout_duration = timedelta(hours=settings.FACE_CAL_FEATURE_TIMEOUT_HOURS)
if datetime.now() - feature.start_time > timeout_duration:
timeout_features.append(feature)
total_pending = len(pending_features)
total_timeout = len(timeout_features)
logger.info(f"发现待处理特征: {total_pending}个, 超时特征: {total_timeout}")
# 处理超时记录
timeout_success = 0
for feature in timeout_features:
feature.status = FeatureStatus.FAILED
feature.finish_time = datetime.now()
timeout_success += 1
if timeout_features:
session.commit()
# 处理待处理记录
processed_count = 0
success_count = 0
for feature in pending_features:
processed_count += 1
if process_feature_calculation(feature.id):
success_count += 1
# 每处理10个记录输出一次进度
if processed_count % 10 == 0:
logger.info(f"处理进度: {processed_count}/{total_pending}")
return {
"total_pending": total_pending,
"total_timeout": total_timeout,
"processed_count": processed_count,
"success_count": success_count,
"timeout_handled": timeout_success
}
except Exception as e:
logger.error(f"批量处理特征计算时发生异常: {str(e)}")
return {
"total_pending": 0,
"total_timeout": 0,
"processed_count": 0,
"success_count": 0,
"timeout_handled": 0,
"error": str(e)
}
@router.post("/start-feature-calculation", summary="开始人脸特征计算")
async def start_feature_calculation(background_tasks: BackgroundTasks):
"""
开始处理人脸特征计算
此接口会:
1. 查找所有feature_type为当前模型版本且status为0的记录
2. 将状态改为1设置开始时间
3. 提取人脸特征值
4. 对于status为1且超时的记录标记为失败
返回处理结果统计
"""
try:
# 在后台任务中处理,避免阻塞请求
result = process_pending_features()
return {
"success": True,
"message": "特征计算处理完成",
"data": result
}
except Exception as e:
logger.error(f"启动特征计算失败: {str(e)}")
raise HTTPException(status_code=500, detail=f"启动特征计算失败: {str(e)}")
@router.get("/feature-calculation-status", summary="获取特征计算状态")
async def get_feature_calculation_status():
"""
获取当前特征计算的状态统计
"""
try:
with db_manager.get_session() as session:
repository = FaceFeatureRepository(session)
# 获取统计信息
stats = repository.get_statistics()
# 获取当前模型版本的特定统计
current_model_stats = {
"total": repository.count_by_type_and_status(
feature_type=settings.FACE_MODEL_VERSION
),
"not_started": repository.count_by_type_and_status(
feature_type=settings.FACE_MODEL_VERSION,
status=FeatureStatus.NOT_STARTED
),
"processing": repository.count_by_type_and_status(
feature_type=settings.FACE_MODEL_VERSION,
status=FeatureStatus.PROCESSING
),
"success": repository.count_by_type_and_status(
feature_type=settings.FACE_MODEL_VERSION,
status=FeatureStatus.SUCCESS
),
"failed": repository.count_by_type_and_status(
feature_type=settings.FACE_MODEL_VERSION,
status=FeatureStatus.FAILED
)
}
return {
"success": True,
"data": {
"overall_stats": stats,
"current_model_stats": current_model_stats,
"model_version": settings.FACE_MODEL_VERSION,
"timeout_hours": settings.FACE_CAL_FEATURE_TIMEOUT_HOURS
}
}
except Exception as e:
logger.error(f"获取特征计算状态失败: {str(e)}")
raise HTTPException(status_code=500, detail=f"获取特征计算状态失败: {str(e)}")
@router.post("/calculate-single-feature/{feature_id}", summary="计算单个特征")
async def calculate_single_feature(feature_id: int):
"""
计算单个特征记录的人脸特征
参数:
feature_id: 特征记录ID
"""
try:
success = process_feature_calculation(feature_id)
if success:
return {
"success": True,
"message": f"特征计算完成: {feature_id}"
}
else:
return {
"success": False,
"message": f"特征计算失败: {feature_id}"
}
except Exception as e:
logger.error(f"计算单个特征失败: {feature_id}, 错误: {str(e)}")
raise HTTPException(status_code=500, detail=f"计算单个特征失败: {str(e)}")
# 导出路由器
__all__ = ["router"]

View File

@@ -18,6 +18,7 @@ from fastapi.openapi.docs import (
from fastapi.staticfiles import StaticFiles
from src.api.routes import face_features
from src.api.routes.algorithm_router import router as algorithm_router
from src.api.errors import (
APIError,
validation_exception_handler,
@@ -166,6 +167,11 @@ app.include_router(
prefix=settings.API_V1_PREFIX
)
app.include_router(
algorithm_router,
prefix=settings.API_V1_PREFIX
)
# 自定义404处理器
@app.exception_handler(404)

View File

@@ -47,6 +47,14 @@ class Settings(BaseSettings):
# 异步配置
ASYNC_MODE: bool = False
# 资源文件夹配置
FACE_REGISTER_IMAGE_RESOURCE_DIR: str = "D:/ruoyi/uploadPath/face"
VIDEO_RESOURCE_DIR: str = "D:/ruoyi/uploadPath/video"
FACE_CAL_FEATURE_TIMEOUT_HOURS: int = 10
FACE_MODEL_VERSION: int = 0 #insight_face_buffalo_l
FACE_USE_GPU: bool = True
FACE_USE_NPU: bool = False
# JWT配置预留
SECRET_KEY: str = "your-secret-key-here-change-in-production"
ALGORITHM: str = "HS256"

View File

@@ -488,3 +488,73 @@ class FaceRecognitionAlgorithm:
def get_list_mode(self) -> str:
"""获取当前名单模式"""
return self.list_mode
def extract_face_feature(self, image_path: str) -> Optional[np.ndarray]:
"""
从单张图片中提取人脸特征值
参数:
image_path: 人脸图片路径
返回:
numpy数组格式的人脸特征值如果检测失败返回None
"""
if not os.path.exists(image_path):
print(f"❌ 图片文件不存在: {image_path}")
return None
# 读取图片
img = cv2.imread(image_path)
if img is None:
print(f"❌ 无法读取图片: {image_path}")
return None
# 人脸检测
faces = self.app.get(img)
if not faces:
print(f"❌ 图片中未检测到人脸: {image_path}")
return None
# 使用第一张检测到的人脸
face = faces[0]
# 检查人脸质量
is_acceptable, quality_metrics = self.is_face_quality_acceptable(face, img)
if not is_acceptable:
print(f"⚠️ 人脸质量不可接受: {image_path}")
print(f" 质量得分: {quality_metrics['quality_score']:.3f}")
print(f" 清晰度: {quality_metrics['clarity_score']:.1f}")
print(f" 姿态角度: pitch={quality_metrics['pitch']:.1f}°, yaw={quality_metrics['yaw']:.1f}°")
return None
print(f"✅ 成功提取人脸特征: {image_path}")
print(f" 检测得分: {quality_metrics['det_score']:.3f}")
print(f" 质量得分: {quality_metrics['quality_score']:.3f}")
# 返回特征向量
return face.embedding
def extract_face_features_batch(self, image_paths: List[str]) -> Dict[str, Optional[np.ndarray]]:
"""
批量从多张图片中提取人脸特征值
参数:
image_paths: 人脸图片路径列表
返回:
字典格式 {图片路径: 特征值}失败的特征值为None
"""
results = {}
for image_path in image_paths:
feature = self.extract_face_feature(image_path)
results[image_path] = feature
# 统计结果
success_count = sum(1 for feature in results.values() if feature is not None)
total_count = len(image_paths)
print(f"🎉 批量提取完成: 成功 {success_count}/{total_count} 张图片")
return results

View File

@@ -37,7 +37,7 @@ def demo_sync_operations():
print("\n1. 创建特征记录")
# 固定的测试ID
test_person_id = 1001
test_person_id = 10
test_feature_type = 1
# 检查是否已存在

View File

@@ -595,3 +595,82 @@ class FaceFeatureRepository:
logger.error(f"Database error cleaning up old features: {e}")
self.session.rollback()
raise
# ===== 新增方法:支持算法路由 =====
def get_features_by_type_and_status(
self,
feature_type: int,
status: Optional[int] = None
) -> List[SurFaceFeature]:
"""
根据特征类型和状态获取特征记录
Args:
feature_type: 特征类型
status: 状态(可选)
Returns:
特征记录列表
"""
try:
conditions = [SurFaceFeature.feature_type == feature_type]
if status is not None:
conditions.append(SurFaceFeature.status == status)
stmt = (
select(SurFaceFeature)
.where(and_(*conditions))
.order_by(asc(SurFaceFeature.created_time))
)
result = self.session.execute(stmt)
features = list(result.scalars().all())
logger.debug(f"Retrieved {len(features)} features by type={feature_type}, status={status}")
return features
except SQLAlchemyError as e:
logger.error(f"Database error getting features by type and status: {e}")
raise
def count_by_type_and_status(
self,
feature_type: int,
status: Optional[int] = None
) -> int:
"""
统计指定特征类型和状态的记录数量
Args:
feature_type: 特征类型
status: 状态(可选)
Returns:
记录数量
"""
try:
conditions = [SurFaceFeature.feature_type == feature_type]
if status is not None:
conditions.append(SurFaceFeature.status == status)
stmt = select(func.count()).select_from(SurFaceFeature).where(and_(*conditions))
result = self.session.execute(stmt)
count = result.scalar_one()
return count
except SQLAlchemyError as e:
logger.error(f"Database error counting features by type and status: {e}")
raise
def get_statistics(self) -> Dict[str, Any]:
"""
获取统计信息(兼容性方法)
Returns:
统计信息字典
"""
return self.get_stats()