Feat/direct bulk upload

app/core/config.py

@@ -35,6 +35,10 @@ class Settings(BaseSettings):
 
     PHOTO_APPROVAL_TIMEOUT_DAYS: int = 7
     EVENT_LIFECYCLE_POLL_INTERVAL_SECONDS: int = 60
+    DIRECT_UPLOAD_PRESIGN_EXPIRES_SECONDS: int = 1800
+    DIRECT_UPLOAD_STALE_PENDING_MINUTES: int = 45
+    DIRECT_UPLOAD_RECONCILE_POLL_INTERVAL_SECONDS: int = 300
+    DIRECT_UPLOAD_MAX_BATCH_SIZE: int = 200
 
     # Mobile auth/session defaults
     MOBILE_SESSION_LIMIT: int = 3
@@ -79,9 +83,20 @@ class Settings(BaseSettings):
     GOOGLE_CLIENT_ID: str = ""
     GOOGLE_CLIENT_SECRET: str = ""
     GOOGLE_REDIRECT_URI: str = ""
+    # drive.readonly alone can't write; drive.file alone can only see files
+    # the app itself created, which would break browsing/importing existing
+    # Drive folders. Both scopes together preserve the existing read/import
+    # flow and add write access for syncing approved direct uploads back to
+    # Drive. Existing staff connections keep their old readonly-only grant
+    # until they disconnect and reconnect through the consent screen.
     GOOGLE_OAUTH_SCOPES: str = (
-        "https://www.googleapis.com/auth/drive.readonly openid email profile"
+        "https://www.googleapis.com/auth/drive.readonly "
+        "https://www.googleapis.com/auth/drive.file openid email profile"
     )
+    # Folder ID (from the Drive URL) that approved direct-upload photos get
+    # synced into. Empty means uploads land in the connected account's Drive
+    # root instead of a specific folder.
+    GOOGLE_CLUB_DRIVE_FOLDER_ID: str = ""
 
     FACE_ENCRYPTION_KEY: str
     FIREBASE_CREDENTIALS_PATH: str

app/infra/google_drive.py

@@ -189,6 +189,67 @@ async def get_file_metadata(
             size_bytes=size_bytes,
         )
 
