Add staged upload review workflow for Google Drive photos

app/container.py

@@ -7,6 +7,7 @@
 from app.service.device import DeviceService
 from app.service.face_embedding import FaceEmbeddingService
 from app.service.session import SessionService
+from app.service.staged_upload_storage import StagedUploadStorageService
 from app.service.staff_drive import StaffDriveService
 from app.service.staff_notifications import StaffNotificationsService
 from app.service.staff_user import StaffUserService
@@ -71,11 +72,14 @@ def __init__(
         self.staff_notifications_service = StaffNotificationsService(
             notification_querier=self.staff_notification_querier,
         )
+        self.staged_upload_storage_service = StagedUploadStorageService()
 
         self.upload_requests_service = UploadRequestsService(
             upload_request_querier=self.upload_request_querier,
             upload_request_photo_querier=self.upload_request_photo_querier,
             photo_querier=self.photo_querier,
+            staged_upload_storage=self.staged_upload_storage_service,
+            staff_drive_service=self.staff_drive_service,
             staff_notifications_service=self.staff_notifications_service,
         )
 

app/infra/google_drive.py

@@ -5,6 +5,7 @@
 import urllib.request
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
+from email.message import Message
 
 from app.core.exceptions import AppException
 from app.core.config import settings
@@ -13,6 +14,7 @@
 GOOGLE_AUTH_URL = "https://accounts.google.com/o/oauth2/v2/auth"
 GOOGLE_TOKEN_URL = "https://oauth2.googleapis.com/token"
 GOOGLE_USERINFO_URL = "https://www.googleapis.com/oauth2/v2/userinfo"
+GOOGLE_DRIVE_FILES_URL = "https://www.googleapis.com/drive/v3/files/{file_id}"
 
 
 @dataclass
@@ -31,6 +33,20 @@ class GoogleUserInfo:
     verified_email: bool
 
 
+@dataclass
+class GoogleDriveFileMetadata:
+    id: str
+    name: str
+    mime_type: str
+    size_bytes: int
+
+
+@dataclass
+class GoogleDriveFileDownload:
+    metadata: GoogleDriveFileMetadata
+    content: bytes
+
+
 class GoogleDriveClient:
     @staticmethod
     def _require_str(data: dict[str, object], key: str) -> str:
@@ -107,13 +123,64 @@ async def get_user_info(access_token: str) -> GoogleUserInfo:
         data = await GoogleDriveClient._get_json(
             GOOGLE_USERINFO_URL,
             headers={"Authorization": f"Bearer {access_token}"},
+            error_context="Google user info request",
         )
         return GoogleUserInfo(
             id=GoogleDriveClient._require_str(data, "id"),
             email=GoogleDriveClient._require_str(data, "email"),
             verified_email=bool(data.get("verified_email", False)),
         )
 
+    @staticmethod
+    async def get_file_metadata(
+        *,
+        access_token: str,
+        file_id: str,
+    ) -> GoogleDriveFileMetadata:
+        data = await GoogleDriveClient._get_json(
+            GOOGLE_DRIVE_FILES_URL.format(file_id=urllib.parse.quote(file_id, safe="")),
+            headers={"Authorization": f"Bearer {access_token}"},
+            query_params={
+                "fields": "id,name,mimeType,size",
+                "supportsAllDrives": "true",
+            },
+            error_context="Google Drive file metadata request",
+        )
+        size_raw = data.get("size", "0")
+        if not isinstance(size_raw, (str, int)):
+            raise AppException.bad_request("Google Drive file size is invalid")
+        try:
+            size_bytes = int(size_raw)
+        except (TypeError, ValueError) as exc:
+            raise AppException.bad_request("Google Drive file size is invalid") from exc
+
+        return GoogleDriveFileMetadata(
+            id=GoogleDriveClient._require_str(data, "id"),
+            name=GoogleDriveClient._require_str(data, "name"),
+            mime_type=GoogleDriveClient._require_str(data, "mimeType"),
+            size_bytes=size_bytes,
+        )
+
+    @staticmethod
+    async def download_file(
+        *,
+        access_token: str,
+        file_id: str,
+    ) -> GoogleDriveFileDownload:
+        metadata = await GoogleDriveClient.get_file_metadata(
+            access_token=access_token,
+            file_id=file_id,
+        )
+        content, _, _ = await GoogleDriveClient._get_bytes(
+            GOOGLE_DRIVE_FILES_URL.format(file_id=urllib.parse.quote(file_id, safe="")),
+            headers={"Authorization": f"Bearer {access_token}"},
+            query_params={
+                "alt": "media",
+                "supportsAllDrives": "true",
+            },
+        )
+        return GoogleDriveFileDownload(metadata=metadata, content=content)
+
     @staticmethod
     async def _post_form(url: str, payload: dict[str, str]) -> dict[str, object]:
         encoded = urllib.parse.urlencode(payload).encode("utf-8")
@@ -144,20 +211,53 @@ def _request() -> dict[str, object]:
     async def _get_json(
         url: str,
         headers: dict[str, str] | None = None,
+        query_params: dict[str, str] | None = None,
+        error_context: str = "Google API request",
     ) -> dict[str, object]:
         def _request() -> dict[str, object]:
-            request = urllib.request.Request(url, headers=headers or {}, method="GET")
+            final_url = url
+            if query_params:
+                final_url = f"{url}?{urllib.parse.urlencode(query_params)}"
+            request = urllib.request.Request(final_url, headers=headers or {}, method="GET")
             try:
                 with urllib.request.urlopen(request, timeout=15) as response:
                     return json.loads(response.read().decode("utf-8"))
             except urllib.error.HTTPError as exc:
                 details = exc.read().decode("utf-8", errors="ignore")
                 raise AppException.bad_request(
-                    f"Google user info request failed: {details or exc.reason}"
+                    f"{error_context} failed: {details or exc.reason}"
                 ) from exc
             except urllib.error.URLError as exc:
                 raise AppException.internal_error(
                     "Unable to reach Google APIs"
                 ) from exc
 
         return await asyncio.to_thread(_request)
+
+    @staticmethod
+    async def _get_bytes(
+        url: str,
+        headers: dict[str, str] | None = None,
+        query_params: dict[str, str] | None = None,
+    ) -> tuple[bytes, str, str]:
+        def _request() -> tuple[bytes, str, str]:
+            final_url = url
+            if query_params:
+                final_url = f"{url}?{urllib.parse.urlencode(query_params)}"
+            request = urllib.request.Request(final_url, headers=headers or {}, method="GET")
+            try:
+                with urllib.request.urlopen(request, timeout=30) as response:
+                    body = response.read()
+                    response_headers: Message = response.headers
+                    content_type = response_headers.get_content_type()
+                    file_name = response_headers.get_filename() or ""
+                    return body, content_type, file_name
+            except urllib.error.HTTPError as exc:
+                details = exc.read().decode("utf-8", errors="ignore")
+                raise AppException.bad_request(
+                    f"Google file download failed: {details or exc.reason}"
+                ) from exc
+            except urllib.error.URLError as exc:
+                raise AppException.internal_error("Unable to download file from Google Drive") from exc
+
+        return await asyncio.to_thread(_request)

app/infra/minio.py

@@ -1,7 +1,9 @@
+import io
 import random
 import string
 import uuid
 from fastapi import UploadFile
+from miniopy_async.commonconfig import CopySource
 from miniopy_async.error import S3Error
 from miniopy_async.api import Minio
 
@@ -36,6 +38,11 @@ def __init__(self, bucket_name: str, file_prefix: str):
         self.bucket_name = bucket_name
         self.file_prefix = file_prefix
 
+    def _object_path(self, object_name: str) -> str:
+        if self.file_prefix:
+            return f"{self.file_prefix}/{object_name}"
+        return object_name
+
     async def put(self, file: UploadFile, object_name: str | None = None) -> str:
         if object_name is None:
             object_name = str(uuid.uuid4())
@@ -48,7 +55,7 @@ async def put(self, file: UploadFile, object_name: str | None = None) -> str:
 
         await self.client.put_object(
             bucket_name=self.bucket_name,
-            object_name=f"{self.file_prefix}/{object_name}",
+            object_name=self._object_path(object_name),
             data=file.file,
             length=-1,
             part_size=10 * 1024 * 1024,
@@ -63,7 +70,7 @@ async def get(self, object_name: str) -> tuple[bytes, str, str]:
         try:
             res = await self.client.get_object(
                 bucket_name=self.bucket_name,
-                object_name=f"{self.file_prefix}/{object_name}",
+                object_name=self._object_path(object_name),
             )
         except S3Error as e:
             if e.code == "NoSuchKey":
@@ -83,8 +90,38 @@ async def get(self, object_name: str) -> tuple[bytes, str, str]:
     async def delete(self, object_name: str) -> None:
         await self.client.remove_object(
             bucket_name=self.bucket_name,
-            object_name=f"{self.file_prefix}/{object_name}",
+            object_name=self._object_path(object_name),
+        )
+
+    async def put_bytes(
+        self,
+        *,
+        data: bytes,
+        object_name: str,
+        content_type: str,
+        filename: str | None = None,
+    ) -> str:
+        await self.client.put_object(
+            bucket_name=self.bucket_name,
+            object_name=self._object_path(object_name),
+            data=io.BytesIO(data),
+            length=len(data),
+            part_size=10 * 1024 * 1024,
+            content_type=content_type,
+            metadata={"filename": filename or object_name},
+        )
+        return object_name
+
+    async def copy(self, *, source_object_name: str, target_object_name: str) -> str:
+        await self.client.copy_object(
+            bucket_name=self.bucket_name,
+            object_name=self._object_path(target_object_name),
+            source=CopySource(
+                self.bucket_name,
+                self._object_path(source_object_name),
+            ),
         )
+        return target_object_name
 
 image_ext_content_type_map = {
     "apng": ["image/apng"],

app/infra/nats.py

@@ -15,6 +15,9 @@ class NatsSubjects(Enum):
     USER_SIGNUP = "user.signup"
     USER_LOGIN = "user.login"
     USER_LOGOUT = "user.logout"
+    STAFF_UPLOAD_REQUEST_CREATED = "staff.upload_request.created"
+    STAFF_UPLOAD_REQUEST_APPROVED = "staff.upload_request.approved"
+    STAFF_UPLOAD_REQUEST_REJECTED = "staff.upload_request.rejected"
 
 class NatsClient:
     _nc: Optional[NATS] = None

app/router/staff/__init__.py

@@ -1,6 +1,10 @@
-from app.router.staff.drive import router as staff_drive_router
 from fastapi import APIRouter
 
+from app.router.staff.drive import router as staff_drive_router
+from app.router.staff.notifications import router as staff_notifications_router
+from app.router.staff.uploads import router as staff_uploads_router
 
-router = APIRouter(prefix="/stuff",tags=["stuff"])
+router = APIRouter(prefix="/stuff", tags=["stuff"])
 router.include_router(staff_drive_router)
+router.include_router(staff_notifications_router)
+router.include_router(staff_uploads_router)

app/router/staff/uploads.py

@@ -2,6 +2,7 @@
 from uuid import UUID
 
 from fastapi import APIRouter, Depends, Query
+from fastapi.responses import Response
 
 from app.container import Container, get_container
 from app.deps.staff_auth import (
@@ -14,6 +15,7 @@
 )
 from app.schema.response.staff.uploads import (
     UploadRequestListResponse,
+    UploadRequestPhotoListResponse,
     UploadRequestSchema,
 )
 from db.generated.models import StaffUser, UploadRequestStatus
@@ -53,6 +55,48 @@ async def list_upload_requests(
     )
 
 
+@router.get("/{request_id}", response_model=UploadRequestSchema)
+async def get_upload_request(
+    request_id: UUID,
+    current_staff_user: StaffUser = Depends(get_current_staff_user),
+    container: Container = Depends(get_container),
+) -> UploadRequestSchema:
+    upload_request = await container.upload_requests_service.get_request_details(
+        request_id=request_id,
+        current_staff_user=current_staff_user,
+    )
+    return UploadRequestSchema.from_models(upload_request.request, upload_request.photos)
+
+
+@router.get("/{request_id}/photos", response_model=UploadRequestPhotoListResponse)
+async def list_upload_request_photos(
+    request_id: UUID,
+    current_staff_user: StaffUser = Depends(get_current_staff_user),
+    container: Container = Depends(get_container),
+) -> UploadRequestPhotoListResponse:
+    upload_request = await container.upload_requests_service.get_request_details(
+        request_id=request_id,
+        current_staff_user=current_staff_user,
+    )
+    return UploadRequestPhotoListResponse.from_models(upload_request.photos)
+
+
+@router.get("/{request_id}/photos/{photo_id}/preview")
+async def preview_upload_request_photo(
+    request_id: UUID,
+    photo_id: UUID,
+    current_staff_user: StaffUser = Depends(get_current_staff_user),
+    container: Container = Depends(get_container),
+) -> Response:
+    preview = await container.upload_requests_service.get_request_photo_preview(
+        request_id=request_id,
+        photo_id=photo_id,
+        current_staff_user=current_staff_user,
+    )
+    headers = {"Content-Disposition": f'inline; filename="{preview.file_name}"'}
+    return Response(content=preview.data, media_type=preview.content_type, headers=headers)
+
+
 @router.post("/{request_id}/approve", response_model=UploadRequestSchema)
 async def approve_upload_request(
     request_id: UUID,

app/schema/dto/staff/uploads.py

@@ -5,9 +5,6 @@
 @dataclass(frozen=True)
 class UploadPhotoInput:
     drive_file_id: str
-    file_name: str
-    mime_type: str
-    size_bytes: int
     taken_at: datetime | None
     day_number: int | None
     visibility: str