""" Admin router — user management and app configuration. All endpoints require admin role. """ import os import logging from typing import Optional, List from fastapi import APIRouter, Depends, HTTPException, status from pydantic import BaseModel from sqlalchemy import select, func as sa_func from sqlalchemy.ext.asyncio import AsyncSession from app.auth import hash_password from app.database import get_db from app.dependencies import require_admin from app.models.user import User from app.models.photos import Photo from app.models.folders import SourceRoot from app.config import settings from app.services.feature_flags import ( ALL_FLAGS, snapshot as flags_snapshot, set_flag, reset_flag, is_enabled, FLAG_VISION_ENABLED, ) logger = logging.getLogger(__name__) router = APIRouter() # --------------------------------------------------------------------------- # Schemas # --------------------------------------------------------------------------- class CreateUserRequest(BaseModel): username: str password: str role: str = "user" # 'admin' | 'user' class UpdateUserRequest(BaseModel): role: Optional[str] = None is_active: Optional[bool] = None new_password: Optional[str] = None class UserDetailResponse(BaseModel): id: str username: str email: Optional[str] role: str is_active: bool media_path: str created_at: Optional[str] photo_count: int = 0 class UserListResponse(BaseModel): users: List[UserDetailResponse] total: int # --------------------------------------------------------------------------- # User CRUD # --------------------------------------------------------------------------- @router.get("/users", response_model=UserListResponse) async def list_users( admin: User = Depends(require_admin), db: AsyncSession = Depends(get_db), ): """List all users with their photo counts.""" result = await db.execute(select(User).order_by(User.created_at)) users = result.scalars().all() user_list = [] for u in users: count_result = await db.execute( select(sa_func.count(Photo.id)).where(Photo.user_id == u.id) ) photo_count = count_result.scalar() or 0 user_list.append(UserDetailResponse( id=u.id, username=u.username, email=u.email, role=u.role, is_active=u.is_active, media_path=u.media_path, created_at=u.created_at.isoformat() if u.created_at else None, photo_count=photo_count, )) return UserListResponse(users=user_list, total=len(user_list)) @router.post("/users", status_code=201, response_model=UserDetailResponse) async def create_user( body: CreateUserRequest, admin: User = Depends(require_admin), db: AsyncSession = Depends(get_db), ): """Create a new user. Creates their media directory and source root.""" if body.role not in ("admin", "user"): raise HTTPException(status_code=400, detail="Role must be 'admin' or 'user'") if len(body.username.strip()) < 2: raise HTTPException(status_code=400, detail="Username must be at least 2 characters") if len(body.password) < 6: raise HTTPException(status_code=400, detail="Password must be at least 6 characters") # Check for duplicate username existing = await db.execute( select(User).where(User.username == body.username.strip()) ) if existing.scalar_one_or_none() is not None: raise HTTPException(status_code=409, detail="Username already taken") media_path = os.path.join(settings.photo_dirs, body.username.strip()) os.makedirs(media_path, exist_ok=True) user = User( username=body.username.strip(), hashed_password=hash_password(body.password), role=body.role, media_path=media_path, ) db.add(user) await db.flush() # get user.id before creating source root source_root = SourceRoot( name=f"{user.username}'s Library", path=media_path, user_id=user.id, ) db.add(source_root) await db.commit() logger.info(f"Admin '{admin.username}' created user '{user.username}' (role={user.role})") return UserDetailResponse( id=user.id, username=user.username, email=user.email, role=user.role, is_active=user.is_active, media_path=user.media_path, created_at=user.created_at.isoformat() if user.created_at else None, photo_count=0, ) @router.get("/users/{user_id}", response_model=UserDetailResponse) async def get_user( user_id: str, admin: User = Depends(require_admin), db: AsyncSession = Depends(get_db), ): """Get a single user's details.""" result = await db.execute(select(User).where(User.id == user_id)) user = result.scalar_one_or_none() if user is None: raise HTTPException(status_code=404, detail="User not found") count_result = await db.execute( select(sa_func.count(Photo.id)).where(Photo.user_id == user.id) ) photo_count = count_result.scalar() or 0 return UserDetailResponse( id=user.id, username=user.username, email=user.email, role=user.role, is_active=user.is_active, media_path=user.media_path, created_at=user.created_at.isoformat() if user.created_at else None, photo_count=photo_count, ) @router.patch("/users/{user_id}", response_model=UserDetailResponse) async def update_user( user_id: str, body: UpdateUserRequest, admin: User = Depends(require_admin), db: AsyncSession = Depends(get_db), ): """Update a user's role, active status, or password.""" result = await db.execute(select(User).where(User.id == user_id)) user = result.scalar_one_or_none() if user is None: raise HTTPException(status_code=404, detail="User not found") if body.role is not None: if body.role not in ("admin", "user"): raise HTTPException(status_code=400, detail="Role must be 'admin' or 'user'") # Prevent demoting the last admin if user.role == "admin" and body.role == "user": admin_count = (await db.execute( select(sa_func.count(User.id)).where(User.role == "admin", User.is_active == True) )).scalar() if admin_count <= 1: raise HTTPException(status_code=400, detail="Cannot demote the last admin") user.role = body.role if body.is_active is not None: # Prevent deactivating the last admin if user.role == "admin" and not body.is_active: admin_count = (await db.execute( select(sa_func.count(User.id)).where(User.role == "admin", User.is_active == True) )).scalar() if admin_count <= 1: raise HTTPException(status_code=400, detail="Cannot deactivate the last admin") user.is_active = body.is_active if body.new_password is not None: if len(body.new_password) < 6: raise HTTPException(status_code=400, detail="Password must be at least 6 characters") user.hashed_password = hash_password(body.new_password) await db.commit() count_result = await db.execute( select(sa_func.count(Photo.id)).where(Photo.user_id == user.id) ) photo_count = count_result.scalar() or 0 return UserDetailResponse( id=user.id, username=user.username, email=user.email, role=user.role, is_active=user.is_active, media_path=user.media_path, created_at=user.created_at.isoformat() if user.created_at else None, photo_count=photo_count, ) @router.delete("/users/{user_id}") async def delete_user( user_id: str, admin: User = Depends(require_admin), db: AsyncSession = Depends(get_db), ): """Soft-delete a user by deactivating them. Media is preserved.""" result = await db.execute(select(User).where(User.id == user_id)) user = result.scalar_one_or_none() if user is None: raise HTTPException(status_code=404, detail="User not found") if user.id == admin.id: raise HTTPException(status_code=400, detail="Cannot delete yourself") # Prevent deleting the last admin if user.role == "admin": admin_count = (await db.execute( select(sa_func.count(User.id)).where(User.role == "admin", User.is_active == True) )).scalar() if admin_count <= 1: raise HTTPException(status_code=400, detail="Cannot delete the last admin") user.is_active = False await db.commit() logger.info(f"Admin '{admin.username}' deactivated user '{user.username}'") return {"status": "ok", "detail": f"User '{user.username}' deactivated"} # --------------------------------------------------------------------------- # AI / vision feature flags + manual triggers # --------------------------------------------------------------------------- class FeatureFlagUpdate(BaseModel): """PATCH body for toggling a feature flag. ``value`` sets an explicit override (true/false); omitting it clears the override and reverts the flag to its YAML default. """ value: Optional[bool] = None @router.get("/feature-flags") async def get_feature_flags(admin: User = Depends(require_admin)): """Return every tunable feature flag with its current effective value, YAML default, and whether an admin override is in effect.""" return {"flags": flags_snapshot()} @router.patch("/feature-flags/{flag_name}") async def update_feature_flag( flag_name: str, body: FeatureFlagUpdate, admin: User = Depends(require_admin), ): """Set or clear an override for one flag. With ``value`` set, the flag is pinned to that boolean; without it, the override is deleted and the YAML default takes over again. New value is observed by vision tasks on their next invocation — there's no worker restart required. """ if flag_name not in ALL_FLAGS: raise HTTPException(status_code=404, detail=f"Unknown flag: {flag_name}") try: if body.value is None: reset_flag(flag_name) action = "cleared override" else: set_flag(flag_name, bool(body.value)) action = f"set to {body.value}" except RuntimeError as e: # Redis unreachable — surface as 503 so the UI doesn't think it # succeeded silently. raise HTTPException(status_code=503, detail=str(e)) logger.info(f"Admin '{admin.username}' {action} for flag '{flag_name}'") return {"flags": flags_snapshot()} class BackfillVisionBody(BaseModel): """POST body for triggering a classifier backfill. ``limit`` caps how many photos are queued.""" limit: Optional[int] = None @router.post("/ai/backfill") async def trigger_ai_backfill( body: BackfillVisionBody, admin: User = Depends(require_admin), ): """Queue a classifier backfill pass.""" if not is_enabled(FLAG_VISION_ENABLED): raise HTTPException( status_code=400, detail="Vision is currently disabled; enable it before running a backfill.", ) if body.limit is not None and body.limit <= 0: raise HTTPException(status_code=400, detail="limit must be positive") from app.tasks.vision import backfill_vision result = backfill_vision.apply_async(kwargs={'limit': body.limit}) logger.info( f"Admin '{admin.username}' queued vision backfill " f"(limit={body.limit}, celery_id={result.id})" ) return { "status": "queued", "task_id": result.id, "limit": body.limit, } @router.post("/ai/rescan") async def trigger_full_rescan(admin: User = Depends(require_admin)): """Dispatch the same scan_all_source_roots job the backend runs at startup. Picks up any new files on disk and, through the post-scan hook, queues a vision backfill for whatever still lacks embeddings / OCR / etc. """ from app.tasks.scan import scan_all_source_roots result = scan_all_source_roots.apply_async() logger.info( f"Admin '{admin.username}' queued full rescan (celery_id={result.id})" ) return {"status": "queued", "task_id": result.id}