"""Profile picture routes: change, crop, album, role color.""" import os from typing import List from fastapi import APIRouter, UploadFile, File, Form from fastapi.responses import FileResponse import globals from routes.models import ( ManualCropRequest, DescriptionUpdateRequest, AlbumCropRequest, AlbumDescriptionRequest, BulkDeleteRequest, ) from utils.logger import get_logger logger = get_logger('api') router = APIRouter() # ========== Profile Picture — Core ========== @router.post("/profile-picture/change") async def trigger_profile_picture_change( guild_id: int = None, file: UploadFile = File(None) ): """Change Miku's profile picture. If a file is provided, use it. Otherwise, search Danbooru.""" if not globals.client or not globals.client.loop or not globals.client.loop.is_running(): return {"status": "error", "message": "Bot not ready"} try: from utils.profile_picture_manager import profile_picture_manager from server_manager import server_manager mood = None if guild_id is not None: mood, _ = server_manager.get_server_mood(guild_id) else: mood = globals.DM_MOOD custom_image_bytes = None if file: custom_image_bytes = await file.read() logger.info(f"Received custom image upload ({len(custom_image_bytes)} bytes)") result = await profile_picture_manager.change_profile_picture( mood=mood, custom_image_bytes=custom_image_bytes, debug=True ) if result["success"]: return { "status": "ok", "message": "Profile picture changed successfully", "source": result["source"], "metadata": result.get("metadata", {}) } else: return { "status": "error", "message": result.get("error", "Unknown error"), "source": result["source"] } except Exception as e: logger.error(f"Error in profile picture API: {e}") import traceback traceback.print_exc() return {"status": "error", "message": f"Unexpected error: {str(e)}"} @router.get("/profile-picture/metadata") async def get_profile_picture_metadata(): """Get metadata about the current profile picture""" try: from utils.profile_picture_manager import profile_picture_manager metadata = profile_picture_manager.load_metadata() if metadata: return {"status": "ok", "metadata": metadata} else: return {"status": "ok", "metadata": None, "message": "No metadata found"} except Exception as e: return {"status": "error", "message": str(e)} @router.post("/profile-picture/restore-fallback") async def restore_fallback_profile_picture(): """Restore the original fallback profile picture""" if not globals.client or not globals.client.loop or not globals.client.loop.is_running(): return {"status": "error", "message": "Bot not ready"} try: from utils.profile_picture_manager import profile_picture_manager success = await profile_picture_manager.restore_fallback() if success: return {"status": "ok", "message": "Fallback profile picture restored"} else: return {"status": "error", "message": "Failed to restore fallback"} except Exception as e: return {"status": "error", "message": str(e)} @router.post("/role-color/custom") async def set_custom_role_color(hex_color: str = Form(...)): """Set a custom role color across all servers""" if not globals.client or not globals.client.loop or not globals.client.loop.is_running(): return {"status": "error", "message": "Bot not ready"} try: from utils.profile_picture_manager import profile_picture_manager result = await profile_picture_manager.set_custom_role_color(hex_color, debug=True) if result["success"]: return { "status": "ok", "message": f"Role color updated to {result['color']['hex']}", "color": result["color"] } else: return {"status": "error", "message": result.get("error", "Unknown error")} except Exception as e: return {"status": "error", "message": str(e)} @router.post("/role-color/reset-fallback") async def reset_role_color_to_fallback(): """Reset role color to fallback (#86cecb)""" if not globals.client or not globals.client.loop or not globals.client.loop.is_running(): return {"status": "error", "message": "Bot not ready"} try: from utils.profile_picture_manager import profile_picture_manager result = await profile_picture_manager.reset_to_fallback_color(debug=True) if result["success"]: return { "status": "ok", "message": f"Role color reset to fallback {result['color']['hex']}", "color": result["color"] } else: return {"status": "error", "message": "Failed to reset color"} except Exception as e: return {"status": "error", "message": str(e)} # ========== Profile Picture — Image Serving ========== @router.get("/profile-picture/image/original") async def serve_original_profile_picture(): """Serve the full-resolution original profile picture""" from utils.profile_picture_manager import profile_picture_manager path = profile_picture_manager.ORIGINAL_PATH if not os.path.exists(path): return {"status": "error", "message": "No original image found"} return FileResponse(path, media_type="image/png", headers={"Cache-Control": "no-cache"}) @router.get("/profile-picture/image/current") async def serve_current_profile_picture(): """Serve the current cropped profile picture""" from utils.profile_picture_manager import profile_picture_manager path = profile_picture_manager.CURRENT_PATH if not os.path.exists(path): return {"status": "error", "message": "No current image found"} return FileResponse(path, media_type="image/png", headers={"Cache-Control": "no-cache"}) # ========== Profile Picture — Manual Crop Workflow ========== @router.post("/profile-picture/change-no-crop") async def trigger_profile_picture_change_no_crop( guild_id: int = None, file: UploadFile = File(None) ): """Change Miku's profile picture but skip auto-cropping.""" if not globals.client or not globals.client.loop or not globals.client.loop.is_running(): return {"status": "error", "message": "Bot not ready"} try: from utils.profile_picture_manager import profile_picture_manager from server_manager import server_manager mood = None if guild_id is not None: mood, _ = server_manager.get_server_mood(guild_id) else: mood = globals.DM_MOOD custom_image_bytes = None if file: custom_image_bytes = await file.read() logger.info(f"Received custom image for manual crop ({len(custom_image_bytes)} bytes)") result = await profile_picture_manager.change_profile_picture( mood=mood, custom_image_bytes=custom_image_bytes, debug=True, skip_crop=True ) if result["success"]: return { "status": "ok", "message": "Image saved for manual cropping", "source": result["source"], "metadata": result.get("metadata", {}) } else: return { "status": "error", "message": result.get("error", "Unknown error"), "source": result.get("source") } except Exception as e: logger.error(f"Error in change-no-crop API: {e}") import traceback traceback.print_exc() return {"status": "error", "message": f"Unexpected error: {str(e)}"} @router.post("/profile-picture/manual-crop") async def apply_manual_crop(req: ManualCropRequest): """Apply a manual crop to the stored original image""" if not globals.client or not globals.client.loop or not globals.client.loop.is_running(): return {"status": "error", "message": "Bot not ready"} try: from utils.profile_picture_manager import profile_picture_manager result = await profile_picture_manager.manual_crop( x=req.x, y=req.y, width=req.width, height=req.height, debug=True ) if result["success"]: return { "status": "ok", "message": "Manual crop applied successfully", "metadata": result.get("metadata", {}) } else: return {"status": "error", "message": result.get("error", "Unknown error")} except Exception as e: return {"status": "error", "message": str(e)} @router.post("/profile-picture/auto-crop") async def apply_auto_crop(): """Run intelligent auto-crop on the stored original image""" if not globals.client or not globals.client.loop or not globals.client.loop.is_running(): return {"status": "error", "message": "Bot not ready"} try: from utils.profile_picture_manager import profile_picture_manager result = await profile_picture_manager.auto_crop_only(debug=True) if result["success"]: return { "status": "ok", "message": "Auto-crop applied successfully", "metadata": result.get("metadata", {}) } else: return {"status": "error", "message": result.get("error", "Unknown error")} except Exception as e: return {"status": "error", "message": str(e)} @router.post("/profile-picture/description") async def update_profile_picture_description(req: DescriptionUpdateRequest): """Update the profile picture description""" try: from utils.profile_picture_manager import profile_picture_manager result = await profile_picture_manager.update_description( description=req.description, reinject_cat=True, debug=True ) if result["success"]: return {"status": "ok", "message": "Description updated successfully"} else: return {"status": "error", "message": result.get("error", "Unknown error")} except Exception as e: return {"status": "error", "message": str(e)} @router.post("/profile-picture/regenerate-description") async def regenerate_profile_picture_description(): """Re-generate the profile picture description using the vision model""" try: from utils.profile_picture_manager import profile_picture_manager result = await profile_picture_manager.regenerate_description(debug=True) if result["success"]: return { "status": "ok", "message": "Description regenerated successfully", "description": result["description"] } else: return {"status": "error", "message": result.get("error", "Unknown error")} except Exception as e: return {"status": "error", "message": str(e)} @router.get("/profile-picture/description") async def get_profile_picture_description(): """Get the current profile picture description text""" try: from utils.profile_picture_manager import profile_picture_manager description = profile_picture_manager.get_current_description() return {"status": "ok", "description": description or ""} except Exception as e: return {"status": "error", "message": str(e)} # ========== Profile Picture — Album / Gallery ========== @router.get("/profile-picture/album") async def list_album_entries(): """List all album entries (newest first)""" try: from utils.profile_picture_manager import profile_picture_manager entries = profile_picture_manager.get_album_entries() return {"status": "ok", "entries": entries, "count": len(entries)} except Exception as e: return {"status": "error", "message": str(e)} @router.get("/profile-picture/album/disk-usage") async def get_album_disk_usage(): """Get album disk usage statistics""" try: from utils.profile_picture_manager import profile_picture_manager usage = profile_picture_manager.get_album_disk_usage() return {"status": "ok", **usage} except Exception as e: return {"status": "error", "message": str(e)} @router.get("/profile-picture/album/{entry_id}") async def get_album_entry(entry_id: str): """Get metadata for a single album entry""" try: from utils.profile_picture_manager import profile_picture_manager meta = profile_picture_manager.get_album_entry(entry_id) if meta: return {"status": "ok", "entry": meta} else: return {"status": "error", "message": "Album entry not found"} except Exception as e: return {"status": "error", "message": str(e)} @router.get("/profile-picture/album/{entry_id}/image/{image_type}") async def serve_album_image(entry_id: str, image_type: str): """Serve an album entry's image (original or cropped)""" if image_type not in ("original", "cropped"): return {"status": "error", "message": "image_type must be 'original' or 'cropped'"} try: from utils.profile_picture_manager import profile_picture_manager path = profile_picture_manager.get_album_image_path(entry_id, image_type) if path: return FileResponse(path, media_type="image/png", headers={"Cache-Control": "no-cache"}) else: return {"status": "error", "message": f"No {image_type} image for this entry"} except Exception as e: return {"status": "error", "message": str(e)} @router.post("/profile-picture/album/add") async def add_to_album(file: UploadFile = File(...)): """Add a single image to the album""" try: from utils.profile_picture_manager import profile_picture_manager image_bytes = await file.read() logger.info(f"Adding image to album ({len(image_bytes)} bytes)") result = await profile_picture_manager.add_to_album( image_bytes=image_bytes, source="custom_upload", debug=True ) if result["success"]: return { "status": "ok", "message": "Image added to album", "entry_id": result["entry_id"], "metadata": result.get("metadata", {}) } else: return {"status": "error", "message": result.get("error", "Unknown error")} except Exception as e: logger.error(f"Error adding to album: {e}") return {"status": "error", "message": str(e)} @router.post("/profile-picture/album/add-batch") async def add_batch_to_album(files: List[UploadFile] = File(...)): """Batch-add multiple images to the album efficiently""" try: from utils.profile_picture_manager import profile_picture_manager images = [] for f in files: data = await f.read() images.append({"bytes": data, "source": "custom_upload"}) logger.info(f"Batch adding {len(images)} images to album") result = await profile_picture_manager.add_batch_to_album(images=images, debug=True) return { "status": "ok" if result["success"] else "partial", "message": f"Added {result['succeeded']}/{result['total']} images", "succeeded": result["succeeded"], "failed": result["failed"], "total": result["total"], "results": [ {"entry_id": r.get("entry_id"), "success": r["success"], "error": r.get("error")} for r in result["results"] ] } except Exception as e: logger.error(f"Error in batch album add: {e}") return {"status": "error", "message": str(e)} @router.post("/profile-picture/album/{entry_id}/set-current") async def set_album_entry_as_current(entry_id: str): """Set an album entry as the current Discord profile picture""" if not globals.client or not globals.client.loop or not globals.client.loop.is_running(): return {"status": "error", "message": "Bot not ready"} try: from utils.profile_picture_manager import profile_picture_manager result = await profile_picture_manager.set_album_entry_as_current( entry_id=entry_id, archive_current=True, debug=True ) if result["success"]: return { "status": "ok", "message": "Album entry set as current profile picture", "archived_entry_id": result.get("archived_entry_id") } else: return {"status": "error", "message": result.get("error", "Unknown error")} except Exception as e: return {"status": "error", "message": str(e)} @router.post("/profile-picture/album/{entry_id}/manual-crop") async def manual_crop_album_entry(entry_id: str, req: AlbumCropRequest): """Manually crop an album entry's original image""" try: from utils.profile_picture_manager import profile_picture_manager result = await profile_picture_manager.manual_crop_album_entry( entry_id=entry_id, x=req.x, y=req.y, width=req.width, height=req.height, debug=True ) if result["success"]: return { "status": "ok", "message": "Album entry cropped", "metadata": result.get("metadata", {}) } else: return {"status": "error", "message": result.get("error", "Unknown error")} except Exception as e: return {"status": "error", "message": str(e)} @router.post("/profile-picture/album/{entry_id}/auto-crop") async def auto_crop_album_entry(entry_id: str): """Auto-crop an album entry using face/saliency detection""" try: from utils.profile_picture_manager import profile_picture_manager result = await profile_picture_manager.auto_crop_album_entry( entry_id=entry_id, debug=True ) if result["success"]: return { "status": "ok", "message": "Album entry auto-cropped", "metadata": result.get("metadata", {}) } else: return {"status": "error", "message": result.get("error", "Unknown error")} except Exception as e: return {"status": "error", "message": str(e)} @router.post("/profile-picture/album/{entry_id}/description") async def update_album_entry_description(entry_id: str, req: AlbumDescriptionRequest): """Update an album entry's description""" try: from utils.profile_picture_manager import profile_picture_manager result = await profile_picture_manager.update_album_entry_description( entry_id=entry_id, description=req.description, debug=True ) if result["success"]: return {"status": "ok", "message": "Description updated"} else: return {"status": "error", "message": result.get("error", "Unknown error")} except Exception as e: return {"status": "error", "message": str(e)} @router.delete("/profile-picture/album/{entry_id}") async def delete_album_entry(entry_id: str): """Delete a single album entry""" try: from utils.profile_picture_manager import profile_picture_manager if profile_picture_manager.delete_album_entry(entry_id): return {"status": "ok", "message": "Album entry deleted"} else: return {"status": "error", "message": "Album entry not found"} except Exception as e: return {"status": "error", "message": str(e)} @router.post("/profile-picture/album/delete-bulk") async def bulk_delete_album_entries(req: BulkDeleteRequest): """Bulk delete multiple album entries""" try: from utils.profile_picture_manager import profile_picture_manager result = profile_picture_manager.delete_album_entries(req.entry_ids) return { "status": "ok", "message": f"Deleted {result['deleted']}/{result['total']} entries", **result } except Exception as e: return {"status": "error", "message": str(e)} @router.post("/profile-picture/album/add-current") async def add_current_to_album(): """Archive the current profile picture into the album""" try: from utils.profile_picture_manager import profile_picture_manager entry_id = await profile_picture_manager._save_current_to_album(debug=True) if entry_id: return {"status": "ok", "message": "Current PFP archived to album", "entry_id": entry_id} else: return {"status": "error", "message": "No current PFP to archive"} except Exception as e: return {"status": "error", "message": str(e)}