"""CRUD for the `admin` profile table (linked to users.id for login).""" from __future__ import annotations import uuid from pathlib import Path from fastapi import HTTPException, UploadFile, status from sqlalchemy import inspect, text from sqlalchemy.engine import Engine from sqlalchemy.orm import Session from app.config import get_settings from app.models import Admin, User, UserRole from app.schemas import AdminProfilePublic, AdminProfileUpdate, RoleEnum from app.services.admin_service import media_public_url, uploads_base_dir ALLOWED_ADMIN_IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp"} ALLOWED_ADMIN_IMAGE_CONTENT_TYPES = { "image/jpeg", "image/jpg", "image/pjpeg", "image/png", "image/x-png", "image/gif", "image/webp", } def sync_admin_table(engine: Engine) -> None: """Ensure `admin` table exists (create_all also runs; this adds columns on old DBs).""" from app.database import Base from app.models import Admin # noqa: F401 Base.metadata.create_all(bind=engine, tables=[Admin.__table__]) try: insp = inspect(engine) if "admin" not in insp.get_table_names(): return cols = {c["name"] for c in insp.get_columns("admin")} except Exception: return alters: list[str] = [] if "phone_number" not in cols: alters.append("phone_number VARCHAR(32)") if "profile_image" not in cols: alters.append("profile_image VARCHAR(512)") if not alters: return is_pg = engine.dialect.name == "postgresql" with engine.begin() as conn: for col_def in alters: col_name = col_def.split()[0] if is_pg: conn.execute(text(f"ALTER TABLE admin ADD COLUMN IF NOT EXISTS {col_def}")) else: try: conn.execute(text(f"ALTER TABLE admin ADD COLUMN {col_def}")) except Exception: if col_name not in {c["name"] for c in inspect(engine).get_columns("admin")}: raise def _split_full_name(full_name: str) -> tuple[str, str]: parts = (full_name or "").strip().split(None, 1) if not parts: return "", "" if len(parts) == 1: return parts[0], "" return parts[0], parts[1] def ensure_admin_profile(db: Session, user: User) -> Admin: row = db.query(Admin).filter(Admin.user_id == user.id).first() if row: return row first, last = _split_full_name(user.full_name) row = Admin( user_id=user.id, first_name=first, last_name=last, phone_number=None, email=user.email, profile_image=None, ) db.add(row) db.commit() db.refresh(row) return row def backfill_admin_profiles(db: Session) -> None: admins = db.query(User).filter(User.role == UserRole.admin).all() for user in admins: ensure_admin_profile(db, user) def serialize_admin_profile(admin: Admin) -> AdminProfilePublic: full_name = f"{admin.first_name} {admin.last_name}".strip() image_url = media_public_url(admin.profile_image) return AdminProfilePublic( id=admin.id, user_id=admin.user_id, first_name=admin.first_name, last_name=admin.last_name, full_name=full_name, phone_number=admin.phone_number, email=admin.email, role=RoleEnum.admin, profile_image=admin.profile_image, profile_image_url=image_url, has_profile_image=bool(admin.profile_image), created_at=admin.created_at, updated_at=admin.updated_at, ) def _resolve_names(body: AdminProfileUpdate) -> tuple[str | None, str | None]: if body.first_name is not None or body.last_name is not None: return body.first_name, body.last_name if body.full_name is not None: return _split_full_name(body.full_name) return None, None def update_admin_profile(db: Session, user: User, body: AdminProfileUpdate) -> AdminProfilePublic: admin = ensure_admin_profile(db, user) first, last = _resolve_names(body) if first is not None: admin.first_name = first.strip() if last is not None: admin.last_name = last.strip() if body.phone_number is not None: admin.phone_number = body.phone_number.strip() or None if body.email is not None: new_email = str(body.email) if new_email != user.email: exists = db.query(User).filter(User.email == new_email, User.id != user.id).first() if exists: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Email already in use") user.email = new_email admin.email = new_email user.full_name = f"{admin.first_name} {admin.last_name}".strip() db.commit() db.refresh(admin) db.refresh(user) return serialize_admin_profile(admin) def admin_image_upload_dir() -> Path: dest = uploads_base_dir() / "admin" dest.mkdir(parents=True, exist_ok=True) return dest def delete_admin_profile_image(relative_path: str | None) -> None: if not relative_path: return path = admin_image_upload_dir() / Path(relative_path).name if path.is_file(): path.unlink() async def save_admin_profile_image(media: UploadFile) -> str: settings = get_settings() content_type = (media.content_type or "").split(";")[0].strip().lower() filename = media.filename or "" ext = Path(filename).suffix.lower() if not ext and content_type.startswith("image/"): ext = "." + content_type.split("/", 1)[1] if ext not in ALLOWED_ADMIN_IMAGE_EXTENSIONS: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Unsupported image type. Allowed: {', '.join(sorted(ALLOWED_ADMIN_IMAGE_EXTENSIONS))}", ) if ( content_type and content_type not in ALLOWED_ADMIN_IMAGE_CONTENT_TYPES and not content_type.startswith("image/") ): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Unsupported image content type") data = await media.read() max_bytes = settings.community_max_upload_mb * 1024 * 1024 if len(data) > max_bytes: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Image too large (max {settings.community_max_upload_mb} MB)", ) if not data: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Empty image file") stored_name = f"{uuid.uuid4().hex}{ext}" dest = admin_image_upload_dir() / stored_name dest.write_bytes(data) return f"admin/{stored_name}" def set_admin_profile_image(db: Session, user: User, image_path: str) -> AdminProfilePublic: admin = ensure_admin_profile(db, user) delete_admin_profile_image(admin.profile_image) admin.profile_image = image_path db.commit() db.refresh(admin) return serialize_admin_profile(admin)