opencloud-backend/app/routers/favorites.py

import math
import uuid

from fastapi import APIRouter, HTTPException, Query, status
from sqlalchemy import delete, func, select
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
from sqlalchemy.orm import joinedload

from app.deps import CurrentUser, DbSession
from app.models import Cloud, CloudFavorite, User
from app.schemas import CloudOut, FavoriteIn, FavoriteOut, PageOut
from app.serializers import cloud_out


router = APIRouter(tags=["点赞收藏"])


def _cloud_options():
    return (
        joinedload(Cloud.user).joinedload(User.profile),
        joinedload(Cloud.cloud_type),
    )


async def _get_visible_cloud(
    db: DbSession,
    cloud_id: uuid.UUID,
    user: CurrentUser,
) -> Cloud:
    result = await db.execute(
        select(Cloud).options(*_cloud_options()).where(Cloud.id == cloud_id)
    )
    cloud = result.scalar_one_or_none()
    if not cloud:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="图片不存在")

    can_manage = cloud.user_id == user.id or user.profile.role == "admin"
    if not can_manage and (cloud.status != "approved" or cloud.is_hidden):
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="图片不存在")
    return cloud


async def _favorite_count(db: DbSession, cloud_id: uuid.UUID) -> int:
    return (
        await db.scalar(
            select(func.count())
            .select_from(CloudFavorite)
            .where(CloudFavorite.cloud_id == cloud_id)
        )
        or 0
    )


async def _is_favorited(db: DbSession, user_id: uuid.UUID, cloud_id: uuid.UUID) -> bool:
    favorite_id = await db.scalar(
        select(CloudFavorite.id).where(
            CloudFavorite.user_id == user_id,
            CloudFavorite.cloud_id == cloud_id,
        )
    )
    return favorite_id is not None


async def _favorite_out(
    db: DbSession,
    *,
    cloud_id: uuid.UUID,
    user_id: uuid.UUID,
    favorited: bool | None = None,
) -> FavoriteOut:
    if favorited is None:
        favorited = await _is_favorited(db, user_id, cloud_id)
    return FavoriteOut(
        cloud_id=cloud_id,
        user_id=user_id,
        favorited=favorited,
        favorite_count=await _favorite_count(db, cloud_id),
    )


@router.get("/clouds/{cloud_id}/favorite", response_model=FavoriteOut)
async def get_cloud_favorite(
    cloud_id: uuid.UUID,
    user: CurrentUser,
    db: DbSession,
    user_id: uuid.UUID | None = Query(
        None,
        description="要查询的用户 ID；默认当前登录用户。仅本人或管理员可查他人。",
    ),
) -> FavoriteOut:
    """查询某用户对某张照片的点赞/收藏状态。"""
    await _get_visible_cloud(db, cloud_id, user)

    target_user_id = user_id or user.id
    if target_user_id != user.id and user.profile.role != "admin":
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="只能查询自己的喜欢状态",
        )

    return await _favorite_out(db, cloud_id=cloud_id, user_id=target_user_id)


@router.put("/clouds/{cloud_id}/favorite", response_model=FavoriteOut)
async def set_cloud_favorite(
    cloud_id: uuid.UUID,
    payload: FavoriteIn,
    user: CurrentUser,
    db: DbSession,
) -> FavoriteOut:
    """点赞/收藏照片，或取消。"""
    await _get_visible_cloud(db, cloud_id, user)

    if payload.favorited:
        insert_factory = (
            pg_insert if db.bind and db.bind.dialect.name == "postgresql" else sqlite_insert
        )
        await db.execute(
            insert_factory(CloudFavorite)
            .values(user_id=user.id, cloud_id=cloud_id)
            .on_conflict_do_nothing(
                index_elements=[CloudFavorite.user_id, CloudFavorite.cloud_id]
            )
        )
    else:
        await db.execute(
            delete(CloudFavorite).where(
                CloudFavorite.user_id == user.id,
                CloudFavorite.cloud_id == cloud_id,
            )
        )

    await db.commit()
    return await _favorite_out(
        db,
        cloud_id=cloud_id,
        user_id=user.id,
        favorited=payload.favorited,
    )


@router.get("/favorites/me", response_model=PageOut[CloudOut])
async def list_my_favorites(
    user: CurrentUser,
    db: DbSession,
    page: int = Query(1, ge=1),
    page_size: int = Query(50, ge=1, le=100),
) -> PageOut[CloudOut]:
    """当前用户点赞收藏的照片列表（按收藏时间倒序）。"""
    total = (
        await db.scalar(
            select(func.count())
            .select_from(CloudFavorite)
            .where(CloudFavorite.user_id == user.id)
        )
        or 0
    )
    result = await db.execute(
        select(Cloud)
        .join(CloudFavorite, CloudFavorite.cloud_id == Cloud.id)
        .options(*_cloud_options())
        .where(CloudFavorite.user_id == user.id)
        .order_by(CloudFavorite.created_at.desc(), CloudFavorite.id.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
    )
    items = [
        cloud_out(item, include_private=item.user_id == user.id or user.profile.role == "admin")
        for item in result.scalars().unique().all()
    ]
    return PageOut(
        items=items,
        page=page,
        page_size=page_size,
        total=total,
        total_pages=max(1, math.ceil(total / page_size)) if total else 1,
    )