feat: Implement photo search functionality with filtering and tagging support
This commit introduces a comprehensive photo search feature in the application, allowing users to search photos by various criteria including name, date, and tags. The API has been updated to support these search parameters, returning results that match the specified filters. Additionally, new endpoints for managing tags have been added, enabling users to add and remove tags from multiple photos. The frontend has been enhanced with a user-friendly interface for search inputs and results display, improving overall usability. Documentation and tests have been updated to reflect these new features and ensure reliability.
This commit is contained in:
tanyar09
+183
-5
@@ -2,7 +2,10 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, status
|
||||
from datetime import date, datetime
|
||||
from typing import List, Optional
|
||||
|
||||
from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile, status
|
||||
from fastapi.responses import JSONResponse, FileResponse
|
||||
from rq import Queue
|
||||
from redis import Redis
|
||||
@@ -18,19 +21,194 @@ from src.web.schemas.photos import (
|
||||
PhotoImportResponse,
|
||||
PhotoResponse,
|
||||
)
|
||||
from src.web.schemas.search import (
|
||||
PhotoSearchResult,
|
||||
SearchPhotosResponse,
|
||||
)
|
||||
from src.web.services.photo_service import (
|
||||
find_photos_in_folder,
|
||||
import_photo_from_path,
|
||||
)
|
||||
from src.web.services.search_service import (
|
||||
get_photo_face_count,
|
||||
get_photo_person,
|
||||
get_photo_tags,
|
||||
get_photos_without_faces,
|
||||
get_photos_without_tags,
|
||||
search_photos_by_date,
|
||||
search_photos_by_name,
|
||||
search_photos_by_tags,
|
||||
)
|
||||
# Note: Function passed as string path to avoid RQ serialization issues
|
||||
|
||||
router = APIRouter(prefix="/photos", tags=["photos"])
|
||||
|
||||
|
||||
@router.get("")
|
||||
def list_photos() -> dict:
|
||||
"""List photos - placeholder for Phase 2."""
|
||||
return {"message": "Photos endpoint - to be implemented in Phase 2"}
|
||||
@router.get("", response_model=SearchPhotosResponse)
|
||||
def search_photos(
|
||||
search_type: str = Query("name", description="Search type: name, date, tags, no_faces, no_tags"),
|
||||
person_name: Optional[str] = Query(None, description="Person name for name search"),
|
||||
tag_names: Optional[str] = Query(None, description="Comma-separated tag names for tag search"),
|
||||
match_all: bool = Query(False, description="Match all tags (for tag search)"),
|
||||
date_from: Optional[str] = Query(None, description="Date from (YYYY-MM-DD)"),
|
||||
date_to: Optional[str] = Query(None, description="Date to (YYYY-MM-DD)"),
|
||||
folder_path: Optional[str] = Query(None, description="Filter by folder path"),
|
||||
page: int = Query(1, ge=1),
|
||||
page_size: int = Query(50, ge=1, le=200),
|
||||
db: Session = Depends(get_db),
|
||||
) -> SearchPhotosResponse:
|
||||
"""Search photos with filters.
|
||||
|
||||
Matches desktop search functionality exactly:
|
||||
- Search by name: person_name required
|
||||
- Search by date: date_from or date_to required
|
||||
- Search by tags: tag_names required (comma-separated)
|
||||
- Search no faces: returns photos without faces
|
||||
- Search no tags: returns photos without tags
|
||||
"""
|
||||
items: List[PhotoSearchResult] = []
|
||||
total = 0
|
||||
|
||||
if search_type == "name":
|
||||
if not person_name:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="person_name is required for name search",
|
||||
)
|
||||
results, total = search_photos_by_name(
|
||||
db, person_name, folder_path, page, page_size
|
||||
)
|
||||
for photo, full_name in results:
|
||||
tags = get_photo_tags(db, photo.id)
|
||||
face_count = get_photo_face_count(db, photo.id)
|
||||
# Convert datetime to date for date_added
|
||||
date_added = photo.date_added.date() if isinstance(photo.date_added, datetime) else photo.date_added
|
||||
items.append(
|
||||
PhotoSearchResult(
|
||||
id=photo.id,
|
||||
path=photo.path,
|
||||
filename=photo.filename,
|
||||
date_taken=photo.date_taken,
|
||||
date_added=date_added,
|
||||
processed=photo.processed,
|
||||
person_name=full_name,
|
||||
tags=tags,
|
||||
has_faces=face_count > 0,
|
||||
face_count=face_count,
|
||||
)
|
||||
)
|
||||
elif search_type == "date":
|
||||
if not date_from and not date_to:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="At least one of date_from or date_to is required",
|
||||
)
|
||||
df = date.fromisoformat(date_from) if date_from else None
|
||||
dt = date.fromisoformat(date_to) if date_to else None
|
||||
results, total = search_photos_by_date(db, df, dt, folder_path, page, page_size)
|
||||
for photo in results:
|
||||
tags = get_photo_tags(db, photo.id)
|
||||
face_count = get_photo_face_count(db, photo.id)
|
||||
person_name_val = get_photo_person(db, photo.id)
|
||||
# Convert datetime to date for date_added
|
||||
date_added = photo.date_added.date() if isinstance(photo.date_added, datetime) else photo.date_added
|
||||
items.append(
|
||||
PhotoSearchResult(
|
||||
id=photo.id,
|
||||
path=photo.path,
|
||||
filename=photo.filename,
|
||||
date_taken=photo.date_taken,
|
||||
date_added=date_added,
|
||||
processed=photo.processed,
|
||||
person_name=person_name_val,
|
||||
tags=tags,
|
||||
has_faces=face_count > 0,
|
||||
face_count=face_count,
|
||||
)
|
||||
)
|
||||
elif search_type == "tags":
|
||||
if not tag_names:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="tag_names is required for tag search",
|
||||
)
|
||||
tag_list = [t.strip() for t in tag_names.split(",") if t.strip()]
|
||||
if not tag_list:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="At least one tag name is required",
|
||||
)
|
||||
results, total = search_photos_by_tags(
|
||||
db, tag_list, match_all, folder_path, page, page_size
|
||||
)
|
||||
for photo in results:
|
||||
tags = get_photo_tags(db, photo.id)
|
||||
face_count = get_photo_face_count(db, photo.id)
|
||||
person_name_val = get_photo_person(db, photo.id)
|
||||
# Convert datetime to date for date_added
|
||||
date_added = photo.date_added.date() if isinstance(photo.date_added, datetime) else photo.date_added
|
||||
items.append(
|
||||
PhotoSearchResult(
|
||||
id=photo.id,
|
||||
path=photo.path,
|
||||
filename=photo.filename,
|
||||
date_taken=photo.date_taken,
|
||||
date_added=date_added,
|
||||
processed=photo.processed,
|
||||
person_name=person_name_val,
|
||||
tags=tags,
|
||||
has_faces=face_count > 0,
|
||||
face_count=face_count,
|
||||
)
|
||||
)
|
||||
elif search_type == "no_faces":
|
||||
results, total = get_photos_without_faces(db, folder_path, page, page_size)
|
||||
for photo in results:
|
||||
tags = get_photo_tags(db, photo.id)
|
||||
# Convert datetime to date for date_added
|
||||
date_added = photo.date_added.date() if isinstance(photo.date_added, datetime) else photo.date_added
|
||||
items.append(
|
||||
PhotoSearchResult(
|
||||
id=photo.id,
|
||||
path=photo.path,
|
||||
filename=photo.filename,
|
||||
date_taken=photo.date_taken,
|
||||
date_added=date_added,
|
||||
processed=photo.processed,
|
||||
person_name=None,
|
||||
tags=tags,
|
||||
has_faces=False,
|
||||
face_count=0,
|
||||
)
|
||||
)
|
||||
elif search_type == "no_tags":
|
||||
results, total = get_photos_without_tags(db, folder_path, page, page_size)
|
||||
for photo in results:
|
||||
face_count = get_photo_face_count(db, photo.id)
|
||||
person_name_val = get_photo_person(db, photo.id)
|
||||
# Convert datetime to date for date_added
|
||||
date_added = photo.date_added.date() if isinstance(photo.date_added, datetime) else photo.date_added
|
||||
items.append(
|
||||
PhotoSearchResult(
|
||||
id=photo.id,
|
||||
path=photo.path,
|
||||
filename=photo.filename,
|
||||
date_taken=photo.date_taken,
|
||||
date_added=date_added,
|
||||
processed=photo.processed,
|
||||
person_name=person_name_val,
|
||||
tags=[],
|
||||
has_faces=face_count > 0,
|
||||
face_count=face_count,
|
||||
)
|
||||
)
|
||||
else:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f"Invalid search_type: {search_type}",
|
||||
)
|
||||
|
||||
return SearchPhotosResponse(items=items, page=page, page_size=page_size, total=total)
|
||||
|
||||
|
||||
@router.post("/import", response_model=PhotoImportResponse)
|
||||
|
||||
+84
-16
@@ -2,28 +2,96 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi import APIRouter
|
||||
from typing import List
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from src.web.db.session import get_db
|
||||
from src.web.schemas.tags import (
|
||||
PhotoTagsRequest,
|
||||
PhotoTagsResponse,
|
||||
TagCreateRequest,
|
||||
TagResponse,
|
||||
TagsResponse,
|
||||
)
|
||||
from src.web.services.tag_service import (
|
||||
add_tags_to_photos,
|
||||
get_or_create_tag,
|
||||
list_tags,
|
||||
remove_tags_from_photos,
|
||||
)
|
||||
|
||||
router = APIRouter(prefix="/tags", tags=["tags"])
|
||||
|
||||
|
||||
@router.get("")
|
||||
def list_tags() -> dict:
|
||||
"""List tags - placeholder for Phase 3."""
|
||||
return {"message": "Tags endpoint - to be implemented in Phase 3"}
|
||||
@router.get("", response_model=TagsResponse)
|
||||
def get_tags(db: Session = Depends(get_db)) -> TagsResponse:
|
||||
"""List all tags."""
|
||||
tags = list_tags(db)
|
||||
items = [TagResponse.model_validate(t) for t in tags]
|
||||
return TagsResponse(items=items, total=len(items))
|
||||
|
||||
|
||||
@router.post("")
|
||||
def create_tag() -> dict:
|
||||
"""Create tag - placeholder for Phase 3."""
|
||||
return {"message": "Create tag endpoint - to be implemented in Phase 3"}
|
||||
@router.post("", response_model=TagResponse)
|
||||
def create_tag(
|
||||
request: TagCreateRequest, db: Session = Depends(get_db)
|
||||
) -> TagResponse:
|
||||
"""Create a new tag (or return existing if already exists)."""
|
||||
tag = get_or_create_tag(db, request.tag_name)
|
||||
db.commit()
|
||||
db.refresh(tag)
|
||||
return TagResponse.model_validate(tag)
|
||||
|
||||
|
||||
@router.post("/photos/{photo_id}/tags")
|
||||
def add_tags_to_photo(photo_id: int) -> dict:
|
||||
"""Add tags to photo - placeholder for Phase 3."""
|
||||
return {
|
||||
"message": f"Add tags to photo {photo_id} - to be implemented in Phase 3",
|
||||
"id": photo_id,
|
||||
}
|
||||
@router.post("/photos/add", response_model=PhotoTagsResponse)
|
||||
def add_tags_to_photos_endpoint(
|
||||
request: PhotoTagsRequest, db: Session = Depends(get_db)
|
||||
) -> PhotoTagsResponse:
|
||||
"""Add tags to multiple photos."""
|
||||
if not request.photo_ids:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST, detail="photo_ids is required"
|
||||
)
|
||||
if not request.tag_names:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST, detail="tag_names is required"
|
||||
)
|
||||
|
||||
photos_updated, tags_added = add_tags_to_photos(
|
||||
db, request.photo_ids, request.tag_names
|
||||
)
|
||||
|
||||
return PhotoTagsResponse(
|
||||
message=f"Added tags to {photos_updated} photos",
|
||||
photos_updated=photos_updated,
|
||||
tags_added=tags_added,
|
||||
tags_removed=0,
|
||||
)
|
||||
|
||||
|
||||
@router.post("/photos/remove", response_model=PhotoTagsResponse)
|
||||
def remove_tags_from_photos_endpoint(
|
||||
request: PhotoTagsRequest, db: Session = Depends(get_db)
|
||||
) -> PhotoTagsResponse:
|
||||
"""Remove tags from multiple photos."""
|
||||
if not request.photo_ids:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST, detail="photo_ids is required"
|
||||
)
|
||||
if not request.tag_names:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST, detail="tag_names is required"
|
||||
)
|
||||
|
||||
photos_updated, tags_removed = remove_tags_from_photos(
|
||||
db, request.photo_ids, request.tag_names
|
||||
)
|
||||
|
||||
return PhotoTagsResponse(
|
||||
message=f"Removed tags from {photos_updated} photos",
|
||||
photos_updated=photos_updated,
|
||||
tags_added=0,
|
||||
tags_removed=tags_removed,
|
||||
)
|
||||
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
"""Search schemas for Phase 5."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date
|
||||
from typing import List, Optional
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class SearchPhotosQuery(BaseModel):
|
||||
"""Query parameters for searching photos."""
|
||||
|
||||
person_ids: Optional[List[int]] = Field(None, description="Filter by person IDs")
|
||||
tag_ids: Optional[List[int]] = Field(None, description="Filter by tag IDs")
|
||||
date_from: Optional[date] = Field(None, description="Filter by date taken (from)")
|
||||
date_to: Optional[date] = Field(None, description="Filter by date taken (to)")
|
||||
min_quality: Optional[float] = Field(None, ge=0.0, le=1.0, description="Minimum face quality score")
|
||||
folder_path: Optional[str] = Field(None, description="Filter by folder path prefix")
|
||||
sort_by: str = Field("date_taken", description="Sort column: date_taken, date_added, filename, path")
|
||||
sort_dir: str = Field("desc", description="Sort direction: asc|desc")
|
||||
page: int = Field(1, ge=1, description="Page number")
|
||||
page_size: int = Field(50, ge=1, le=200, description="Page size")
|
||||
|
||||
|
||||
class PhotoSearchResult(BaseModel):
|
||||
"""Photo search result item."""
|
||||
|
||||
id: int
|
||||
path: str
|
||||
filename: str
|
||||
date_taken: Optional[date] = None
|
||||
date_added: date
|
||||
processed: bool
|
||||
person_name: Optional[str] = None # For name search
|
||||
tags: List[str] = Field(default_factory=list) # All tags for the photo
|
||||
has_faces: bool = False
|
||||
face_count: int = 0
|
||||
|
||||
|
||||
class SearchPhotosResponse(BaseModel):
|
||||
"""Response for photo search."""
|
||||
|
||||
items: List[PhotoSearchResult]
|
||||
page: int
|
||||
page_size: int
|
||||
total: int
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
"""Tag schemas for Phase 5."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
from typing import List, Optional
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class TagResponse(BaseModel):
|
||||
"""Tag response schema."""
|
||||
|
||||
id: int
|
||||
tag_name: str
|
||||
created_date: datetime
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
|
||||
class TagCreateRequest(BaseModel):
|
||||
"""Request to create a tag."""
|
||||
|
||||
tag_name: str = Field(..., description="Tag name")
|
||||
|
||||
|
||||
class TagsResponse(BaseModel):
|
||||
"""Response for listing tags."""
|
||||
|
||||
items: List[TagResponse]
|
||||
total: int
|
||||
|
||||
|
||||
class PhotoTagsRequest(BaseModel):
|
||||
"""Request to add/remove tags from photos."""
|
||||
|
||||
photo_ids: List[int] = Field(..., description="Photo IDs")
|
||||
tag_names: List[str] = Field(..., description="Tag names to add/remove")
|
||||
|
||||
|
||||
class PhotoTagsResponse(BaseModel):
|
||||
"""Response for photo tagging operations."""
|
||||
|
||||
message: str
|
||||
photos_updated: int
|
||||
tags_added: int
|
||||
tags_removed: int
|
||||
|
||||
@@ -0,0 +1,264 @@
|
||||
"""Search service for photos - matches desktop search functionality exactly."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date
|
||||
from typing import List, Optional, Tuple
|
||||
|
||||
from sqlalchemy import func, or_
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from src.web.db.models import Face, Photo, Person, PhotoTagLinkage, Tag
|
||||
|
||||
|
||||
def search_photos_by_name(
|
||||
db: Session,
|
||||
person_name: str,
|
||||
folder_path: Optional[str] = None,
|
||||
page: int = 1,
|
||||
page_size: int = 50,
|
||||
) -> Tuple[List[Tuple[Photo, str]], int]:
|
||||
"""Search photos by person name (partial, case-insensitive).
|
||||
|
||||
Matches desktop behavior exactly:
|
||||
- Searches first_name, last_name, middle_name, maiden_name
|
||||
- Returns (photo, full_name) tuples
|
||||
- Filters by folder_path if provided
|
||||
"""
|
||||
search_name = (person_name or "").strip().lower()
|
||||
if not search_name:
|
||||
return [], 0
|
||||
|
||||
# Find matching people
|
||||
matching_people = (
|
||||
db.query(Person)
|
||||
.filter(
|
||||
or_(
|
||||
func.lower(Person.first_name).contains(search_name),
|
||||
func.lower(Person.last_name).contains(search_name),
|
||||
func.lower(Person.middle_name).contains(search_name),
|
||||
func.lower(Person.maiden_name).contains(search_name),
|
||||
)
|
||||
)
|
||||
.all()
|
||||
)
|
||||
|
||||
if not matching_people:
|
||||
return [], 0
|
||||
|
||||
person_ids = [p.id for p in matching_people]
|
||||
|
||||
# Query photos with faces linked to matching people
|
||||
query = (
|
||||
db.query(Photo, Person)
|
||||
.join(Face, Photo.id == Face.photo_id)
|
||||
.join(Person, Face.person_id == Person.id)
|
||||
.filter(Face.person_id.in_(person_ids))
|
||||
.distinct()
|
||||
)
|
||||
|
||||
# Apply folder filter if provided
|
||||
if folder_path:
|
||||
folder_path = folder_path.strip()
|
||||
query = query.filter(Photo.path.startswith(folder_path))
|
||||
|
||||
# Total count
|
||||
total = query.count()
|
||||
|
||||
# Pagination
|
||||
results = query.order_by(Person.last_name, Person.first_name, Photo.path).offset((page - 1) * page_size).limit(page_size).all()
|
||||
|
||||
# Format results: (photo, full_name)
|
||||
formatted = []
|
||||
for photo, person in results:
|
||||
full_name = f"{person.first_name or ''} {person.last_name or ''}".strip() or "Unknown"
|
||||
formatted.append((photo, full_name))
|
||||
|
||||
return formatted, total
|
||||
|
||||
|
||||
def search_photos_by_date(
|
||||
db: Session,
|
||||
date_from: Optional[date] = None,
|
||||
date_to: Optional[date] = None,
|
||||
folder_path: Optional[str] = None,
|
||||
page: int = 1,
|
||||
page_size: int = 50,
|
||||
) -> Tuple[List[Photo], int]:
|
||||
"""Search photos by date range.
|
||||
|
||||
Matches desktop behavior exactly:
|
||||
- Filters by date_taken
|
||||
- Requires at least one date
|
||||
- Returns photos ordered by date_taken DESC
|
||||
"""
|
||||
query = db.query(Photo).filter(Photo.date_taken.is_not(None))
|
||||
|
||||
if date_from:
|
||||
query = query.filter(Photo.date_taken >= date_from)
|
||||
if date_to:
|
||||
query = query.filter(Photo.date_taken <= date_to)
|
||||
|
||||
# Apply folder filter if provided
|
||||
if folder_path:
|
||||
folder_path = folder_path.strip()
|
||||
query = query.filter(Photo.path.startswith(folder_path))
|
||||
|
||||
# Total count
|
||||
total = query.count()
|
||||
|
||||
# Pagination and sorting
|
||||
results = query.order_by(Photo.date_taken.desc().nullslast(), Photo.filename).offset((page - 1) * page_size).limit(page_size).all()
|
||||
|
||||
return results, total
|
||||
|
||||
|
||||
def search_photos_by_tags(
|
||||
db: Session,
|
||||
tag_names: List[str],
|
||||
match_all: bool = False,
|
||||
folder_path: Optional[str] = None,
|
||||
page: int = 1,
|
||||
page_size: int = 50,
|
||||
) -> Tuple[List[Photo], int]:
|
||||
"""Search photos by tags.
|
||||
|
||||
Matches desktop behavior exactly:
|
||||
- match_all=True: photos must have ALL tags
|
||||
- match_all=False: photos with ANY tag
|
||||
- Case-insensitive tag matching
|
||||
"""
|
||||
if not tag_names:
|
||||
return [], 0
|
||||
|
||||
# Find tag IDs (case-insensitive)
|
||||
tag_ids = (
|
||||
db.query(Tag.id)
|
||||
.filter(func.lower(Tag.tag_name).in_([t.lower().strip() for t in tag_names]))
|
||||
.all()
|
||||
)
|
||||
tag_ids = [tid[0] for tid in tag_ids]
|
||||
|
||||
if not tag_ids:
|
||||
return [], 0
|
||||
|
||||
if match_all:
|
||||
# Photos that have ALL specified tags
|
||||
query = (
|
||||
db.query(Photo)
|
||||
.join(PhotoTagLinkage, Photo.id == PhotoTagLinkage.photo_id)
|
||||
.filter(PhotoTagLinkage.tag_id.in_(tag_ids))
|
||||
.group_by(Photo.id)
|
||||
.having(func.count(func.distinct(PhotoTagLinkage.tag_id)) == len(tag_ids))
|
||||
)
|
||||
else:
|
||||
# Photos that have ANY of the specified tags
|
||||
query = (
|
||||
db.query(Photo)
|
||||
.join(PhotoTagLinkage, Photo.id == PhotoTagLinkage.photo_id)
|
||||
.filter(PhotoTagLinkage.tag_id.in_(tag_ids))
|
||||
.distinct()
|
||||
)
|
||||
|
||||
# Apply folder filter if provided
|
||||
if folder_path:
|
||||
folder_path = folder_path.strip()
|
||||
query = query.filter(Photo.path.startswith(folder_path))
|
||||
|
||||
# Total count
|
||||
total = query.count()
|
||||
|
||||
# Pagination and sorting
|
||||
results = query.order_by(Photo.path).offset((page - 1) * page_size).limit(page_size).all()
|
||||
|
||||
return results, total
|
||||
|
||||
|
||||
def get_photos_without_faces(
|
||||
db: Session,
|
||||
folder_path: Optional[str] = None,
|
||||
page: int = 1,
|
||||
page_size: int = 50,
|
||||
) -> Tuple[List[Photo], int]:
|
||||
"""Get photos that have no detected faces.
|
||||
|
||||
Matches desktop behavior exactly.
|
||||
"""
|
||||
query = (
|
||||
db.query(Photo)
|
||||
.outerjoin(Face, Photo.id == Face.photo_id)
|
||||
.filter(Face.photo_id.is_(None))
|
||||
)
|
||||
|
||||
# Apply folder filter if provided
|
||||
if folder_path:
|
||||
folder_path = folder_path.strip()
|
||||
query = query.filter(Photo.path.startswith(folder_path))
|
||||
|
||||
# Total count
|
||||
total = query.count()
|
||||
|
||||
# Pagination and sorting
|
||||
results = query.order_by(Photo.filename).offset((page - 1) * page_size).limit(page_size).all()
|
||||
|
||||
return results, total
|
||||
|
||||
|
||||
def get_photos_without_tags(
|
||||
db: Session,
|
||||
folder_path: Optional[str] = None,
|
||||
page: int = 1,
|
||||
page_size: int = 50,
|
||||
) -> Tuple[List[Photo], int]:
|
||||
"""Get photos that have no tags.
|
||||
|
||||
Matches desktop behavior exactly.
|
||||
"""
|
||||
query = (
|
||||
db.query(Photo)
|
||||
.outerjoin(PhotoTagLinkage, Photo.id == PhotoTagLinkage.photo_id)
|
||||
.filter(PhotoTagLinkage.photo_id.is_(None))
|
||||
)
|
||||
|
||||
# Apply folder filter if provided
|
||||
if folder_path:
|
||||
folder_path = folder_path.strip()
|
||||
query = query.filter(Photo.path.startswith(folder_path))
|
||||
|
||||
# Total count
|
||||
total = query.count()
|
||||
|
||||
# Pagination and sorting
|
||||
results = query.order_by(Photo.filename).offset((page - 1) * page_size).limit(page_size).all()
|
||||
|
||||
return results, total
|
||||
|
||||
|
||||
def get_photo_tags(db: Session, photo_id: int) -> List[str]:
|
||||
"""Get all tags for a photo."""
|
||||
tags = (
|
||||
db.query(Tag.tag_name)
|
||||
.join(PhotoTagLinkage, Tag.id == PhotoTagLinkage.tag_id)
|
||||
.filter(PhotoTagLinkage.photo_id == photo_id)
|
||||
.all()
|
||||
)
|
||||
return [t[0] for t in tags]
|
||||
|
||||
|
||||
def get_photo_person(db: Session, photo_id: int) -> Optional[str]:
|
||||
"""Get person name for a photo (first face found)."""
|
||||
person = (
|
||||
db.query(Person)
|
||||
.join(Face, Person.id == Face.person_id)
|
||||
.filter(Face.photo_id == photo_id)
|
||||
.first()
|
||||
)
|
||||
if person:
|
||||
return f"{person.first_name or ''} {person.last_name or ''}".strip() or "Unknown"
|
||||
return None
|
||||
|
||||
|
||||
def get_photo_face_count(db: Session, photo_id: int) -> int:
|
||||
"""Get face count for a photo."""
|
||||
return db.query(Face).filter(Face.photo_id == photo_id).count()
|
||||
|
||||
@@ -0,0 +1,135 @@
|
||||
"""Tag service for managing tags - matches desktop functionality exactly."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import List, Optional
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from src.web.db.models import Photo, PhotoTagLinkage, Tag
|
||||
|
||||
|
||||
def list_tags(db: Session) -> List[Tag]:
|
||||
"""Get all tags."""
|
||||
return db.query(Tag).order_by(Tag.tag_name).all()
|
||||
|
||||
|
||||
def get_or_create_tag(db: Session, tag_name: str) -> Tag:
|
||||
"""Get existing tag or create new one (case-insensitive)."""
|
||||
# Check if tag exists (case-insensitive)
|
||||
existing = (
|
||||
db.query(Tag)
|
||||
.filter(Tag.tag_name.ilike(tag_name.strip()))
|
||||
.first()
|
||||
)
|
||||
if existing:
|
||||
return existing
|
||||
|
||||
# Create new tag
|
||||
tag = Tag(tag_name=tag_name.strip())
|
||||
db.add(tag)
|
||||
db.flush()
|
||||
return tag
|
||||
|
||||
|
||||
def add_tags_to_photos(
|
||||
db: Session, photo_ids: List[int], tag_names: List[str]
|
||||
) -> tuple[int, int]:
|
||||
"""Add tags to photos.
|
||||
|
||||
Returns:
|
||||
Tuple of (photos_updated, tags_added)
|
||||
"""
|
||||
photos_updated = 0
|
||||
tags_added = 0
|
||||
|
||||
# Deduplicate tag names (case-insensitive)
|
||||
seen_tags = set()
|
||||
unique_tags = []
|
||||
for tag_name in tag_names:
|
||||
normalized = tag_name.strip().lower()
|
||||
if normalized and normalized not in seen_tags:
|
||||
seen_tags.add(normalized)
|
||||
unique_tags.append(tag_name.strip())
|
||||
|
||||
if not unique_tags:
|
||||
return 0, 0
|
||||
|
||||
# Get or create tags
|
||||
tag_objs = []
|
||||
for tag_name in unique_tags:
|
||||
tag = get_or_create_tag(db, tag_name)
|
||||
tag_objs.append(tag)
|
||||
|
||||
# Add tags to photos
|
||||
for photo_id in photo_ids:
|
||||
photo = db.query(Photo).filter(Photo.id == photo_id).first()
|
||||
if not photo:
|
||||
continue
|
||||
|
||||
for tag in tag_objs:
|
||||
# Check if linkage already exists
|
||||
existing = (
|
||||
db.query(PhotoTagLinkage)
|
||||
.filter(
|
||||
PhotoTagLinkage.photo_id == photo_id,
|
||||
PhotoTagLinkage.tag_id == tag.id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not existing:
|
||||
linkage = PhotoTagLinkage(photo_id=photo_id, tag_id=tag.id)
|
||||
db.add(linkage)
|
||||
tags_added += 1
|
||||
|
||||
photos_updated += 1
|
||||
|
||||
db.commit()
|
||||
return photos_updated, tags_added
|
||||
|
||||
|
||||
def remove_tags_from_photos(
|
||||
db: Session, photo_ids: List[int], tag_names: List[str]
|
||||
) -> tuple[int, int]:
|
||||
"""Remove tags from photos.
|
||||
|
||||
Returns:
|
||||
Tuple of (photos_updated, tags_removed)
|
||||
"""
|
||||
photos_updated = 0
|
||||
tags_removed = 0
|
||||
|
||||
# Find tag IDs (case-insensitive)
|
||||
tag_ids = []
|
||||
for tag_name in tag_names:
|
||||
tag = (
|
||||
db.query(Tag)
|
||||
.filter(Tag.tag_name.ilike(tag_name.strip()))
|
||||
.first()
|
||||
)
|
||||
if tag:
|
||||
tag_ids.append(tag.id)
|
||||
|
||||
if not tag_ids:
|
||||
return 0, 0
|
||||
|
||||
# Remove linkages
|
||||
for photo_id in photo_ids:
|
||||
linkages = (
|
||||
db.query(PhotoTagLinkage)
|
||||
.filter(
|
||||
PhotoTagLinkage.photo_id == photo_id,
|
||||
PhotoTagLinkage.tag_id.in_(tag_ids),
|
||||
)
|
||||
.all()
|
||||
)
|
||||
for linkage in linkages:
|
||||
db.delete(linkage)
|
||||
tags_removed += 1
|
||||
|
||||
if linkages:
|
||||
photos_updated += 1
|
||||
|
||||
db.commit()
|
||||
return photos_updated, tags_removed
|
||||
|
||||
Reference in New Issue
Block a user