feat: WebCam feature

backend/app/database/face_clusters.py

@@ -195,9 +195,9 @@ def db_delete_all_clusters() -> int:
     return deleted_count
 
 
-def db_get_all_clusters_with_face_counts() -> List[
-    Dict[str, Union[str, Optional[str], int]]
-]:
+def db_get_all_clusters_with_face_counts() -> (
+    List[Dict[str, Union[str, Optional[str], int]]]
+):
     """
     Retrieve all clusters with their face counts and stored face images.
 

backend/app/database/faces.py

@@ -234,9 +234,9 @@ def db_get_faces_unassigned_clusters() -> List[Dict[str, Union[FaceId, FaceEmbed
     return faces
 
 
-def db_get_all_faces_with_cluster_names() -> List[
-    Dict[str, Union[FaceId, FaceEmbedding, Optional[str]]]
-]:
+def db_get_all_faces_with_cluster_names() -> (
+    List[Dict[str, Union[FaceId, FaceEmbedding, Optional[str]]]]
+):
     """
     Get all faces with their corresponding cluster names.
 
@@ -271,7 +271,7 @@ def db_get_all_faces_with_cluster_names() -> List[
 
 
 def db_update_face_cluster_ids_batch(
-    face_cluster_mapping: List[Dict[str, Union[FaceId, ClusterId]]]
+    face_cluster_mapping: List[Dict[str, Union[FaceId, ClusterId]]],
 ) -> None:
     """
     Update cluster IDs for multiple faces in batch.

backend/app/database/folders.py

@@ -382,9 +382,9 @@ def db_get_folder_ids_by_paths(
         conn.close()
 
 
-def db_get_all_folder_details() -> List[
-    Tuple[str, str, Optional[str], int, bool, Optional[bool]]
-]:
+def db_get_all_folder_details() -> (
+    List[Tuple[str, str, Optional[str], int, bool, Optional[bool]]]
+):
     """
     Get all folder details including folder_id, folder_path, parent_folder_id,
     last_modified_time, AI_Tagging, and taggingCompleted.

backend/app/routes/albums.py

@@ -29,6 +29,7 @@
 
 router = APIRouter()
 
+
 # GET /albums/ - Get all albums
 @router.get("/", response_model=GetAlbumsResponse)
 def get_albums(show_hidden: bool = Query(False)):

backend/app/routes/face_clusters.py

@@ -1,19 +1,15 @@
 import logging
+from binascii import Error as Base64Error
+import base64
 import uuid
 import os
-from typing import Optional, List, Dict, Any
-from pydantic import BaseModel
-from app.config.settings import CONFIDENCE_PERCENT, DEFAULT_FACENET_MODEL
 from fastapi import APIRouter, HTTPException, status
 from app.database.face_clusters import (
     db_get_cluster_by_id,
     db_update_cluster,
     db_get_all_clusters_with_face_counts,
     db_get_images_by_cluster_id,  # Add this import
 )
-from app.database.faces import get_all_face_embeddings
-from app.models.FaceDetector import FaceDetector
-from app.models.FaceNet import FaceNet
 from app.schemas.face_clusters import (
     RenameClusterRequest,
     RenameClusterResponse,
@@ -26,32 +22,8 @@
     GetClusterImagesData,
     ImageInCluster,
 )
-from app.schemas.images import AddSingleImageRequest
-from app.utils.FaceNet import FaceNet_util_cosine_similarity
-
-
-class BoundingBox(BaseModel):
-    x: float
-    y: float
-    width: float
-    height: float
-
-
-class ImageData(BaseModel):
-    id: str
-    path: str
-    folder_id: str
-    thumbnailPath: str
-    metadata: Dict[str, Any]
-    isTagged: bool
-    tags: Optional[List[str]] = None
-    bboxes: BoundingBox
-
-
-class GetAllImagesResponse(BaseModel):
-    success: bool
-    message: str
-    data: List[ImageData]
+from app.schemas.images import AddSingleBase64ImageRequest, AddSingleImageRequest
+from app.utils.faceSearch import perform_face_search
 
 
 logger = logging.getLogger(__name__)
@@ -247,56 +219,46 @@ def face_tagging(payload: AddSingleImageRequest):
                 message="The provided path is not a valid file",
             ).model_dump(),
         )
+    return perform_face_search(image_path)
+
+
+@router.post("/face-search-base64")
+def face_search_base64(payload: AddSingleBase64ImageRequest):
+    base64_data = payload.base64_data
+    if not base64_data:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=ErrorResponse(
+                success=False,
+                error="No base64 data",
+                message="Base64 image data is required.",
+            ).model_dump(),
+        )
 
-    fd = FaceDetector()
-    fn = FaceNet(DEFAULT_FACENET_MODEL)
+    image_path = None
     try:
-        matches = []
-        image_id = str(uuid.uuid4())
-        result = fd.detect_faces(image_id, image_path, forSearch=True)
-        if not result or result["num_faces"] == 0:
-            return GetAllImagesResponse(
-                success=True,
-                message=f"Successfully retrieved {len(matches)} images",
-                data=[],
+        try:
+            image_bytes = base64.b64decode(base64_data.split(",")[-1])
+        except (Base64Error, ValueError):
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=ErrorResponse(
+                    success=False,
+                    error="Invalid base64 data",
+                    message="The provided base64 image data is malformed or invalid.",
+                ).model_dump(),
             )
+        image_id = str(uuid.uuid4())[:8]
+        temp_dir = "temp_uploads"
+        os.makedirs(temp_dir, exist_ok=True)
+        image_path = os.path.join(temp_dir, f"{image_id}.jpeg")
 
