562 lines
16 KiB
Python
562 lines
16 KiB
Python
"""
|
|
人脸特征业务逻辑服务层
|
|
"""
|
|
|
|
import logging
|
|
from datetime import datetime
|
|
from typing import Optional, List, Dict, Any
|
|
from contextlib import contextmanager
|
|
|
|
from repositories.face_feature_repository import FaceFeatureRepository
|
|
from schemas.face_feature import (
|
|
FaceFeatureCreate,
|
|
FaceFeatureUpdate,
|
|
FaceFeatureQuery,
|
|
FaceFeatureResponse,
|
|
FaceFeatureListResponse,
|
|
FaceFeatureStatsResponse,
|
|
BatchFaceFeatureCreate,
|
|
FeatureStatus
|
|
)
|
|
from models.face_feature import FeatureStatusEnum
|
|
from utils.logger import setup_logger
|
|
|
|
logger = setup_logger(__name__)
|
|
|
|
|
|
class FaceFeatureService:
|
|
"""人脸特征业务服务"""
|
|
|
|
def __init__(self, repository: FaceFeatureRepository):
|
|
"""
|
|
初始化服务
|
|
|
|
Args:
|
|
repository: 特征仓库实例
|
|
"""
|
|
self.repository = repository
|
|
|
|
# ===== CRUD操作 =====
|
|
|
|
def create_feature(self, feature_data: FaceFeatureCreate) -> FaceFeatureResponse:
|
|
"""
|
|
创建特征记录
|
|
|
|
Args:
|
|
feature_data: 特征数据
|
|
|
|
Returns:
|
|
创建的特征记录响应
|
|
"""
|
|
logger.info(f"Creating face feature for person_id={feature_data.person_id}")
|
|
|
|
# 业务逻辑验证
|
|
self._validate_feature_data(feature_data)
|
|
|
|
# 检查是否已存在相同记录
|
|
if feature_data.feature_type is not None:
|
|
existing = self.repository.get_by_person_and_type(
|
|
feature_data.person_id,
|
|
feature_data.feature_type
|
|
)
|
|
if existing:
|
|
raise ValueError(
|
|
f"Feature record already exists for person_id={feature_data.person_id}, "
|
|
f"feature_type={feature_data.feature_type}"
|
|
)
|
|
|
|
# 创建记录
|
|
feature = self.repository.create(feature_data)
|
|
|
|
# 转换为响应模型
|
|
return FaceFeatureResponse.model_validate(feature)
|
|
|
|
def create_features_batch(self, batch_data: BatchFaceFeatureCreate) -> List[FaceFeatureResponse]:
|
|
"""
|
|
批量创建特征记录
|
|
|
|
Args:
|
|
batch_data: 批量特征数据
|
|
|
|
Returns:
|
|
创建的特征记录响应列表
|
|
"""
|
|
logger.info(f"Creating {len(batch_data.items)} face features in batch")
|
|
|
|
# 验证所有数据
|
|
for feature_data in batch_data.items:
|
|
self._validate_feature_data(feature_data)
|
|
|
|
# 批量创建
|
|
features = self.repository.create_batch(batch_data.items)
|
|
|
|
# 转换为响应模型
|
|
return [FaceFeatureResponse.model_validate(f) for f in features]
|
|
|
|
def get_feature(self, feature_id: int) -> Optional[FaceFeatureResponse]:
|
|
"""
|
|
获取特征记录
|
|
|
|
Args:
|
|
feature_id: 特征记录ID
|
|
|
|
Returns:
|
|
特征记录响应或None
|
|
"""
|
|
logger.debug(f"Getting face feature: id={feature_id}")
|
|
|
|
feature = self.repository.get_by_id(feature_id)
|
|
if not feature:
|
|
return None
|
|
|
|
return FaceFeatureResponse.model_validate(feature)
|
|
|
|
def get_feature_by_person_and_type(
|
|
self,
|
|
person_id: int,
|
|
feature_type: int
|
|
) -> Optional[FaceFeatureResponse]:
|
|
"""
|
|
根据人员ID和特征类型获取特征记录
|
|
|
|
Args:
|
|
person_id: 人员ID
|
|
feature_type: 特征类型
|
|
|
|
Returns:
|
|
特征记录响应或None
|
|
"""
|
|
logger.debug(f"Getting face feature: person_id={person_id}, feature_type={feature_type}")
|
|
|
|
feature = self.repository.get_by_person_and_type(person_id, feature_type)
|
|
if not feature:
|
|
return None
|
|
|
|
return FaceFeatureResponse.model_validate(feature)
|
|
|
|
def list_features_by_person(
|
|
self,
|
|
person_id: int,
|
|
limit: int = 100
|
|
) -> List[FaceFeatureResponse]:
|
|
"""
|
|
获取人员的特征记录列表
|
|
|
|
Args:
|
|
person_id: 人员ID
|
|
limit: 返回数量限制
|
|
|
|
Returns:
|
|
特征记录响应列表
|
|
"""
|
|
logger.debug(f"Listing face features for person_id={person_id}")
|
|
|
|
features = self.repository.get_by_person(person_id, limit)
|
|
return [FaceFeatureResponse.model_validate(f) for f in features]
|
|
|
|
def query_features(
|
|
self,
|
|
query: FaceFeatureQuery,
|
|
page: int = 1,
|
|
page_size: int = 20,
|
|
order_by: str = "created_time",
|
|
desc_order: bool = True
|
|
) -> FaceFeatureListResponse:
|
|
"""
|
|
查询特征记录
|
|
|
|
Args:
|
|
query: 查询条件
|
|
page: 页码
|
|
page_size: 每页数量
|
|
order_by: 排序字段
|
|
desc_order: 是否降序
|
|
|
|
Returns:
|
|
特征记录列表响应
|
|
"""
|
|
logger.debug(f"Querying face features with filters: {query.model_dump(exclude_unset=True)}")
|
|
|
|
features, total = self.repository.query_features(
|
|
query, page, page_size, order_by, desc_order
|
|
)
|
|
|
|
items = [FaceFeatureResponse.model_validate(f) for f in features]
|
|
|
|
return FaceFeatureListResponse(
|
|
total=total,
|
|
items=items,
|
|
page=page,
|
|
page_size=page_size
|
|
)
|
|
|
|
def update_feature(
|
|
self,
|
|
feature_id: int,
|
|
update_data: FaceFeatureUpdate
|
|
) -> Optional[FaceFeatureResponse]:
|
|
"""
|
|
更新特征记录
|
|
|
|
Args:
|
|
feature_id: 特征记录ID
|
|
update_data: 更新数据
|
|
|
|
Returns:
|
|
更新后的特征记录响应或None
|
|
"""
|
|
logger.info(f"Updating face feature: id={feature_id}")
|
|
|
|
# 业务逻辑验证
|
|
if update_data.status is not None:
|
|
self._validate_status_transition(feature_id, update_data.status)
|
|
|
|
# 更新记录
|
|
feature = self.repository.update(feature_id, update_data)
|
|
if not feature:
|
|
return None
|
|
|
|
return FaceFeatureResponse.model_validate(feature)
|
|
|
|
def update_feature_data(self, feature_id: int, feature_data: bytes) -> bool:
|
|
"""
|
|
更新特征数据
|
|
|
|
Args:
|
|
feature_id: 特征记录ID
|
|
feature_data: 特征数据
|
|
|
|
Returns:
|
|
是否成功更新
|
|
"""
|
|
logger.info(f"Updating feature data for face feature: id={feature_id}")
|
|
|
|
# 验证特征数据
|
|
if not feature_data or len(feature_data) == 0:
|
|
raise ValueError("Feature data cannot be empty")
|
|
|
|
if len(feature_data) > 1024 * 1024: # 1MB限制
|
|
raise ValueError("Feature data is too large (max 1MB)")
|
|
|
|
return self.repository.update_feature_data(feature_id, feature_data)
|
|
|
|
def update_status(
|
|
self,
|
|
feature_id: int,
|
|
status: FeatureStatus,
|
|
start_time: Optional[datetime] = None,
|
|
finish_time: Optional[datetime] = None
|
|
) -> bool:
|
|
"""
|
|
更新计算状态
|
|
|
|
Args:
|
|
feature_id: 特征记录ID
|
|
status: 新状态
|
|
start_time: 开始时间
|
|
finish_time: 结束时间
|
|
|
|
Returns:
|
|
是否成功更新
|
|
"""
|
|
logger.info(f"Updating status to {status} for face feature: id={feature_id}")
|
|
|
|
# 验证状态转换
|
|
self._validate_status_transition(feature_id, status)
|
|
|
|
return self.repository.update_status(feature_id, status, start_time, finish_time)
|
|
|
|
def delete_feature(self, feature_id: int) -> bool:
|
|
"""
|
|
删除特征记录
|
|
|
|
Args:
|
|
feature_id: 特征记录ID
|
|
|
|
Returns:
|
|
是否成功删除
|
|
"""
|
|
logger.info(f"Deleting face feature: id={feature_id}")
|
|
|
|
return self.repository.delete(feature_id)
|
|
|
|
def delete_features_by_person(self, person_id: int) -> int:
|
|
"""
|
|
删除指定人员的所有特征记录
|
|
|
|
Args:
|
|
person_id: 人员ID
|
|
|
|
Returns:
|
|
删除的记录数
|
|
"""
|
|
logger.info(f"Deleting all face features for person_id={person_id}")
|
|
|
|
return self.repository.delete_by_person(person_id)
|
|
|
|
# ===== 业务操作 =====
|
|
|
|
def start_processing(self, feature_id: int) -> bool:
|
|
"""
|
|
开始处理特征计算
|
|
|
|
Args:
|
|
feature_id: 特征记录ID
|
|
|
|
Returns:
|
|
是否成功开始
|
|
"""
|
|
logger.info(f"Starting processing for face feature: id={feature_id}")
|
|
|
|
# 获取当前特征
|
|
feature = self.repository.get_by_id(feature_id)
|
|
if not feature:
|
|
return False
|
|
|
|
# 验证状态
|
|
if feature.status != FeatureStatusEnum.NOT_STARTED:
|
|
raise ValueError(
|
|
f"Cannot start processing for feature with status {feature.status_name}"
|
|
)
|
|
|
|
# 更新状态
|
|
return self.repository.update_status(
|
|
feature_id,
|
|
FeatureStatus.PROCESSING,
|
|
start_time=datetime.now()
|
|
)
|
|
|
|
def finish_processing(self, feature_id: int, success: bool = True) -> bool:
|
|
"""
|
|
完成特征计算
|
|
|
|
Args:
|
|
feature_id: 特征记录ID
|
|
success: 是否成功
|
|
|
|
Returns:
|
|
是否成功完成
|
|
"""
|
|
logger.info(f"Finishing processing for face feature: id={feature_id}, success={success}")
|
|
|
|
# 获取当前特征
|
|
feature = self.repository.get_by_id(feature_id)
|
|
if not feature:
|
|
return False
|
|
|
|
# 验证状态
|
|
if feature.status != FeatureStatusEnum.PROCESSING:
|
|
raise ValueError(
|
|
f"Cannot finish processing for feature with status {feature.status_name}"
|
|
)
|
|
|
|
# 更新状态
|
|
status = FeatureStatus.SUCCESS if success else FeatureStatus.FAILED
|
|
return self.repository.update_status(
|
|
feature_id,
|
|
status,
|
|
finish_time=datetime.now()
|
|
)
|
|
|
|
def process_pending_features(self, limit: int = 100) -> List[FaceFeatureResponse]:
|
|
"""
|
|
处理待计算的特征记录
|
|
|
|
Args:
|
|
limit: 最大处理数量
|
|
|
|
Returns:
|
|
处理中的特征记录列表
|
|
"""
|
|
logger.info(f"Processing up to {limit} pending face features")
|
|
|
|
features = self.repository.mark_for_processing(limit)
|
|
return [FaceFeatureResponse.model_validate(f) for f in features]
|
|
|
|
# ===== 统计和分析 =====
|
|
|
|
def get_statistics(self) -> FaceFeatureStatsResponse:
|
|
"""
|
|
获取特征记录统计信息
|
|
|
|
Returns:
|
|
统计信息响应
|
|
"""
|
|
logger.debug("Getting face feature statistics")
|
|
|
|
stats = self.repository.get_stats()
|
|
|
|
# 转换状态枚举名称
|
|
by_status = {}
|
|
for status_value, count in stats["by_status"].items():
|
|
try:
|
|
status_name = FeatureStatusEnum(int(status_value)).name
|
|
by_status[status_name] = count
|
|
except ValueError:
|
|
by_status[f"未知({status_value})"] = count
|
|
|
|
return FaceFeatureStatsResponse(
|
|
total_count=stats["total_count"],
|
|
by_status=by_status,
|
|
by_feature_type=stats["by_feature_type"],
|
|
avg_processing_time=stats["avg_processing_time"]
|
|
)
|
|
|
|
def get_person_statistics(self, person_id: int) -> Dict[str, Any]:
|
|
"""
|
|
获取人员的特征统计信息
|
|
|
|
Args:
|
|
person_id: 人员ID
|
|
|
|
Returns:
|
|
人员统计信息
|
|
"""
|
|
logger.debug(f"Getting statistics for person_id={person_id}")
|
|
|
|
# 获取该人员的所有特征
|
|
features = self.repository.get_by_person(person_id, limit=1000)
|
|
|
|
if not features:
|
|
return {
|
|
"person_id": person_id,
|
|
"total_features": 0,
|
|
"status_summary": {},
|
|
"feature_types": []
|
|
}
|
|
|
|
# 统计信息
|
|
status_summary = {}
|
|
feature_types = set()
|
|
successful_features = []
|
|
|
|
for feature in features:
|
|
# 状态统计
|
|
status_name = feature.status_name
|
|
status_summary[status_name] = status_summary.get(status_name, 0) + 1
|
|
|
|
# 特征类型
|
|
if feature.feature_type is not None:
|
|
feature_types.add(feature.feature_type)
|
|
|
|
# 成功完成的特征
|
|
if feature.status == FeatureStatusEnum.SUCCESS:
|
|
successful_features.append(feature)
|
|
|
|
# 计算平均处理时间（仅成功记录）
|
|
total_time = 0
|
|
valid_count = 0
|
|
|
|
for feature in successful_features:
|
|
if feature.processing_time:
|
|
total_time += feature.processing_time
|
|
valid_count += 1
|
|
|
|
avg_time = total_time / valid_count if valid_count > 0 else None
|
|
|
|
return {
|
|
"person_id": person_id,
|
|
"total_features": len(features),
|
|
"status_summary": status_summary,
|
|
"feature_types": sorted(list(feature_types)),
|
|
"avg_processing_time": avg_time,
|
|
"successful_count": len(successful_features)
|
|
}
|
|
|
|
def cleanup_old_records(self, days: int = 30) -> Dict[str, Any]:
|
|
"""
|
|
清理旧的特征记录
|
|
|
|
Args:
|
|
days: 保留天数
|
|
|
|
Returns:
|
|
清理结果
|
|
"""
|
|
logger.info(f"Cleaning up face features older than {days} days")
|
|
|
|
# 先获取清理前的统计
|
|
before_stats = self.repository.get_stats()
|
|
|
|
# 执行清理
|
|
deleted_count = self.repository.cleanup_old_features(days)
|
|
|
|
# 获取清理后的统计
|
|
after_stats = self.repository.get_stats()
|
|
|
|
return {
|
|
"days_retained": days,
|
|
"deleted_count": deleted_count,
|
|
"before_total": before_stats["total_count"],
|
|
"after_total": after_stats["total_count"],
|
|
"reduction_percentage": (
|
|
(before_stats["total_count"] - after_stats["total_count"]) /
|
|
before_stats["total_count"] * 100
|
|
if before_stats["total_count"] > 0 else 0
|
|
)
|
|
}
|
|
|
|
# ===== 私有方法 =====
|
|
|
|
def _validate_feature_data(self, feature_data: FaceFeatureCreate) -> None:
|
|
"""
|
|
验证特征数据
|
|
|
|
Args:
|
|
feature_data: 特征数据
|
|
|
|
Raises:
|
|
ValueError: 如果数据无效
|
|
"""
|
|
# 验证人员ID
|
|
if feature_data.person_id <= 0:
|
|
raise ValueError("person_id must be greater than 0")
|
|
|
|
# 验证特征类型
|
|
if feature_data.feature_type is not None and feature_data.feature_type < 0:
|
|
raise ValueError("feature_type must be non-negative")
|
|
|
|
# 验证特征数据大小
|
|
if feature_data.feature_data and len(feature_data.feature_data) > 1024 * 1024:
|
|
raise ValueError("feature_data is too large (max 1MB)")
|
|
|
|
# 验证状态
|
|
if feature_data.status:
|
|
try:
|
|
FeatureStatus(feature_data.status)
|
|
except ValueError:
|
|
raise ValueError(f"Invalid status value: {feature_data.status}")
|
|
|
|
def _validate_status_transition(self, feature_id: int, new_status: FeatureStatus) -> None:
|
|
"""
|
|
验证状态转换是否有效
|
|
|
|
Args:
|
|
feature_id: 特征记录ID
|
|
new_status: 新状态
|
|
|
|
Raises:
|
|
ValueError: 如果状态转换无效
|
|
"""
|
|
# 获取当前特征
|
|
feature = self.repository.get_by_id(feature_id)
|
|
if not feature:
|
|
return
|
|
|
|
current_status = FeatureStatusEnum(feature.status)
|
|
|
|
# 定义允许的状态转换
|
|
allowed_transitions = {
|
|
FeatureStatusEnum.NOT_STARTED: [FeatureStatusEnum.PROCESSING],
|
|
FeatureStatusEnum.PROCESSING: [FeatureStatusEnum.SUCCESS, FeatureStatusEnum.FAILED],
|
|
FeatureStatusEnum.SUCCESS: [],
|
|
FeatureStatusEnum.FAILED: [FeatureStatusEnum.PROCESSING] # 允许重试
|
|
}
|
|
|
|
new_status_enum = FeatureStatusEnum(new_status.value if isinstance(new_status, FeatureStatus) else new_status)
|
|
|
|
# 检查转换是否允许
|
|
if new_status_enum not in allowed_transitions.get(current_status, []):
|
|
raise ValueError(
|
|
f"Cannot transition from {current_status.name} to {new_status_enum.name}"
|
|
) |