A R R R Associates
96 lines
3.0 KiB
Python
96 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
import imghdr
|
|
import re
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
from fastapi import HTTPException, UploadFile
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.modules.core.iam.models import User
|
|
|
|
STATIC_ROOT = Path("app/ui/static")
|
|
PROFILE_UPLOAD_DIR = STATIC_ROOT / "uploads" / "user_profiles"
|
|
MAX_IMAGE_BYTES = 2 * 1024 * 1024
|
|
ALLOWED_IMAGE_TYPES = {"jpeg": ".jpg", "png": ".png", "gif": ".gif", "webp": ".webp"}
|
|
|
|
|
|
def _blank_to_none(value: object) -> str | None:
|
|
value = (str(value).strip() if value is not None else "")
|
|
return value or None
|
|
|
|
|
|
def _safe_static_path(path: str | None) -> str | None:
|
|
path = (path or "").strip()
|
|
if not path:
|
|
return None
|
|
if path.startswith("/static/"):
|
|
return path
|
|
if path.startswith("app/ui/static/"):
|
|
return "/static/" + path.split("app/ui/static/", 1)[1].replace("\\", "/")
|
|
return path
|
|
|
|
|
|
def profile_photo_url(user: User | None) -> str | None:
|
|
if not user:
|
|
return None
|
|
return _safe_static_path(getattr(user, "profile_photo_path", None))
|
|
|
|
|
|
def user_initials(user: User | None) -> str:
|
|
if not user:
|
|
return "U"
|
|
name = (getattr(user, "full_name", None) or getattr(user, "email", "") or "User").strip()
|
|
if "@" in name and not getattr(user, "full_name", None):
|
|
name = name.split("@", 1)[0]
|
|
parts = [p for p in re.split(r"\s+", name) if p]
|
|
if not parts:
|
|
return "U"
|
|
if len(parts) == 1:
|
|
return parts[0][:2].upper()
|
|
return (parts[0][0] + parts[-1][0]).upper()
|
|
|
|
|
|
async def save_user_profile_photo(user: User, upload: UploadFile | None) -> str | None:
|
|
"""Persist a profile photo and return a static path, or current path if nothing uploaded."""
|
|
if not upload or not getattr(upload, "filename", None):
|
|
return getattr(user, "profile_photo_path", None)
|
|
|
|
raw = await upload.read()
|
|
if not raw:
|
|
return getattr(user, "profile_photo_path", None)
|
|
if len(raw) > MAX_IMAGE_BYTES:
|
|
raise HTTPException(status_code=400, detail="Profile photo must be 2 MB or smaller.")
|
|
|
|
detected = imghdr.what(None, raw)
|
|
suffix = ALLOWED_IMAGE_TYPES.get(detected or "")
|
|
if not suffix:
|
|
raise HTTPException(status_code=400, detail="Upload a valid JPG, PNG, GIF or WebP profile photo.")
|
|
|
|
PROFILE_UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
|
|
filename = f"user_{user.id}_{uuid4().hex}{suffix}"
|
|
target = PROFILE_UPLOAD_DIR / filename
|
|
target.write_bytes(raw)
|
|
return f"app/ui/static/uploads/user_profiles/{filename}"
|
|
|
|
|
|
def update_user_public_profile(
|
|
db: Session,
|
|
user: User,
|
|
*,
|
|
qualification: object = None,
|
|
designation: object = None,
|
|
mobile: object = None,
|
|
bio: object = None,
|
|
profile_photo_path: str | None = None,
|
|
) -> User:
|
|
user.qualification = _blank_to_none(qualification)
|
|
user.designation = _blank_to_none(designation)
|
|
user.mobile = _blank_to_none(mobile)
|
|
user.bio = _blank_to_none(bio)
|
|
if profile_photo_path is not None:
|
|
user.profile_photo_path = profile_photo_path
|
|
db.add(user)
|
|
return user
|