feat: Implement auto-match people and person matches API with frontend integration
This commit introduces new API endpoints for retrieving a list of people for auto-matching and fetching matches for specific individuals. The frontend has been updated to utilize these endpoints, allowing for lazy loading of matches and improved state management. The AutoMatch component now supports caching of matches and session storage for user settings, enhancing performance and user experience. Documentation has been updated to reflect these changes.
This commit is contained in:
tanyar09
@@ -30,6 +30,9 @@ from src.web.schemas.faces import (
|
||||
AutoMatchResponse,
|
||||
AutoMatchPersonItem,
|
||||
AutoMatchFaceItem,
|
||||
AutoMatchPeopleResponse,
|
||||
AutoMatchPersonSummary,
|
||||
AutoMatchPersonMatchesResponse,
|
||||
AcceptMatchesRequest,
|
||||
MaintenanceFacesResponse,
|
||||
MaintenanceFaceItem,
|
||||
@@ -44,6 +47,8 @@ from src.web.services.face_service import (
|
||||
calculate_batch_similarities,
|
||||
find_auto_match_matches,
|
||||
accept_auto_match_matches,
|
||||
get_auto_match_people_list,
|
||||
get_auto_match_person_matches as get_person_matches_service,
|
||||
)
|
||||
# Note: Function passed as string path to avoid RQ serialization issues
|
||||
|
||||
@@ -702,6 +707,125 @@ def auto_match_faces(
|
||||
)
|
||||
|
||||
|
||||
@router.get("/auto-match/people", response_model=AutoMatchPeopleResponse)
|
||||
def get_auto_match_people(
|
||||
filter_frontal_only: bool = Query(False, description="Only include frontal/tilted reference faces"),
|
||||
tolerance: float = Query(0.6, ge=0.0, le=1.0, description="Tolerance threshold"),
|
||||
db: Session = Depends(get_db),
|
||||
) -> AutoMatchPeopleResponse:
|
||||
"""Get list of people for auto-match (without matches) - fast initial load.
|
||||
|
||||
Returns just the people list with reference faces, without calculating matches.
|
||||
This allows fast initial page load, then matches can be loaded on-demand via
|
||||
/auto-match/people/{person_id}/matches endpoint.
|
||||
|
||||
Note: Only returns people if there are unidentified faces in the database
|
||||
(since people can't have matches if there are no unidentified faces).
|
||||
"""
|
||||
from src.web.db.models import Person, Photo
|
||||
|
||||
# Get people list (fast - no match calculations, but checks for unidentified faces)
|
||||
people_data = get_auto_match_people_list(
|
||||
db,
|
||||
filter_frontal_only=filter_frontal_only,
|
||||
tolerance=tolerance
|
||||
)
|
||||
|
||||
if not people_data:
|
||||
return AutoMatchPeopleResponse(people=[], total_people=0)
|
||||
|
||||
# Build response
|
||||
people_items = []
|
||||
for person_id, reference_face, person_name, face_count in people_data:
|
||||
# Get person details
|
||||
person = db.query(Person).filter(Person.id == person_id).first()
|
||||
if not person:
|
||||
continue
|
||||
|
||||
# Get reference face photo info
|
||||
reference_photo = db.query(Photo).filter(Photo.id == reference_face.photo_id).first()
|
||||
if not reference_photo:
|
||||
continue
|
||||
|
||||
# Get reference face pose_mode
|
||||
reference_pose_mode = reference_face.pose_mode or 'frontal'
|
||||
|
||||
people_items.append(
|
||||
AutoMatchPersonSummary(
|
||||
person_id=person_id,
|
||||
person_name=person_name,
|
||||
reference_face_id=reference_face.id,
|
||||
reference_photo_id=reference_face.photo_id,
|
||||
reference_photo_filename=reference_photo.filename,
|
||||
reference_location=reference_face.location,
|
||||
reference_pose_mode=reference_pose_mode,
|
||||
face_count=face_count,
|
||||
total_matches=0, # Will be loaded separately
|
||||
)
|
||||
)
|
||||
|
||||
return AutoMatchPeopleResponse(
|
||||
people=people_items,
|
||||
total_people=len(people_items),
|
||||
)
|
||||
|
||||
|
||||
@router.get("/auto-match/people/{person_id}/matches", response_model=AutoMatchPersonMatchesResponse)
|
||||
def get_auto_match_person_matches(
|
||||
person_id: int,
|
||||
tolerance: float = Query(0.6, ge=0.0, le=1.0, description="Tolerance threshold"),
|
||||
filter_frontal_only: bool = Query(False, description="Only return frontal/tilted faces"),
|
||||
db: Session = Depends(get_db),
|
||||
) -> AutoMatchPersonMatchesResponse:
|
||||
"""Get matches for a specific person - for lazy loading.
|
||||
|
||||
This endpoint is called on-demand when user navigates to a person.
|
||||
"""
|
||||
from src.web.db.models import Photo
|
||||
|
||||
# Get matches for this person
|
||||
similar_faces = get_person_matches_service(
|
||||
db,
|
||||
person_id=person_id,
|
||||
tolerance=tolerance,
|
||||
filter_frontal_only=filter_frontal_only,
|
||||
)
|
||||
|
||||
if not similar_faces:
|
||||
return AutoMatchPersonMatchesResponse(
|
||||
person_id=person_id,
|
||||
matches=[],
|
||||
total_matches=0,
|
||||
)
|
||||
|
||||
# Build matches list
|
||||
match_items = []
|
||||
for face, distance, confidence_pct in similar_faces:
|
||||
# Get photo info for this match
|
||||
match_photo = db.query(Photo).filter(Photo.id == face.photo_id).first()
|
||||
if not match_photo:
|
||||
continue
|
||||
|
||||
match_items.append(
|
||||
AutoMatchFaceItem(
|
||||
id=face.id,
|
||||
photo_id=face.photo_id,
|
||||
photo_filename=match_photo.filename,
|
||||
location=face.location,
|
||||
quality_score=float(face.quality_score),
|
||||
similarity=confidence_pct, # Confidence percentage (0-100)
|
||||
distance=distance,
|
||||
pose_mode=face.pose_mode or 'frontal',
|
||||
)
|
||||
)
|
||||
|
||||
return AutoMatchPersonMatchesResponse(
|
||||
person_id=person_id,
|
||||
matches=match_items,
|
||||
total_matches=len(match_items),
|
||||
)
|
||||
|
||||
|
||||
@router.get("/maintenance", response_model=MaintenanceFacesResponse)
|
||||
def list_all_faces(
|
||||
page: int = Query(1, ge=1),
|
||||
|
||||
@@ -248,6 +248,41 @@ class AutoMatchPersonItem(BaseModel):
|
||||
total_matches: int
|
||||
|
||||
|
||||
class AutoMatchPersonSummary(BaseModel):
|
||||
"""Person summary without matches (for fast initial load)."""
|
||||
|
||||
model_config = ConfigDict(protected_namespaces=())
|
||||
|
||||
person_id: int
|
||||
person_name: str
|
||||
reference_face_id: int
|
||||
reference_photo_id: int
|
||||
reference_photo_filename: str
|
||||
reference_location: str
|
||||
reference_pose_mode: str = Field("frontal", description="Reference face pose classification")
|
||||
face_count: int # Number of faces already identified for this person
|
||||
total_matches: int = Field(0, description="Total matches (loaded separately)")
|
||||
|
||||
|
||||
class AutoMatchPeopleResponse(BaseModel):
|
||||
"""Response containing people list without matches (for fast initial load)."""
|
||||
|
||||
model_config = ConfigDict(protected_namespaces=())
|
||||
|
||||
people: list[AutoMatchPersonSummary]
|
||||
total_people: int
|
||||
|
||||
|
||||
class AutoMatchPersonMatchesResponse(BaseModel):
|
||||
"""Response containing matches for a specific person."""
|
||||
|
||||
model_config = ConfigDict(protected_namespaces=())
|
||||
|
||||
person_id: int
|
||||
matches: list[AutoMatchFaceItem]
|
||||
total_matches: int
|
||||
|
||||
|
||||
class AutoMatchResponse(BaseModel):
|
||||
"""Response from auto-match start operation."""
|
||||
|
||||
|
||||
@@ -12,7 +12,7 @@ from datetime import date
|
||||
import numpy as np
|
||||
from PIL import Image
|
||||
from sqlalchemy.orm import Session, joinedload
|
||||
from sqlalchemy import and_, func
|
||||
from sqlalchemy import and_, func, case
|
||||
|
||||
try:
|
||||
from deepface import DeepFace
|
||||
@@ -1299,6 +1299,20 @@ def list_unidentified_faces(
|
||||
query = query.order_by(sort_col.asc().nullslast())
|
||||
else:
|
||||
query = query.order_by(sort_col.desc().nullslast())
|
||||
|
||||
# Add secondary sort by quality (descending, best first) when primary sort is not quality
|
||||
if sort_by != "quality":
|
||||
query = query.order_by(Face.quality_score.desc().nullslast())
|
||||
|
||||
# Add tertiary sort by pose mode: frontal first, then tilted, then profile last
|
||||
# Check profile first to catch poses that might contain both 'tilted' and 'profile'
|
||||
pose_order = case(
|
||||
(func.lower(Face.pose_mode).like('%profile%'), 2), # Profile = 2 (last)
|
||||
(func.lower(Face.pose_mode).like('%tilted%'), 1), # Tilted = 1 (second)
|
||||
(func.lower(Face.pose_mode).like('frontal%'), 0), # Frontal = 0 (first)
|
||||
else_=2 # Default to last for unknown poses
|
||||
)
|
||||
query = query.order_by(pose_order.asc())
|
||||
|
||||
# Total count for pagination
|
||||
total = query.count()
|
||||
@@ -1752,13 +1766,21 @@ def find_auto_match_matches(
|
||||
# FROM faces f
|
||||
# JOIN photos p ON f.photo_id = p.id
|
||||
# WHERE f.person_id IS NOT NULL AND f.quality_score >= 0.3
|
||||
# ORDER BY f.person_id, f.quality_score DESC
|
||||
# ORDER BY f.person_id, f.quality_score DESC, pose_mode (frontal first, then tilted, then profile)
|
||||
# Add pose mode ordering: frontal first, then tilted, then profile last
|
||||
pose_order = case(
|
||||
(func.lower(Face.pose_mode).like('%profile%'), 2), # Profile = 2 (last)
|
||||
(func.lower(Face.pose_mode).like('%tilted%'), 1), # Tilted = 1 (second)
|
||||
(func.lower(Face.pose_mode).like('frontal%'), 0), # Frontal = 0 (first)
|
||||
else_=2 # Default to last for unknown poses
|
||||
)
|
||||
|
||||
identified_faces: List[Face] = (
|
||||
db.query(Face)
|
||||
.join(Photo, Face.photo_id == Photo.id)
|
||||
.filter(Face.person_id.isnot(None))
|
||||
.filter(Face.quality_score >= 0.3)
|
||||
.order_by(Face.person_id, Face.quality_score.desc())
|
||||
.order_by(Face.person_id, Face.quality_score.desc(), pose_order.asc())
|
||||
.all()
|
||||
)
|
||||
|
||||
@@ -1824,6 +1846,153 @@ def find_auto_match_matches(
|
||||
return results
|
||||
|
||||
|
||||
def get_auto_match_people_list(
|
||||
db: Session,
|
||||
filter_frontal_only: bool = False,
|
||||
tolerance: float = 0.6,
|
||||
) -> List[Tuple[int, Face, str, int]]:
|
||||
"""Get list of people for auto-match (without matches) - fast initial load.
|
||||
|
||||
Returns just the people list with reference faces, without calculating matches.
|
||||
This allows fast initial page load, then matches can be loaded on-demand.
|
||||
|
||||
However, we do a quick check to see if there are any unidentified faces at all.
|
||||
If there are no unidentified faces, we return an empty list (no one can have matches).
|
||||
|
||||
Args:
|
||||
filter_frontal_only: Only include persons with frontal or tilted reference face (not profile)
|
||||
tolerance: Similarity tolerance (used to check if there are potential matches)
|
||||
|
||||
Returns:
|
||||
List of (person_id, reference_face, person_name, face_count) tuples
|
||||
"""
|
||||
from src.web.db.models import Person, Photo
|
||||
from src.core.config import DEFAULT_FACE_TOLERANCE
|
||||
from sqlalchemy import func, case
|
||||
|
||||
if tolerance is None:
|
||||
tolerance = DEFAULT_FACE_TOLERANCE
|
||||
|
||||
# Quick check: if there are no unidentified faces, no one can have matches
|
||||
# This is a fast query that avoids loading people who can't possibly have matches
|
||||
unidentified_count = (
|
||||
db.query(func.count(Face.id))
|
||||
.filter(Face.person_id.is_(None))
|
||||
.scalar() or 0
|
||||
)
|
||||
|
||||
if unidentified_count == 0:
|
||||
return []
|
||||
|
||||
# Get all identified faces (one per person) to use as reference faces
|
||||
# Same logic as find_auto_match_matches but without finding matches
|
||||
pose_order = case(
|
||||
(func.lower(Face.pose_mode).like('%profile%'), 2), # Profile = 2 (last)
|
||||
(func.lower(Face.pose_mode).like('%tilted%'), 1), # Tilted = 1 (second)
|
||||
(func.lower(Face.pose_mode).like('frontal%'), 0), # Frontal = 0 (first)
|
||||
else_=2 # Default to last for unknown poses
|
||||
)
|
||||
|
||||
identified_faces: List[Face] = (
|
||||
db.query(Face)
|
||||
.join(Photo, Face.photo_id == Photo.id)
|
||||
.filter(Face.person_id.isnot(None))
|
||||
.filter(Face.quality_score >= 0.3)
|
||||
.order_by(Face.person_id, Face.quality_score.desc(), pose_order.asc())
|
||||
.all()
|
||||
)
|
||||
|
||||
if not identified_faces:
|
||||
return []
|
||||
|
||||
# Filter by pose_mode if requested (only frontal or tilted faces)
|
||||
if filter_frontal_only:
|
||||
identified_faces = [
|
||||
f for f in identified_faces
|
||||
if _is_acceptable_pose_for_auto_match(f.pose_mode)
|
||||
]
|
||||
|
||||
if not identified_faces:
|
||||
return []
|
||||
|
||||
# Group by person and get the best quality face per person
|
||||
person_faces: Dict[int, Face] = {}
|
||||
for face in identified_faces:
|
||||
person_id = face.person_id
|
||||
if person_id not in person_faces:
|
||||
person_faces[person_id] = face
|
||||
|
||||
# Convert to ordered list with person names
|
||||
person_faces_list = []
|
||||
for person_id, face in person_faces.items():
|
||||
# Get person name for ordering
|
||||
person = db.query(Person).filter(Person.id == person_id).first()
|
||||
if person:
|
||||
if person.last_name and person.first_name:
|
||||
person_name = f"{person.last_name}, {person.first_name}"
|
||||
elif person.last_name:
|
||||
person_name = person.last_name
|
||||
elif person.first_name:
|
||||
person_name = person.first_name
|
||||
else:
|
||||
person_name = "Unknown"
|
||||
else:
|
||||
person_name = "Unknown"
|
||||
|
||||
# Get face count for this person
|
||||
face_count = (
|
||||
db.query(func.count(Face.id))
|
||||
.filter(Face.person_id == person_id)
|
||||
.scalar() or 0
|
||||
)
|
||||
|
||||
person_faces_list.append((person_id, face, person_name, face_count))
|
||||
|
||||
# Sort by person name for consistent, user-friendly ordering
|
||||
person_faces_list.sort(key=lambda x: x[2]) # Sort by person name (index 2)
|
||||
|
||||
return person_faces_list
|
||||
|
||||
|
||||
def get_auto_match_person_matches(
|
||||
db: Session,
|
||||
person_id: int,
|
||||
tolerance: float = 0.6,
|
||||
filter_frontal_only: bool = False,
|
||||
) -> List[Tuple[Face, float, float]]:
|
||||
"""Get matches for a specific person - for lazy loading.
|
||||
|
||||
Args:
|
||||
person_id: Person ID to get matches for
|
||||
tolerance: Similarity tolerance (default: 0.6)
|
||||
filter_frontal_only: Only return frontal or tilted faces (not profile)
|
||||
|
||||
Returns:
|
||||
List of (face, distance, confidence_pct) tuples
|
||||
"""
|
||||
from src.web.db.models import Person, Face
|
||||
|
||||
# Get reference face for this person (best quality >= 0.3)
|
||||
reference_face = (
|
||||
db.query(Face)
|
||||
.filter(Face.person_id == person_id)
|
||||
.filter(Face.quality_score >= 0.3)
|
||||
.order_by(Face.quality_score.desc())
|
||||
.first()
|
||||
)
|
||||
|
||||
if not reference_face:
|
||||
return []
|
||||
|
||||
# Find similar faces using existing function
|
||||
similar_faces = find_similar_faces(
|
||||
db, reference_face.id, tolerance=tolerance,
|
||||
filter_frontal_only=filter_frontal_only
|
||||
)
|
||||
|
||||
return similar_faces
|
||||
|
||||
|
||||
def accept_auto_match_matches(
|
||||
db: Session,
|
||||
person_id: int,
|
||||
|
||||
Reference in New Issue
Block a user