import math import uuid from datetime import datetime, timezone from decimal import Decimal, ROUND_HALF_UP from typing import Annotated, Literal from fastapi import APIRouter, File, Form, HTTPException, Query, UploadFile, status from sqlalchemy import func, or_, select, update 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, OptionalUser from app.models import Cloud, CloudFavorite, CloudType, User, UserCollection from app.schemas import ( BadgeOut, BatchIdsIn, CloudCreateOut, CloudOut, CloudUpdateIn, DeleteResultOut, PageOut, ) from app.serializers import cloud_out from app.services.storage import delete_files, save_cloud_image router = APIRouter(prefix="/clouds", tags=["云图"]) def _cloud_options(): return ( joinedload(Cloud.user), joinedload(Cloud.cloud_type), ) def _normalize_text(value: str | None) -> str | None: if value is None: return None stripped = value.strip() return stripped or None def _blur_coordinate(value: float | None) -> Decimal | None: if value is None: return None return Decimal(str(value)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) def _normalize_datetime(value: datetime | None) -> datetime | None: if value is None: return None if value.tzinfo is None: return value.replace(tzinfo=timezone.utc) return value def _normalize_required_datetime(value: datetime) -> datetime: return _normalize_datetime(value) or value.replace(tzinfo=timezone.utc) async def _require_cloud_type(db: DbSession, cloud_type_id: int) -> CloudType: cloud_type = await db.get(CloudType, cloud_type_id) if not cloud_type: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="云类型不存在") return cloud_type async def _get_cloud(db: DbSession, cloud_id: uuid.UUID) -> Cloud | None: result = await db.execute( select(Cloud).options(*_cloud_options()).where(Cloud.id == cloud_id) ) return result.scalar_one_or_none() @router.get("", response_model=PageOut[CloudOut]) async def list_gallery_clouds( db: DbSession, page: int = Query(1, ge=1), page_size: int = Query(50, ge=1, le=100), type_id: int | None = Query(None, ge=1), search: str | None = Query(None, max_length=80), ) -> PageOut[CloudOut]: filters = [Cloud.status == "approved", Cloud.is_hidden.is_(False)] if type_id is not None: filters.append(Cloud.cloud_type_id == type_id) data_query = select(Cloud).options(*_cloud_options()) count_query = select(func.count(Cloud.id)) term = _normalize_text(search) if term and term.startswith("@"): username = term[1:].strip() if not username: return PageOut(items=[], page=1, page_size=page_size, total=0, total_pages=1) data_query = data_query.join(Cloud.user) count_query = count_query.join(Cloud.user) filters.append(User.username.ilike(f"%{username}%")) elif term: data_query = data_query.outerjoin(Cloud.cloud_type) count_query = count_query.outerjoin(Cloud.cloud_type) filters.append( or_( CloudType.name.ilike(f"%{term}%"), CloudType.name_en.ilike(f"%{term}%"), Cloud.custom_cloud_type.ilike(f"%{term}%"), ) ) total = await db.scalar(count_query.where(*filters)) or 0 result = await db.execute( data_query.where(*filters) .order_by(Cloud.created_at.desc(), Cloud.id.desc()) .offset((page - 1) * page_size) .limit(page_size) ) return PageOut( items=[cloud_out(item, include_private=False) for item in result.scalars().unique().all()], page=page, page_size=page_size, total=total, total_pages=max(1, math.ceil(total / page_size)), ) @router.get("/map", response_model=list[CloudOut]) async def list_map_clouds( db: DbSession, start: datetime, end: datetime, time_field: Literal["captured_at", "created_at"] = "captured_at", limit: int = Query(1000, ge=1, le=1000), ) -> list[CloudOut]: start = _normalize_required_datetime(start) end = _normalize_required_datetime(end) if start >= end: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="结束时间必须晚于开始时间") field = Cloud.captured_at if time_field == "captured_at" else Cloud.created_at result = await db.execute( select(Cloud) .options(*_cloud_options()) .where( Cloud.status == "approved", Cloud.is_hidden.is_(False), Cloud.latitude.is_not(None), Cloud.longitude.is_not(None), field >= start, field < end, ) .order_by(field.asc(), Cloud.id.asc()) .limit(limit) ) return [cloud_out(item, include_private=False) for item in result.scalars().unique().all()] @router.get("/{cloud_id}", response_model=CloudOut) async def get_cloud(cloud_id: uuid.UUID, db: DbSession, viewer: OptionalUser) -> CloudOut: cloud = await _get_cloud(db, cloud_id) if not cloud: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="图片不存在") can_manage = bool(viewer and (viewer.id == cloud.user_id or viewer.role == "admin")) if not can_manage and (cloud.status != "approved" or cloud.is_hidden): raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="图片不存在") favorite_count = cloud.favorite_count is_favorited = False if viewer: fav_id = await db.scalar( select(CloudFavorite.id).where( CloudFavorite.user_id == viewer.id, CloudFavorite.cloud_id == cloud_id, ) ) is_favorited = fav_id is not None return cloud_out(cloud, include_private=can_manage, favorite_count=favorite_count, is_favorited=is_favorited) @router.post("", response_model=CloudCreateOut, status_code=status.HTTP_201_CREATED) async def create_cloud( db: DbSession, user: CurrentUser, image: Annotated[UploadFile, File()], cloud_type_id: Annotated[int | None, Form()] = None, custom_cloud_type: Annotated[str | None, Form(max_length=80)] = None, latitude: Annotated[float | None, Form(ge=-90, le=90)] = None, longitude: Annotated[float | None, Form(ge=-180, le=180)] = None, location_name: Annotated[str | None, Form(max_length=120)] = None, description: Annotated[str | None, Form(max_length=2000)] = None, captured_at: Annotated[datetime | None, Form()] = None, is_hidden: Annotated[bool, Form()] = False, ) -> CloudCreateOut: custom_cloud_type = _normalize_text(custom_cloud_type) if (cloud_type_id is None) == (custom_cloud_type is None): raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="必须选择一个云类型，或填写自定义云类型", ) if (latitude is None) != (longitude is None): raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="经纬度必须同时填写") cloud_type = await _require_cloud_type(db, cloud_type_id) if cloud_type_id else None image_path, thumbnail_path = await save_cloud_image(image, user.id) cloud = Cloud( user_id=user.id, cloud_type_id=cloud_type_id, custom_cloud_type=custom_cloud_type, image_path=image_path, thumbnail_path=thumbnail_path, latitude=_blur_coordinate(latitude), longitude=_blur_coordinate(longitude), location_name=_normalize_text(location_name), description=_normalize_text(description), captured_at=_normalize_datetime(captured_at) or datetime.now(timezone.utc), status="pending", is_hidden=is_hidden, ) db.add(cloud) unlocked_badge: BadgeOut | None = None try: await db.flush() await db.execute( update(User) .where(User.id == user.id) .values(cloud_count=User.cloud_count + 1) ) if cloud_type: insert_factory = pg_insert if db.bind and db.bind.dialect.name == "postgresql" else sqlite_insert insert_result = await db.execute( insert_factory(UserCollection) .values( user_id=user.id, cloud_type_id=cloud_type.id, first_cloud_id=cloud.id, ) .on_conflict_do_nothing( index_elements=[UserCollection.user_id, UserCollection.cloud_type_id] ) .returning(UserCollection.unlocked_at) ) unlocked_at = insert_result.scalar_one_or_none() if unlocked_at: await db.execute( update(User) .where(User.id == user.id) .values(collection_count=User.collection_count + 1) ) unlocked_badge = BadgeOut( cloud_type_id=cloud_type.id, cloud_name=cloud_type.name, cloud_name_en=cloud_type.name_en, rarity=cloud_type.rarity, unlocked_at=unlocked_at, ) await db.commit() except Exception: await db.rollback() await delete_files([image_path, thumbnail_path]) raise saved = await _get_cloud(db, cloud.id) if not saved: raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="图片记录创建失败") return CloudCreateOut(cloud=cloud_out(saved), unlocked_badge=unlocked_badge) @router.patch("/{cloud_id}", response_model=CloudOut) async def update_cloud( cloud_id: uuid.UUID, payload: CloudUpdateIn, db: DbSession, user: CurrentUser, ) -> CloudOut: cloud = await _get_cloud(db, cloud_id) if not cloud: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="图片不存在") if cloud.user_id != user.id and user.role != "admin": raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="不能修改其他用户的图片") previous_public = cloud.status == "approved" and not cloud.is_hidden changes = payload.model_dump(exclude_unset=True) final_type_id = changes.get("cloud_type_id", cloud.cloud_type_id) final_custom = _normalize_text(changes.get("custom_cloud_type", cloud.custom_cloud_type)) if "cloud_type_id" in changes and changes["cloud_type_id"] is not None and "custom_cloud_type" not in changes: final_custom = None if "custom_cloud_type" in changes and final_custom is not None and "cloud_type_id" not in changes: final_type_id = None if (final_type_id is None) == (final_custom is None): raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="必须选择一个云类型，或填写自定义云类型", ) if final_type_id is not None: await _require_cloud_type(db, final_type_id) if "cloud_type_id" in changes or "custom_cloud_type" in changes: cloud.cloud_type_id = final_type_id cloud.custom_cloud_type = final_custom if "latitude" in changes: cloud.latitude = _blur_coordinate(changes["latitude"]) cloud.longitude = _blur_coordinate(changes["longitude"]) if "location_name" in changes: cloud.location_name = _normalize_text(changes["location_name"]) if "description" in changes: cloud.description = _normalize_text(changes["description"]) if "captured_at" in changes: cloud.captured_at = _normalize_datetime(changes["captured_at"]) if "is_hidden" in changes: cloud.is_hidden = changes["is_hidden"] await db.flush() new_public = cloud.status == "approved" and not cloud.is_hidden if previous_public != new_public: delta = 1 if new_public else -1 await db.execute( update(User) .where(User.id == cloud.user_id) .values(public_cloud_count=User.public_cloud_count + delta) ) await db.commit() updated = await _get_cloud(db, cloud.id) if not updated: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="图片不存在") return cloud_out(updated) @router.delete("/{cloud_id}", response_model=DeleteResultOut) async def delete_cloud(cloud_id: uuid.UUID, db: DbSession, user: CurrentUser) -> DeleteResultOut: cloud = await _get_cloud(db, cloud_id) if not cloud: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="图片不存在") if cloud.user_id != user.id and user.role != "admin": raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="不能删除其他用户的图片") paths = [cloud.image_path, cloud.thumbnail_path] was_public = cloud.status == "approved" and not cloud.is_hidden await db.delete(cloud) await db.execute( update(User) .where(User.id == cloud.user_id) .values(cloud_count=User.cloud_count - 1) ) if was_public: await db.execute( update(User) .where(User.id == cloud.user_id) .values(public_cloud_count=User.public_cloud_count - 1) ) await db.commit() await delete_files(paths) return DeleteResultOut(deleted=1) @router.post("/batch-delete", response_model=DeleteResultOut) async def batch_delete_clouds( payload: BatchIdsIn, db: DbSession, user: CurrentUser, ) -> DeleteResultOut: result = await db.execute( select(Cloud).where(Cloud.id.in_(payload.ids), Cloud.user_id == user.id) ) clouds = result.scalars().all() if len(clouds) != len(payload.ids): raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="部分图片不存在或不属于当前用户") public_count = sum(1 for c in clouds if c.status == "approved" and not c.is_hidden) paths = [path for cloud in clouds for path in (cloud.image_path, cloud.thumbnail_path)] for cloud in clouds: await db.delete(cloud) await db.execute( update(User) .where(User.id == user.id) .values( cloud_count=User.cloud_count - len(clouds), public_cloud_count=User.public_cloud_count - public_count, ) ) await db.commit() await delete_files(paths) return DeleteResultOut(deleted=len(clouds))