+    @staticmethod
+    async def upload_file(
+        *,
+        access_token: str,
+        file_name: str,
+        content_type: str,
+        data: bytes,
+        folder_id: str | None,
+    ) -> GoogleDriveFileMetadata:
+        boundary = "multai-drive-upload-boundary"
+        metadata: dict[str, object] = {"name": file_name}
+        if folder_id:
+            metadata["parents"] = [folder_id]
+
+        body = (
+            f"--{boundary}\r\n"
+            "Content-Type: application/json; charset=UTF-8\r\n\r\n"
+            f"{json.dumps(metadata)}\r\n"
+            f"--{boundary}\r\n"
+            f"Content-Type: {content_type}\r\n\r\n"
+        ).encode("utf-8") + data + f"\r\n--{boundary}--".encode("utf-8")
+
+        def _request() -> dict[str, object]:
+            url = (
+                "https://www.googleapis.com/upload/drive/v3/files"
+                "?uploadType=multipart&supportsAllDrives=true&fields=id,name,mimeType,size"
+            )
+            request = urllib.request.Request(
+                url,
+                data=body,
+                headers={
+                    "Authorization": f"Bearer {access_token}",
+                    "Content-Type": f"multipart/related; boundary={boundary}",
+                },
+                method="POST",
+            )
+            try:
+                with urllib.request.urlopen(request, timeout=60) 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 Drive file upload failed: {details or exc.reason}"
+                ) from exc
+            except urllib.error.URLError as exc:
+                raise AppException.internal_error("Unable to reach Google APIs") from exc
+
+        result = await asyncio.to_thread(_request)
+        size_raw = result.get("size", "0")
+        try:
+            size_bytes = int(size_raw) if isinstance(size_raw, (str, int)) else len(data)
+        except (TypeError, ValueError):
+            size_bytes = len(data)
+
+        return GoogleDriveFileMetadata(
+            id=GoogleDriveClient._require_str(result, "id"),
+            name=GoogleDriveClient._require_str(result, "name"),
+            mime_type=GoogleDriveClient._require_str(result, "mimeType"),
+            size_bytes=size_bytes,
+        )
+
     @staticmethod
     async def download_file(
         *,

app/infra/minio.py

@@ -2,6 +2,8 @@
 import random
 import string
 import uuid
+from dataclasses import dataclass
+from datetime import timedelta
 from fastapi import UploadFile
 from miniopy_async.commonconfig import CopySource
 from miniopy_async.error import S3Error
@@ -36,6 +38,12 @@ async def init_minio_client(
         if not await Bucket.client.bucket_exists(bucket_name):
             await Bucket.client.make_bucket(bucket_name)
 
+@dataclass(frozen=True)
+class ObjectStat:
+    size: int
+    content_type: str
+
+
 class Bucket:
     bucket_name: str
     file_prefix: str
@@ -130,6 +138,25 @@ async def copy(self, *, source_object_name: str, target_object_name: str) -> str
         )
         return target_object_name
 
+    async def presigned_put_url(self, object_name: str, *, expires_seconds: int) -> str:
+        return await self.client.presigned_put_object(
+            bucket_name=self.bucket_name,
+            object_name=self._object_path(object_name),
+            expires=timedelta(seconds=expires_seconds),
+        )
+
+    async def stat(self, object_name: str) -> ObjectStat | None:
+        try:
+            result = await self.client.stat_object(
+                bucket_name=self.bucket_name,
+                object_name=self._object_path(object_name),
+            )
+        except S3Error as e:
+            if e.code == "NoSuchKey":
+                return None
+            raise
+        return ObjectStat(size=result.size or 0, content_type=result.content_type or DEFAULT_CONTENT_TYPE)
+
 image_ext_content_type_map = {
     "apng": ["image/apng"],
     "avif": ["image/avif"],

app/infra/nats.py

@@ -35,6 +35,7 @@ class NatsSubjects(Enum):
     STAFF_UPLOAD_REQUEST_APPROVED = "staff.upload_request.approved"
     STAFF_UPLOAD_REQUEST_REJECTED = "staff.upload_request.rejected"
     PHOTO_PROCESS = "photo.process"
+    PHOTO_DRIVE_SYNC_REQUESTED = "photo.drive_sync.requested"
 
 
 class NatsClient:

app/router/staff/__init__.py

@@ -3,8 +3,10 @@
 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
+from app.router.staff.uploads_direct import router as staff_uploads_direct_router
 
 router = APIRouter(prefix="/staff", tags=["staff"])
 router.include_router(staff_drive_router)
 router.include_router(staff_notifications_router)
 router.include_router(staff_uploads_router)
+router.include_router(staff_uploads_direct_router)

app/router/staff/uploads_direct.py

@@ -0,0 +1,119 @@
+from uuid import UUID
+
+from fastapi import APIRouter, Depends
+
+from app.container import Container, get_container
+from app.deps.cookie_auth import get_current_staff_user
+from app.schema.internal.uploads import DirectFileInput
+from app.schema.request.staff.uploads_direct import (
+    CreateDirectGroupRequest,
+    RegisterDirectBatchRequest,
+)
+from app.schema.response.staff.upload_groups import UploadRequestGroupSchema
+from app.schema.response.staff.uploads_direct import (
+    DirectUploadFileResponse,
+    RegisterDirectBatchResponse,
+    ResumeDirectGroupResponse,
+)
+from db.generated.models import StaffUser
+
+router = APIRouter(prefix="/uploads/direct")
+# this endpoint are for staff to upload images directly to the system and very large files and they can resume and restart and retry .
+
+@router.post("/groups", response_model=UploadRequestGroupSchema)
+async def create_direct_group(
+    req: CreateDirectGroupRequest,
+    current_staff_user: StaffUser = Depends(get_current_staff_user),
+    container: Container = Depends(get_container),
+) -> UploadRequestGroupSchema:
+    group = await container.upload_requests_service.create_direct_group(
+        event_id=req.event_id, requested_by=current_staff_user,
+    )
+    details = await container.upload_requests_service.get_group_details(
+        group_id=group.id, current_staff_user=current_staff_user,
+    )
+    return UploadRequestGroupSchema.from_details(details)
+
+
+@router.post("/groups/{group_id}/batches", response_model=RegisterDirectBatchResponse)
+async def register_direct_batch(
+    group_id: UUID,
+    req: RegisterDirectBatchRequest,
+    current_staff_user: StaffUser = Depends(get_current_staff_user),
+    container: Container = Depends(get_container),
+) -> RegisterDirectBatchResponse:
+    results = await container.upload_requests_service.register_direct_batch(
+        group_id=group_id,
+        files=[
+            DirectFileInput(
+                file_name=f.file_name,
+                mime_type=f.mime_type,
+                size_bytes=f.size_bytes,
+                taken_at=f.taken_at,
+                day_number=f.day_number,
+                visibility=f.visibility,
+            )
+            for f in req.files
+        ],
+        requested_by=current_staff_user,
+    )
+    return RegisterDirectBatchResponse(
+        group_id=group_id,
+        items=[
+            DirectUploadFileResponse(photo_id=photo.id, file_name=photo.file_name, upload_url=url)
+            for photo, url in results
+        ],
+    )
+
+
+@router.post("/photos/{photo_id}/confirm", response_model=DirectUploadFileResponse)
+async def confirm_direct_upload(
+    photo_id: UUID,
+    current_staff_user: StaffUser = Depends(get_current_staff_user),
+    container: Container = Depends(get_container),
+) -> DirectUploadFileResponse:
+    photo = await container.upload_requests_service.confirm_direct_upload(
+        photo_id=photo_id, requested_by=current_staff_user,
+    )
+    return DirectUploadFileResponse(photo_id=photo.id, file_name=photo.file_name, upload_url="")
+
+
+@router.post("/photos/{photo_id}/fail", response_model=DirectUploadFileResponse)
+async def fail_direct_upload(
+    photo_id: UUID,
+    current_staff_user: StaffUser = Depends(get_current_staff_user),
+    container: Container = Depends(get_container),
+) -> DirectUploadFileResponse:
+    photo = await container.upload_requests_service.fail_direct_upload(
+        photo_id=photo_id, requested_by=current_staff_user,
+    )
+    return DirectUploadFileResponse(photo_id=photo.id, file_name=photo.file_name, upload_url="")
+
+
+@router.post("/groups/{group_id}/resume", response_model=ResumeDirectGroupResponse)
+async def resume_direct_group(
+    group_id: UUID,
+    current_staff_user: StaffUser = Depends(get_current_staff_user),
+    container: Container = Depends(get_container),
+) -> ResumeDirectGroupResponse:
+    results = await container.upload_requests_service.resume_direct_group(
+        group_id=group_id, requested_by=current_staff_user,
+    )
+    return ResumeDirectGroupResponse(
+        items=[
+            DirectUploadFileResponse(photo_id=photo.id, file_name=photo.file_name, upload_url=url)
+            for photo, url in results
+        ]
+    )
+
+
+@router.get("/groups/{group_id}", response_model=UploadRequestGroupSchema)
+async def get_direct_group_status(
+    group_id: UUID,
+    current_staff_user: StaffUser = Depends(get_current_staff_user),
+    container: Container = Depends(get_container),
+) -> UploadRequestGroupSchema:
+    details = await container.upload_requests_service.get_group_details(
+        group_id=group_id, current_staff_user=current_staff_user,
+    )
+    return UploadRequestGroupSchema.from_details(details)

app/schema/internal/uploads.py

@@ -8,3 +8,13 @@ class UploadPhotoInput:
     taken_at: datetime | None
     day_number: int | None
     visibility: str
+
+
+@dataclass(frozen=True)
+class DirectFileInput:
+    file_name: str
+    mime_type: str
+    size_bytes: int
+    taken_at: datetime | None
+    day_number: int | None
+    visibility: str

app/schema/request/staff/uploads_direct.py

@@ -0,0 +1,22 @@
+from datetime import datetime
+from typing import Optional
+from uuid import UUID
+
+from pydantic import BaseModel
+
+
+class DirectFileInputRequest(BaseModel):
+    file_name: str
+    mime_type: str
+    size_bytes: int
+    taken_at: Optional[datetime] = None
+    day_number: Optional[int] = None
+    visibility: str = "private"
+
+
+class CreateDirectGroupRequest(BaseModel):
+    event_id: UUID
+
+
+class RegisterDirectBatchRequest(BaseModel):
+    files: list[DirectFileInputRequest]

app/schema/response/staff/upload_groups.py

@@ -18,10 +18,11 @@ class UploadRequestGroupSchema(BaseModel):
 
     id: UUID
     event_id: UUID
-    folder_id: str
+    folder_id: str | None
     requested_by: UUID
     approved_by: UUID | None
     status: str
+    source: str
     processing_status: str
     total_photo_count: int
     batch_count: int
@@ -72,8 +73,9 @@ class UploadRequestGroupSummarySchema(BaseModel):
     model_config = ConfigDict(from_attributes=True)
     id: UUID
     event_id: UUID
-    folder_id: str
+    folder_id: str | None
     status: str
+    source: str
     processing_status: str
     total_photo_count: int
     batch_count: int

app/schema/response/staff/uploads.py

@@ -11,14 +11,16 @@ class UploadRequestPhotoSchema(BaseModel):
     model_config = ConfigDict(from_attributes=True)
 
     id: UUID
-    drive_file_id: str
+    drive_file_id: str | None
     file_name: str
     mime_type: str
     size_bytes: int
     taken_at: datetime | None
     day_number: int | None
     visibility: str
     status: str
+    source: str
+    transfer_status: str
     created_at: datetime
 
 
@@ -32,6 +34,7 @@ class UploadRequestSchema(BaseModel):
     requested_by: UUID
     approved_by: UUID | None
     status: str
+    source: str
     photo_count: int
     created_at: datetime
     approved_at: datetime | None

app/schema/response/staff/uploads_direct.py

@@ -0,0 +1,18 @@
+from uuid import UUID
+
+from pydantic import BaseModel
+
+
+class DirectUploadFileResponse(BaseModel):
+    photo_id: UUID
+    file_name: str
+    upload_url: str
+
+
+class RegisterDirectBatchResponse(BaseModel):
+    group_id: UUID
+    items: list[DirectUploadFileResponse]
+
+
+class ResumeDirectGroupResponse(BaseModel):
+    items: list[DirectUploadFileResponse]