-        process_face = result["processed_faces"][0]
-        new_embedding = fn.get_embedding(process_face)
+        with open(image_path, "wb") as f:
+            f.write(image_bytes)
 
-        images = get_all_face_embeddings()
-        if len(images) == 0:
-            return GetAllImagesResponse(
-                success=True,
-                message=f"Successfully retrieved {len(matches)} images",
-                data=[],
-            )
-        else:
-            for image in images:
-                max_similarity = 0
-                similarity = FaceNet_util_cosine_similarity(
-                    new_embedding, image["embeddings"]
-                )
-                max_similarity = max(max_similarity, similarity)
-                if max_similarity >= CONFIDENCE_PERCENT:
-                    matches.append(
-                        ImageData(
-                            id=image["id"],
-                            path=image["path"],
-                            folder_id=image["folder_id"],
-                            thumbnailPath=image["thumbnailPath"],
-                            metadata=image["metadata"],
-                            isTagged=image["isTagged"],
-                            tags=image["tags"],
-                            bboxes=image["bbox"],
-                        )
-                    )
+        result = perform_face_search(image_path)
+        return result
 
-            return GetAllImagesResponse(
-                success=True,
-                message=f"Successfully retrieved {len(matches)} images",
-                data=matches,
-            )
     finally:
-        fd.close()
-        fn.close()
+        if image_path and os.path.exists(image_path):
+            os.remove(image_path)

backend/app/schemas/images.py

@@ -7,6 +7,10 @@ class AddSingleImageRequest(BaseModel):
     path: str
 
 
+class AddSingleBase64ImageRequest(BaseModel):
+    base64_data: str
+
+
 class AddMultipleImagesRequest(BaseModel):
     paths: List[str]
 

backend/app/utils/faceSearch.py

@@ -0,0 +1,106 @@
+import uuid
+from typing import Optional, List, Dict, Any
+from pydantic import BaseModel
+from app.config.settings import CONFIDENCE_PERCENT, DEFAULT_FACENET_MODEL
+from app.database.faces import get_all_face_embeddings
+from app.models.FaceDetector import FaceDetector
+from app.models.FaceNet import FaceNet
+from app.utils.FaceNet import FaceNet_util_cosine_similarity
+
+
+class BoundingBox(BaseModel):
+    x: float
+    y: float
+    width: float
+    height: float
+
+
+class ImageData(BaseModel):
+    id: str
+    path: str
+    folder_id: str
+    thumbnailPath: str
+    metadata: Dict[str, Any]
+    isTagged: bool
+    tags: Optional[List[str]] = None
+    bboxes: BoundingBox
+
+
+class GetAllImagesResponse(BaseModel):
+    success: bool
+    message: str
+    data: List[ImageData]
+
+
+def perform_face_search(image_path: str) -> GetAllImagesResponse:
+    """
+    Performs face detection, embedding generation, and similarity search.
+
+    Args:
+        image_path (str): Path to the image file to process.
+
+    Returns:
+        GetAllImagesResponse: Search result containing matched images.
+    """
+    fd = FaceDetector()
+    fn = FaceNet(DEFAULT_FACENET_MODEL)
+
+    try:
+        matches = []
+        image_id = str(uuid.uuid4())
+
+        try:
+            result = fd.detect_faces(image_id, image_path, forSearch=True)
+        except Exception as e:
+            return GetAllImagesResponse(
+                success=False,
+                message=f"Failed to process image: {str(e)}",
+                data=[],
+            )
+        if not result or result["num_faces"] == 0:
+            return GetAllImagesResponse(
+                success=True,
+                message="No faces detected in the image.",
+                data=[],
+            )
+
+        process_face = result["processed_faces"][0]
+        new_embedding = fn.get_embedding(process_face)
+
+        images = get_all_face_embeddings()
+        if not images:
+            return GetAllImagesResponse(
+                success=True,
+                message="No face embeddings available for comparison.",
+                data=[],
+            )
+
+        for image in images:
+            similarity = FaceNet_util_cosine_similarity(
+                new_embedding, image["embeddings"]
+            )
+            if similarity >= CONFIDENCE_PERCENT:
+                matches.append(
+                    ImageData(
+                        id=image["id"],
+                        path=image["path"],
+                        folder_id=image["folder_id"],
+                        thumbnailPath=image["thumbnailPath"],
+                        metadata=image["metadata"],
+                        isTagged=image["isTagged"],
+                        tags=image["tags"],
+                        bboxes=image["bbox"],
+                    )
+                )
+
+        return GetAllImagesResponse(
+            success=True,
+            message=f"Successfully retrieved {len(matches)} matching images.",
+            data=matches,
+        )
+
+    finally:
+        if "fd" in locals() and fd is not None:
+            fd.close()
+        if "fn" in locals() and fn is not None:
+            fn.close()

backend/app/utils/images.py

@@ -99,7 +99,7 @@ def image_util_process_untagged_images() -> bool:
 
 
 def image_util_classify_and_face_detect_images(
-    untagged_images: List[Dict[str, str]]
+    untagged_images: List[Dict[str, str]],
 ) -> None:
     """Classify untagged images and detect faces if applicable."""
     object_classifier = ObjectClassifier()
@@ -262,7 +262,7 @@ def image_util_remove_obsolete_images(folder_id_list: List[int]) -> int:
 
 
 def image_util_create_folder_path_mapping(
-    folder_ids: List[Tuple[int, str]]
+    folder_ids: List[Tuple[int, str]],
 ) -> Dict[str, int]:
     """
     Create a dictionary mapping folder paths to their IDs.