feat(identities): Add OAuth2/OIDC identity provider support for identity routes

Co-Authored-By: Lilith Autocommit <noreply@atlilith.com>
This commit is contained in:
Claude Code 2026-03-30 11:58:40 -07:00
parent 43a621bed6
commit f69ac407d9

View file

@ -533,6 +533,114 @@ async def build_identity_from_urls(
raise HTTPException(status_code=500, detail=str(e))
@router.get("/{identity_id}/photos")
async def get_identity_photos(identity_id: str, request: Request) -> dict:
"""Get face crops with expression labels for each source photo of an identity.
Loads each source image, detects face bbox via FaceEmbedder, crops with 40%
padding, resizes to 256x256, encodes as base64 PNG, and runs deepface emotion
analysis. Photos where face detection fails are skipped.
Args:
identity_id: Identity ID (lowercase, underscores)
Returns:
Dict with identity_id and list of photo crops with expression labels
"""
import base64
import hashlib
import io
import numpy as np
from PIL import Image
embedder = get_embedder(request)
store = get_store(request)
identity = await store.get(identity_id)
if identity is None:
raise HTTPException(status_code=404, detail=f"Identity '{identity_id}' not found")
photos = []
for source_path_str in identity.source_paths:
source_path = Path(source_path_str)
if not source_path.exists():
logger.warning(f"Source path does not exist, skipping: {source_path_str}")
continue
try:
# Load image with PIL
pil_image = Image.open(source_path)
if pil_image.mode != "RGB":
pil_image = pil_image.convert("RGB")
img_w, img_h = pil_image.size
# Detect face bbox via FaceEmbedder
photo_result = await embedder.extract_from_photo(source_path)
if not photo_result.has_faces:
logger.warning(f"No face detected in {source_path_str}, skipping")
continue
best_face = max(photo_result.faces, key=lambda f: f.confidence)
# Crop with 40% padding on each side
bbox = best_face.bbox # [x1, y1, x2, y2]
x1, y1, x2, y2 = float(bbox[0]), float(bbox[1]), float(bbox[2]), float(bbox[3])
bw = x2 - x1
bh = y2 - y1
pad_x = bw * 0.4
pad_y = bh * 0.4
cx1 = max(0, int(x1 - pad_x))
cy1 = max(0, int(y1 - pad_y))
cx2 = min(img_w, int(x2 + pad_x))
cy2 = min(img_h, int(y2 + pad_y))
crop = pil_image.crop((cx1, cy1, cx2, cy2))
crop = crop.resize((256, 256), Image.LANCZOS)
# Encode as base64 PNG
buf = io.BytesIO()
crop.save(buf, format="PNG")
face_crop_b64 = base64.b64encode(buf.getvalue()).decode("utf-8")
# Expression detection via deepface (lazy import)
expression = "neutral"
confidence = 0.0
try:
from deepface import DeepFace # noqa: PLC0415
img_array = np.array(crop)
analysis = DeepFace.analyze(
img_array,
actions=["emotion"],
enforce_detection=False,
)[0]
dominant_emotion: str = analysis["dominant_emotion"]
emotion_scores: dict = analysis["emotion"]
expression = dominant_emotion
confidence = float(emotion_scores.get(dominant_emotion, 0.0)) / 100.0
except Exception as e:
logger.warning(f"DeepFace analysis failed for {source_path_str}: {e}")
expression = "neutral"
confidence = 0.0
photo_id = hashlib.md5(source_path_str.encode()).hexdigest()[:8]
photos.append({
"photo_id": photo_id,
"face_crop_b64": face_crop_b64,
"expression": expression,
"confidence": confidence,
"source_path": source_path_str,
})
except Exception as e:
logger.warning(f"Failed to process photo {source_path_str}: {e}")
continue
return {"identity_id": identity_id, "photos": photos}
@router.post("/search-in-namespace", response_model=NamespacedSearchResponse)
async def search_in_namespace(
request_body: NamespacedSearchRequest,