From f87cd07d5e758f39e59bd15e495a601e576dc951 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EC=84=B1=EA=B2=BD?= Date: Tue, 11 Aug 2026 17:06:21 +0900 Subject: [PATCH] =?UTF-8?q?feat(home):=20=EC=9D=B4=EB=AF=B8=EC=A7=80=20?= =?UTF-8?q?=EC=97=85=EB=A1=9C=EB=93=9C=EC=97=90=20=ED=81=AC=EA=B8=B0=C2=B7?= =?UTF-8?q?=EA=B0=9C=EC=88=98=20=EC=83=81=ED=95=9C=EA=B3=BC=20task=20?= =?UTF-8?q?=EC=A7=81=EB=A0=AC=ED=99=94=20=EB=9D=BD=20=EB=8F=84=EC=9E=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 62 +++- app/home/api/routers/v1/home.py | 549 ++++++++++++++++++++-------- app/home/services/image_upload.py | 380 +++++++++++++++++++ app/utils/upload_blob_as_request.py | 250 ++++++++++++- config.py | 39 +- 5 files changed, 1112 insertions(+), 168 deletions(-) create mode 100644 app/home/services/image_upload.py diff --git a/README.md b/README.md index 5c824dc..7049211 100644 --- a/README.md +++ b/README.md @@ -176,6 +176,66 @@ fastapi dev main.py fastapi run main.py ``` +### 운영 업로드 및 메모리 한도 + +`POST /api/image/upload/blob`은 애플리케이션에서 파일당 15 MiB까지만 허용합니다. +운영 Nginx에서는 multipart 오버헤드를 고려해 이 엔드포인트의 요청 본문을 +25 MiB로 제한합니다. 앱의 요청당 파일 합계 상한은 20 MiB이며, 나머지 5 MiB는 +multipart 헤더와 `images_json`을 위한 여유입니다. 한 task에는 최대 100개 +이미지만 누적할 수 있습니다. 200 MiB 이상의 요청을 +허용하도록 Nginx 한도를 올리지 마세요. 프론트엔드는 이미지를 압축한 뒤 파일 +한 개씩 전송해야 합니다. + +운영 Nginx 설정은 이 저장소에서 관리되지 않으므로, 기존 +`location = /api/image/upload/blob` 블록 안에서 다음 스니펫을 include합니다. + +```nginx +include /배포경로/deploy/nginx/ado2-image-upload-limit.conf; +``` + +기존 설정이 prefix location만 사용한다면 그 블록의 `proxy_pass` 및 헤더 설정을 +그대로 유지한 채, exact location을 추가하고 동일한 프록시 설정을 적용해야 +합니다. 반영 전후에 실제 로드된 설정과 문법을 확인합니다. + +```bash +sudo nginx -T | grep -n -E 'server_name|image/upload/blob|client_max_body_size' +sudo nginx -t +sudo systemctl reload nginx +``` + +`proxy_request_buffering off`는 이 스니펫에 포함하지 않았습니다. 이 옵션만으로 +FastAPI의 multipart 파싱이 Azure 청크 스트리밍으로 바뀌지는 않으며, 느린 +클라이언트 연결이 애플리케이션을 직접 점유하는 시간이 늘어날 수 있습니다. + +Compose로 API를 실행하는 서버에서는 리소스 override를 함께 적용합니다. +이 override는 API 포트를 기본적으로 `127.0.0.1:8000`에만 바인딩해 외부 +클라이언트가 Nginx의 요청 크기 제한을 우회하지 못하게 합니다. 운영 Nginx가 +별도 컨테이너라면 호스트 포트를 공개하는 대신 두 서비스를 같은 내부 Docker +네트워크에 연결하세요. 부득이하게 `APP_BIND_ADDRESS`를 바꿀 때도 방화벽에서 +8000 포트의 외부 접근을 차단해야 합니다. `!override` 구문을 위해 Docker +Compose 2.24.4 이상이 필요합니다. + +```bash +docker compose -f docker-compose.yml -f compose.resources.yaml config --quiet +docker compose -f docker-compose.yml -f compose.resources.yaml up -d --force-recreate app +docker inspect castad-app \ + --format 'memory={{.HostConfig.Memory}} reservation={{.HostConfig.MemoryReservation}} swap={{.HostConfig.MemorySwap}}' +``` + +기본값은 hard limit 2 GiB, reservation 512 MiB이며 추가 swap은 허용하지 +않습니다. 호스트 용량과 실제 렌더링 부하를 측정한 뒤 +`APP_MEMORY_LIMIT`/`APP_MEMORY_RESERVATION`으로 조정할 수 있습니다. 예를 들어 +`APP_MEMORY_LIMIT=3g`를 설정하면 hard limit와 swap limit가 함께 3 GiB로 +변경됩니다. + +주의: 현재 저장소의 Dockerfile은 Uvicorn을 실행하지만 운영 로그 파일명에는 +Gunicorn이 나타납니다. 운영 프로세스가 호스트의 systemd/Gunicorn으로 직접 +실행 중이라면 이 Compose 제한은 적용되지 않습니다. 배포 전에 실제 실행 +주체를 확인하고, Compose 컨테이너가 아니라면 Gunicorn을 loopback 또는 Unix +socket에만 bind하고 해당 서비스 관리자의 메모리 제한을 별도로 설정해야 +합니다. 외부에서 앱 포트로 직접 접근할 수 있으면 Nginx의 25 MiB 제한을 +우회할 수 있습니다. + ## API 문서 서버 실행 후 `/docs` 에서 Scalar API 문서를 확인할 수 있습니다. @@ -277,4 +337,4 @@ fastapi run main.py │ │ │ │ ``` -testAc \ No newline at end of file +testAc diff --git a/app/home/api/routers/v1/home.py b/app/home/api/routers/v1/home.py index 97eefa7..950c1ee 100644 --- a/app/home/api/routers/v1/home.py +++ b/app/home/api/routers/v1/home.py @@ -1,13 +1,14 @@ +import asyncio import json +import secrets import time -from datetime import date +from collections.abc import AsyncIterator from pathlib import Path from typing import Literal, Optional from pydantic import BaseModel from urllib.parse import unquote, urlparse -import aiofiles from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile, status from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.ext.asyncio import AsyncSession @@ -25,14 +26,29 @@ from app.home.schemas.home_schema import ( CrawlingResponse, ErrorResponse, ImageUploadResponse, - ImageUploadResultItem, ImageUrlItem, ManualMarketingRequest, ProcessedInfo, # MarketingAnalysis, ) from app.home.services.naver_search import naver_search_client -from app.utils.upload_blob_as_request import AzureBlobUploader +from app.home.services.image_upload import ( + ALLOWED_IMAGE_EXTENSIONS, + BlobReferenceState, + ImageUploadLockTimeoutError, + assert_continuation_owner as _assert_continuation_owner, + compensate_failed_upload_blobs, + image_result_item as _image_result_item, + image_upload_task_lock, + inspect_upload_file as _inspect_upload_file, + is_valid_image_extension as _is_valid_image_extension, + normalize_continuation_task_id as _normalize_continuation_task_id, + validate_task_image_count, +) +from app.utils.upload_blob_as_request import ( + AzureBlobUploader, + BlobUploadTooLargeError, +) from app.utils.prompts.chatgpt_prompt import ChatgptService, ChatGPTResponseError from app.utils.common import generate_task_id from app.utils.logger import get_logger @@ -43,13 +59,36 @@ from app.utils.address_parser import extract_region_from_address from app.utils.autotag import autotag_images from app.utils.image_filter import filter_marketing_images, assemble_images from app.video.services.video import get_image_tags_by_task_id -from config import MEDIA_ROOT +from config import azure_blob_settings logger = get_logger("home") router = APIRouter() +async def _continuation_image_upload_lock( + task_id: Optional[str] = Form(default=None), + current_user: User = Depends(get_current_user), +) -> AsyncIterator[None]: + """continuation 요청 전체를 task 단위로 직렬화하는 yield dependency입니다.""" + # 인증 완료 자체가 lock 획득의 전제이며 endpoint와 dependency cache를 공유합니다. + del current_user + requested_task_id = task_id.strip() if task_id and task_id.strip() else None + if requested_task_id is None: + yield + return + + normalized_task_id = _normalize_continuation_task_id(requested_task_id) + try: + async with image_upload_task_lock(normalized_task_id): + yield + except ImageUploadLockTimeoutError: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="동일 이미지 작업이 처리 중입니다. 잠시 후 다시 시도해주세요.", + ) + + @router.get( "/search/accommodation", summary="장소 자동완성 검색 (숙박/음식점 등)", @@ -92,13 +131,23 @@ async def search_accommodation( ) -def _extract_region_from_address(road_address: str | None, jibun_address: str | None = None) -> str: +def _extract_region_from_address( + road_address: str | None, jibun_address: str | None = None +) -> str: return extract_region_from_address(road_address, jibun_address) class _IndustryOutput(BaseModel): industry: Literal[ - "stay", "restaurant", "cafe", "salon", "clinic", "fitness", "academy", "attraction", "general" + "stay", + "restaurant", + "cafe", + "salon", + "clinic", + "fitness", + "academy", + "attraction", + "general", ] @@ -164,10 +213,11 @@ async def _resolve_industry(category: str, customer_name: str = "") -> str: tags=["Crawling"], ) async def crawling( - request_body: CrawlingRequest, - session: AsyncSession = Depends(get_session)): + request_body: CrawlingRequest, session: AsyncSession = Depends(get_session) +): return await _crawling_logic(request_body.url, session) + @router.post( "/autocomplete", summary="네이버 자동완성 크롤링", @@ -200,14 +250,13 @@ async def crawling( tags=["Crawling"], ) async def autocomplete_crawling( - request_body: AutoCompleteRequest, - session: AsyncSession = Depends(get_session)): + request_body: AutoCompleteRequest, session: AsyncSession = Depends(get_session) +): url = await _autocomplete_logic(request_body.model_dump()) return await _crawling_logic(url, session) -async def _crawling_logic( - url:str, - session: AsyncSession): + +async def _crawling_logic(url: str, session: AsyncSession): request_start = time.perf_counter() logger.info("[crawling] ========== START ==========") logger.info(f"[crawling] URL: {url[:80]}...") @@ -318,14 +367,17 @@ async def _crawling_logic( extra_pass_flags = await filter_marketing_images( [img["original"] for img in extra_photo_urls], industry ) - except Exception as e: + except Exception: # logger.error(f"[crawling] Step 3 FAILED - 이미지 필터링 중 오류: {e}") raise HTTPException( status_code=status.HTTP_502_BAD_GATEWAY, detail="이미지 마케팅 적합성 필터링 중 오류가 발생했습니다.", ) scraper.image_link_list = assemble_images( - owner_images, extra_photo_urls, extra_pass_flags, NvMapScraper.MAX_IMAGES + owner_images, + extra_photo_urls, + extra_pass_flags, + NvMapScraper.MAX_IMAGES, ) passed_count = sum(extra_pass_flags) step3_elapsed = (time.perf_counter() - step3_start) * 1000 @@ -335,7 +387,9 @@ async def _crawling_logic( f"({step3_elapsed:.1f}ms)" ) if not scraper.image_link_list: - logger.warning("[crawling] Step 3 - 필터링 후 사용 가능 이미지가 0장입니다.") + logger.warning( + "[crawling] Step 3 - 필터링 후 사용 가능 이미지가 0장입니다." + ) # ========== Step 4: ChatGPT 마케팅 분석 ========== step4_start = time.perf_counter() @@ -376,7 +430,9 @@ async def _crawling_logic( await session.commit() await session.refresh(marketing_intel) m_id = marketing_intel.id - logger.debug(f"[MarketingPrompt] INSERT place_id={marketing_intel.place_id} id={marketing_intel.id}") + logger.debug( + f"[MarketingPrompt] INSERT place_id={marketing_intel.place_id} id={marketing_intel.id}" + ) step4_elapsed = (time.perf_counter() - step4_start) * 1000 logger.info( @@ -419,17 +475,16 @@ async def _crawling_logic( logger.info(f"[crawling] - Step 4 (GPT 분석): {step4_elapsed:.1f}ms") return { - "status": gpt_status if 'gpt_status' in locals() else "completed", + "status": gpt_status if "gpt_status" in locals() else "completed", "image_list": scraper.image_link_list, "image_count": len(scraper.image_link_list) if scraper.image_link_list else 0, "processed_info": processed_info, "marketing_analysis": marketing_analysis, "m_id": m_id, - "industry": industry if 'industry' in locals() else "", + "industry": industry if "industry" in locals() else "", } - @router.post( "/marketing", summary="업체명+주소 직접 입력 마케팅 분석", @@ -515,7 +570,7 @@ async def manual_marketing( ) -async def _autocomplete_logic(autocomplete_item:dict): +async def _autocomplete_logic(autocomplete_item: dict): step1_start = time.perf_counter() try: async with NvMapPwScraper() as pw_scraper: @@ -543,6 +598,7 @@ async def _autocomplete_logic(autocomplete_item:dict): return new_url + def _extract_image_name(url: str, index: int) -> str: """URL에서 이미지 이름 추출 또는 기본 이름 생성""" try: @@ -555,30 +611,6 @@ def _extract_image_name(url: str, index: int) -> str: return f"image_{index + 1:03d}" -ALLOWED_IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".webp", ".heic", ".heif"} - - -def _is_valid_image_extension(filename: str | None) -> bool: - """파일명의 확장자가 유효한 이미지 확장자인지 확인""" - if not filename: - return False - ext = Path(filename).suffix.lower() - return ext in ALLOWED_IMAGE_EXTENSIONS - - -def _get_file_extension(filename: str) -> str: - """파일명에서 확장자 추출 (소문자)""" - return Path(filename).suffix.lower() - - -async def _save_upload_file(file: UploadFile, save_path: Path) -> None: - """업로드 파일을 지정된 경로에 저장""" - save_path.parent.mkdir(parents=True, exist_ok=True) - async with aiofiles.open(save_path, "wb") as f: - content = await file.read() - await f.write(content) - - IMAGES_JSON_EXAMPLE = """[ {"url": "https://naverbooking-phinf.pstatic.net/20240514_189/1715688030436xT14o_JPEG/1.jpg"}, {"url": "https://naverbooking-phinf.pstatic.net/20240514_48/1715688030574wTtQd_JPEG/2.jpg"}, @@ -587,12 +619,13 @@ IMAGES_JSON_EXAMPLE = """[ {"url": "https://naverbooking-phinf.pstatic.net/20240514_259/17156880311809wCnY_JPEG/5.jpg", "name": "외관"} ]""" + @router.post( "/image/upload/blob", summary="이미지 업로드 (Azure Blob Storage)", description=""" -이미지를 Azure Blob Storage에 업로드하고 새로운 task_id를 생성합니다. -바이너리 파일은 로컬 서버에 저장하지 않고 Azure Blob에 직접 업로드됩니다. +이미지를 Azure Blob Storage에 업로드하고 task_id를 생성하거나 기존 작업에 이어 붙입니다. +바이너리 파일은 로컬 서버 경로에 복사하지 않고 Azure Blob에 청크 업로드됩니다. ## 인증 **Bearer 토큰 필수** - `Authorization: Bearer {access_token}` 헤더를 포함해야 합니다. @@ -603,8 +636,16 @@ multipart/form-data 형식으로 전송합니다. ## 요청 필드 - **images_json**: 외부 이미지 URL 목록 (JSON 문자열, 선택) - **files**: 이미지 바이너리 파일 목록 (선택) +- **task_id**: 분할 업로드를 이어갈 기존 task_id (선택, UUID7) +- **finalize**: 누적 이미지 태깅 실행 여부 (기본값 true) -**주의**: images_json 또는 files 중 최소 하나는 반드시 전달해야 합니다. +**주의**: +- 기존 단일 요청은 `finalize=true` 기본값으로 이전과 동일하게 동작합니다. +- 분할 업로드 첫 요청은 `finalize=false`와 최소 1개 파일을 보내고, 응답 task_id를 다음 요청에 전달합니다. +- 중간 요청은 `task_id`와 `finalize=false`, 마지막 요청만 `finalize=true`로 보냅니다. +- 파일 1개는 최대 15 MiB, 한 요청의 파일 합계는 최대 20 MiB입니다. +- 한 task에는 기본 최대 100개의 이미지를 누적할 수 있습니다(서버 설정 가능). +- `finalize=true`일 때 해당 task_id에 누적된 전체 이미지를 한 번에 태깅합니다. ## 지원 이미지 확장자 jpg, jpeg, png, webp, heic, heif @@ -632,6 +673,19 @@ curl -X POST "http://localhost:8000/image/upload/blob" \\ -H "Authorization: Bearer {access_token}" \\ -F 'images_json=[{"url":"https://example.com/image.jpg"}]' \\ -F "files=@/path/to/local_image.jpg" + +# 분할 업로드 첫 요청 (응답의 task_id 보관) +curl -X POST "http://localhost:8000/image/upload/blob" \ + -H "Authorization: Bearer {access_token}" \ + -F "files=@/path/to/image1.jpg" \ + -F "finalize=false" + +# 분할 업로드 마지막 요청 +curl -X POST "http://localhost:8000/image/upload/blob" \ + -H "Authorization: Bearer {access_token}" \ + -F "task_id={task_id}" \ + -F "files=@/path/to/image2.jpg" \ + -F "finalize=true" ``` ## 반환 정보 @@ -645,14 +699,17 @@ curl -X POST "http://localhost:8000/image/upload/blob" \\ - **image_urls**: Image 테이블에 저장된 현재 task_id의 이미지 URL 목록 ## 저장 경로 -- 바이너리 파일: Azure Blob Storage ({BASE_URL}/{task_id}/image/{파일명}) +- 바이너리 파일: Azure Blob Storage ({BASE_URL}/{user_uuid}/{task_id}/image/{파일명}) - URL 이미지: 외부 URL 그대로 Image 테이블에 저장 """, response_model=ImageUploadResponse, responses={ 200: {"description": "이미지 업로드 성공"}, - 400: {"description": "이미지가 제공되지 않음", "model": ErrorResponse}, + 400: {"description": "입력 이미지가 유효하지 않음", "model": ErrorResponse}, 401: {"description": "인증 실패 (토큰 없음/만료)"}, + 403: {"description": "continuation task 소유권 검증 실패"}, + 413: {"description": "파일 또는 요청 크기 제한 초과"}, + 502: {"description": "Azure Blob 업로드 실패"}, }, tags=["Image-Blob"], openapi_extra={ @@ -675,11 +732,20 @@ async def upload_images_blob( default=None, description="이미지 바이너리 파일 목록", ), + task_id: Optional[str] = Form( + default=None, + description="분할 업로드를 이어갈 기존 task_id (UUID7)", + ), + finalize: bool = Form( + default=True, + description="true일 때 누적 이미지 태깅을 실행해 업로드를 완료", + ), industry: str = Form( default="", description="업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction|general). 크롤링 응답의 industry 값을 그대로 전달", ), current_user: User = Depends(get_current_user), + _upload_lock: None = Depends(_continuation_image_upload_lock), ) -> ImageUploadResponse: """이미지 업로드 (URL + Azure Blob Storage) @@ -688,108 +754,249 @@ async def upload_images_blob( - Stage 2: Azure Blob 업로드 (세션 없음) - Stage 3: DB 저장 (새 세션으로 빠르게 처리) """ + del _upload_lock request_start = time.perf_counter() + requested_task_id = task_id.strip() if task_id and task_id.strip() else None + is_continuation = requested_task_id is not None - # task_id 생성 - task_id = await generate_task_id() - logger.info(f"[upload_images_blob] START - task_id: {task_id}") + if requested_task_id: + task_id = _normalize_continuation_task_id(requested_task_id) + async with AsyncSessionLocal() as session: + existing_result = await session.execute( + select(Image) + .where( + Image.task_id == task_id, + Image.is_deleted.is_(False), + ) + .order_by(Image.img_order, Image.id) + ) + existing_images = list(existing_result.scalars().all()) + _assert_continuation_owner( + existing_images, + current_user.user_uuid, + task_id, + ) + else: + task_id = await generate_task_id() + existing_images = [] - # ========== Stage 1: 입력 검증 및 파일 데이터 준비 (세션 없음) ========== + logger.info( + f"[upload_images_blob] START - task_id: {task_id}, " + f"continuation: {is_continuation}, finalize: {finalize}" + ) + + # ========== Stage 1: 입력 검증 (세션 없음, 파일 전체 메모리 적재 없음) ========== has_images_json = images_json is not None and images_json.strip() != "" has_files = files is not None and len(files) > 0 - - if not has_images_json and not has_files: + if not has_images_json and not has_files and not (is_continuation and finalize): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="images_json 또는 files 중 하나는 반드시 제공해야 합니다.", ) - # images_json 파싱 url_images: list[ImageUrlItem] = [] if has_images_json and images_json: try: parsed = json.loads(images_json) - if isinstance(parsed, list): - url_images = [ImageUrlItem(**item) for item in parsed if item] + if not isinstance(parsed, list): + raise ValueError("JSON 최상위 값은 배열이어야 합니다.") + url_images = [ImageUrlItem(**item) for item in parsed if item] except (json.JSONDecodeError, TypeError, ValueError) as e: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"images_json 파싱 오류: {str(e)}", ) - # 유효한 파일만 필터링 및 파일 내용 미리 읽기 - valid_files_data: list[tuple[str, str, bytes]] = [] # (original_name, ext, content) - skipped_files: list[str] = [] - if has_files and files: - for f in files: - is_valid_ext = _is_valid_image_extension(f.filename) - is_not_empty = f.size is None or f.size > 0 - is_real_file = f.filename and f.filename != "filename" - - if f and is_real_file and is_valid_ext and is_not_empty: - # 파일 내용을 미리 읽어둠 - content = await f.read() - ext = _get_file_extension(f.filename) # type: ignore[arg-type] - valid_files_data.append((f.filename or "image", ext, content)) - else: - skipped_files.append(f.filename or "unknown") - - if not url_images and not valid_files_data: - detail = ( - f"유효한 이미지가 없습니다. " - f"지원 확장자: {', '.join(ALLOWED_IMAGE_EXTENSIONS)}. " - f"건너뛴 파일: {skipped_files}" + upload_files = files or [] + declared_total_size = sum( + file.size for file in upload_files if file.size is not None and file.size > 0 + ) + max_request_size = azure_blob_settings.IMAGE_UPLOAD_MAX_REQUEST_SIZE_BYTES + if declared_total_size > max_request_size: + raise HTTPException( + status_code=status.HTTP_413_CONTENT_TOO_LARGE, + detail=( + "한 요청의 파일 합계가 최대 크기 " + f"{max_request_size // (1024 * 1024)} MiB를 초과합니다." + ), ) + + valid_files_data: list[tuple[UploadFile, str, str, int]] = [] + skipped_files: list[str] = [] + actual_total_size = 0 + for file in upload_files: + is_real_file = bool(file.filename and file.filename != "filename") + if not is_real_file or not _is_valid_image_extension(file.filename): + skipped_files.append(file.filename or "unknown") + continue + + original_name, extension, actual_size = await _inspect_upload_file(file) + actual_total_size += actual_size + if actual_total_size > max_request_size: + raise HTTPException( + status_code=status.HTTP_413_CONTENT_TOO_LARGE, + detail=( + "한 요청의 실제 파일 합계가 최대 크기 " + f"{max_request_size // (1024 * 1024)} MiB를 초과합니다." + ), + ) + valid_files_data.append((file, original_name, extension, actual_size)) + + provided_new_input = has_images_json or has_files + if provided_new_input and not url_images and not valid_files_data: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, - detail=detail, + detail=( + "유효한 이미지가 없습니다. " + f"지원 확장자: {', '.join(sorted(ALLOWED_IMAGE_EXTENSIONS))}. " + f"건너뛴 파일: {skipped_files}" + ), ) + if not is_continuation and not finalize and not valid_files_data: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="분할 업로드의 첫 요청에는 소유권 확인용 이미지 파일이 필요합니다.", + ) + + validate_task_image_count( + existing_count=len(existing_images), + incoming_count=len(url_images) + len(valid_files_data), + ) stage1_time = time.perf_counter() logger.info( f"[upload_images_blob] Stage 1 done - urls: {len(url_images)}, " - f"files: {len(valid_files_data)}, " + f"files: {len(valid_files_data)}, bytes: {actual_total_size}, " f"elapsed: {(stage1_time - request_start) * 1000:.1f}ms" ) - # ========== Stage 2: Azure Blob 업로드 (세션 없음) ========== - # 업로드 결과를 저장할 리스트 (나중에 DB에 저장) - blob_upload_results: list[tuple[str, str]] = [] # (img_name, blob_url) - img_order = len(url_images) # URL 이미지 다음 순서부터 시작 + # ========== Stage 2: Azure Blob 청크 업로드 (세션 없음) ========== + # (원본명, 공개 URL, Azure 저장 파일명) + blob_upload_results: list[tuple[str, str, str]] = [] + uploader: AzureBlobUploader | None = None + order_hint = max((image.img_order for image in existing_images), default=-1) + 1 + order_hint += len(url_images) + + async def cleanup_current_request_blobs() -> None: + if uploader is None: + return + for _, _, stored_name in blob_upload_results: + await uploader.delete_image(stored_name) + + async def compensate_failed_db_write( + commit_started: bool, + error: BaseException, + ) -> None: + """DB 반영 여부가 불명확하면 Blob 보존을 우선합니다.""" + if not blob_upload_results: + return + + blob_urls = {blob_url for _, blob_url, _ in blob_upload_results} + reference_state = await compensate_failed_upload_blobs( + task_id=task_id, + blob_urls=blob_urls, + commit_started=commit_started, + cleanup=cleanup_current_request_blobs, + ) + if not commit_started: + return + if reference_state == BlobReferenceState.NONE: + logger.warning( + f"[upload_images_blob] Commit failed and independent DB check " + f"confirmed no Blob references; cleaning up - task_id: {task_id}" + ) + return + + logger.error( + f"[upload_images_blob] Preserving Blob after ambiguous commit - " + f"task_id: {task_id}, reference_state: {reference_state}, " + f"error: {type(error).__name__}: {error}" + ) + + async def compensate_after_cancellation( + commit_started: bool, + error: asyncio.CancelledError, + ) -> None: + compensation_task = asyncio.create_task( + compensate_failed_db_write(commit_started, error) + ) + try: + await asyncio.shield(compensation_task) + except asyncio.CancelledError: + # 반복 취소가 와도 확인/정리가 끝나도록 하며, 실패 시에는 Blob을 보존합니다. + try: + await compensation_task + except BaseException as compensation_error: + logger.error( + f"[upload_images_blob] Cancellation compensation failed; " + f"preserving Blob - task_id: {task_id}, " + f"{type(compensation_error).__name__}: {compensation_error}" + ) if valid_files_data: uploader = AzureBlobUploader(user_uuid=current_user.user_uuid, task_id=task_id) total_files = len(valid_files_data) - for idx, (original_name, ext, file_content) in enumerate(valid_files_data): - name_without_ext = ( - original_name.rsplit(".", 1)[0] - if "." in original_name - else original_name + for idx, (file, original_name, extension, actual_size) in enumerate( + valid_files_data + ): + name_without_ext = Path(original_name).stem + unique_suffix = secrets.token_hex(4) + stored_name = ( + f"{name_without_ext}_{order_hint + idx:03d}_{unique_suffix}{extension}" ) - filename = f"{name_without_ext}_{img_order:03d}{ext}" - logger.debug( f"[upload_images_blob] Uploading file {idx + 1}/{total_files}: " - f"{filename} ({len(file_content)} bytes)" + f"{stored_name} ({actual_size} bytes)" ) - # Azure Blob Storage에 직접 업로드 - upload_success = await uploader.upload_image_bytes(file_content, filename) + try: + upload_success = await uploader.upload_image_stream( + file, + stored_name, + expected_size_bytes=actual_size, + max_size_bytes=( + azure_blob_settings.IMAGE_UPLOAD_MAX_FILE_SIZE_BYTES + ), + ) + except BlobUploadTooLargeError as exc: + await cleanup_current_request_blobs() + raise HTTPException( + status_code=status.HTTP_413_CONTENT_TOO_LARGE, + detail=( + f"파일 '{original_name}'이 최대 크기 " + f"{exc.max_size_bytes // (1024 * 1024)} MiB를 초과합니다." + ), + ) + except asyncio.CancelledError: + await asyncio.shield(cleanup_current_request_blobs()) + raise if upload_success: - blob_url = uploader.public_url - blob_upload_results.append((original_name, blob_url)) - img_order += 1 + blob_upload_results.append( + (original_name, uploader.public_url, stored_name) + ) logger.debug( f"[upload_images_blob] File {idx + 1}/{total_files} SUCCESS" ) else: - skipped_files.append(filename) + skipped_files.append(stored_name) logger.warning( f"[upload_images_blob] File {idx + 1}/{total_files} FAILED" ) + if valid_files_data and not blob_upload_results and not url_images: + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, + detail="Azure Blob Storage 이미지 업로드에 실패했습니다.", + ) + if not is_continuation and not finalize and not blob_upload_results: + # URL row만 남으면 다음 요청에서 소유권을 증명할 수 없으므로 저장하지 않습니다. + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, + detail="분할 업로드를 시작할 이미지 파일 업로드에 실패했습니다.", + ) + stage2_time = time.perf_counter() logger.info( f"[upload_images_blob] Stage 2 done - blob uploads: " @@ -799,15 +1006,49 @@ async def upload_images_blob( # ========== Stage 3: DB 저장 (새 세션으로 빠르게 처리) ========== logger.info("[upload_images_blob] Stage 3 starting - DB save...") - result_images: list[ImageUploadResultItem] = [] - img_order = 0 + all_images: list[Image] = [] + commit_started = False try: async with AsyncSessionLocal() as session: - # URL 이미지 저장 + # 같은 task의 append 요청을 직렬화해 img_order 충돌을 줄입니다. + locked_result = await session.execute( + select(Image) + .where( + Image.task_id == task_id, + Image.is_deleted.is_(False), + ) + .order_by(Image.img_order, Image.id) + .with_for_update() + ) + locked_images = list(locked_result.scalars().all()) + if is_continuation: + _assert_continuation_owner( + locked_images, + current_user.user_uuid, + task_id, + ) + elif locked_images: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="이미 사용 중인 task_id가 생성되었습니다. 다시 시도해주세요.", + ) + + validate_task_image_count( + existing_count=len(locked_images), + incoming_count=len(url_images) + len(blob_upload_results), + ) + + img_order = ( + max( + (image.img_order for image in locked_images), + default=-1, + ) + + 1 + ) + new_images: list[Image] = [] for url_item in url_images: img_name = url_item.name or _extract_image_name(url_item.url, img_order) - image = Image( task_id=task_id, img_name=img_name, @@ -815,21 +1056,10 @@ async def upload_images_blob( img_order=img_order, ) session.add(image) - await session.flush() - - result_images.append( - ImageUploadResultItem( - id=image.id, - img_name=img_name, - img_url=url_item.url, - img_order=img_order, - source="url", - ) - ) + new_images.append(image) img_order += 1 - # Blob 업로드 결과 저장 - for img_name, blob_url in blob_upload_results: + for img_name, blob_url, _ in blob_upload_results: image = Image( task_id=task_id, img_name=img_name, @@ -837,28 +1067,31 @@ async def upload_images_blob( img_order=img_order, ) session.add(image) - await session.flush() - - result_images.append( - ImageUploadResultItem( - id=image.id, - img_name=img_name, - img_url=blob_url, - img_order=img_order, - source="blob", - ) - ) + new_images.append(image) img_order += 1 + await session.flush() + all_images = sorted( + [*locked_images, *new_images], + key=lambda image: (image.img_order, image.id), + ) + commit_started = True await session.commit() stage3_time = time.perf_counter() logger.info( f"[upload_images_blob] Stage 3 done - " - f"saved: {len(result_images)}, " + f"task total: {len(all_images)}, added: {len(new_images)}, " f"elapsed: {(stage3_time - stage2_time) * 1000:.1f}ms" ) + except asyncio.CancelledError as e: + await compensate_after_cancellation(commit_started, e) + raise + except HTTPException as e: + await compensate_failed_db_write(commit_started, e) + raise except SQLAlchemyError as e: + await compensate_failed_db_write(commit_started, e) logger.error(f"[upload_images_blob] DB Error - task_id: {task_id}, error: {e}") logger.exception("[upload_images_blob] DB 상세 오류:") raise HTTPException( @@ -866,6 +1099,7 @@ async def upload_images_blob( detail="이미지 저장 중 데이터베이스 오류가 발생했습니다.", ) except Exception as e: + await compensate_failed_db_write(commit_started, e) logger.error( f"[upload_images_blob] Stage 3 EXCEPTION - " f"task_id: {task_id}, error: {type(e).__name__}: {e}" @@ -876,24 +1110,31 @@ async def upload_images_blob( detail="이미지 업로드 중 오류가 발생했습니다.", ) + result_images = [_image_result_item(image) for image in all_images] saved_count = len(result_images) image_urls = [img.img_url for img in result_images] - logger.info(f"[image_tagging] START - task_id: {task_id}") - await tagging_images(image_urls, industry=industry, clear_old_tags=True) - logger.info(f"[image_tagging] Done - task_id: {task_id}") + if finalize: + logger.info(f"[image_tagging] START - task_id: {task_id}") + await tagging_images(image_urls, industry=industry, clear_old_tags=True) + logger.info(f"[image_tagging] Done - task_id: {task_id}") - # 태깅 직후 영상 생성에 사용 가능한 이미지가 하나도 없으면 조기에 실패시킨다. - # (여기서 걸러지지 않으면 훨씬 나중인 영상 생성 단계에서야 슬롯 미배정으로 발견됨) - # marketing_acceptable 필터링은 크롤링 단계에서 이미 완료되었으므로 여기서는 재필터링하지 않는다. - taged_image_list = await get_image_tags_by_task_id(task_id) - logger.info(f"태깅된 이미지: {len(taged_image_list)}개 - task_id: {task_id}") - if not taged_image_list: - logger.error(f"[image_tagging] 영상 생성에 적합한 이미지가 없음 - task_id: {task_id}") - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail="영상 생성에 적합한 이미지가 없습니다. 다른 이미지로 다시 업로드해주세요.", - ) + # 마지막 분할 요청에서 누적된 전체 이미지의 적합성을 확인합니다. + taged_image_list = await get_image_tags_by_task_id(task_id) + logger.info(f"태깅된 이미지: {len(taged_image_list)}개 - task_id: {task_id}") + if not taged_image_list: + logger.error( + f"[image_tagging] 영상 생성에 적합한 이미지가 없음 - task_id: {task_id}" + ) + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=( + "영상 생성에 적합한 이미지가 없습니다. " + "다른 이미지로 다시 업로드해주세요." + ), + ) + else: + logger.info(f"[image_tagging] DEFERRED - task_id: {task_id}") total_time = time.perf_counter() - request_start logger.info( @@ -903,9 +1144,9 @@ async def upload_images_blob( return ImageUploadResponse( task_id=task_id, - total_count=len(result_images), - url_count=len(url_images), - file_count=len(blob_upload_results), + total_count=saved_count, + url_count=sum(image.source == "url" for image in result_images), + file_count=sum(image.source == "blob" for image in result_images), saved_count=saved_count, images=result_images, image_urls=image_urls, @@ -913,10 +1154,8 @@ async def upload_images_blob( async def tagging_images( - image_urls : list[str], - industry: str = "", - clear_old_tags : bool = False - ) -> None: + image_urls: list[str], industry: str = "", clear_old_tags: bool = False +) -> None: # 1. 조회 async with AsyncSessionLocal() as session: stmt = ( @@ -938,9 +1177,11 @@ async def tagging_images( session.add_all(new_imt) null_imts = [imt for imt in image_tags if imt.img_tag is None] + new_imt await session.commit() - + if null_imts: - tag_datas = await autotag_images([img.img_url for img in null_imts], industry=industry) + tag_datas = await autotag_images( + [img.img_url for img in null_imts], industry=industry + ) # print(tag_datas) async with AsyncSessionLocal() as session: diff --git a/app/home/services/image_upload.py b/app/home/services/image_upload.py new file mode 100644 index 0000000..a1ea280 --- /dev/null +++ b/app/home/services/image_upload.py @@ -0,0 +1,380 @@ +"""이미지 업로드 입력 검증과 continuation 소유권 검사 유틸리티.""" + +import asyncio +import time +from collections.abc import AsyncIterator, Awaitable, Callable +from contextlib import asynccontextmanager +from enum import StrEnum +from pathlib import Path +from typing import Literal +from uuid import UUID + +from fastapi import HTTPException, UploadFile, status +from sqlalchemy import select, text + +from app.database.session import AsyncSessionLocal, engine +from app.home.models import Image +from app.home.schemas.home_schema import ImageUploadResultItem +from app.utils.logger import get_logger +from config import azure_blob_settings + +logger = get_logger("image_upload") +_image_upload_lock_slots = asyncio.Semaphore( + azure_blob_settings.IMAGE_UPLOAD_MAX_CONCURRENT_LOCKS +) + +ALLOWED_IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".webp", ".heic", ".heif"} +_HEIF_BRANDS = { + b"heic", + b"heix", + b"hevc", + b"hevx", + b"heim", + b"heis", + b"mif1", + b"msf1", +} +_IMAGE_SIGNATURE_BYTES = 64 + + +class ImageUploadLockTimeoutError(TimeoutError): + """동일 task 이미지 변경 락을 제한 시간 내 획득하지 못했습니다.""" + + +class BlobReferenceState(StrEnum): + """불명확한 DB commit 뒤 Blob URL 참조 확인 결과.""" + + ALL = "all" + NONE = "none" + MIXED = "mixed" + UNKNOWN = "unknown" + + +def should_cleanup_failed_upload_blobs( + *, + commit_started: bool, + reference_state: BlobReferenceState = BlobReferenceState.UNKNOWN, +) -> bool: + """DB commit 시도 후에는 참조가 없다고 확정된 경우에만 Blob을 삭제합니다.""" + return not commit_started or reference_state == BlobReferenceState.NONE + + +def classify_blob_references( + expected_urls: set[str], found_urls: set[str] +) -> BlobReferenceState: + """예상 Blob URL과 독립 조회 결과를 보존 우선 상태로 분류합니다.""" + if not expected_urls: + return BlobReferenceState.NONE + matched_urls = expected_urls & found_urls + if matched_urls == expected_urls: + return BlobReferenceState.ALL + if not matched_urls: + return BlobReferenceState.NONE + return BlobReferenceState.MIXED + + +async def inspect_blob_references( + task_id: str, + blob_urls: set[str], +) -> BlobReferenceState: + """독립 세션에서 이번 요청 Blob URL의 DB 반영 여부를 확인합니다. + + commit 응답 유실 직후의 짧은 가시성 경합을 피하려고 NONE 결과만 세 번 + 재확인합니다. 존재/혼재/조회 실패는 즉시 보존 쪽으로 판정합니다. + """ + if not blob_urls: + return BlobReferenceState.NONE + for attempt in range(3): + try: + async with AsyncSessionLocal() as session: + result = await session.execute( + select(Image.img_url).where( + Image.task_id == task_id, + Image.img_url.in_(blob_urls), + ) + ) + found_urls = set(result.scalars().all()) + state = classify_blob_references(blob_urls, found_urls) + if state != BlobReferenceState.NONE: + return state + if attempt < 2: + await asyncio.sleep(0.1 * (attempt + 1)) + except Exception as exc: + logger.error( + f"[inspect_blob_references] DB verification failed - task_id: " + f"{task_id}, {type(exc).__name__}: {exc}" + ) + return BlobReferenceState.UNKNOWN + return BlobReferenceState.NONE + + +async def compensate_failed_upload_blobs( + *, + task_id: str, + blob_urls: set[str], + commit_started: bool, + cleanup: Callable[[], Awaitable[None]], +) -> BlobReferenceState: + """DB 쓰기 실패 후 안전하다고 확인된 Blob만 보상 삭제합니다.""" + reference_state = BlobReferenceState.UNKNOWN + if commit_started: + reference_state = await inspect_blob_references(task_id, blob_urls) + + if should_cleanup_failed_upload_blobs( + commit_started=commit_started, + reference_state=reference_state, + ): + await cleanup() + return reference_state + + +def validate_task_image_count(*, existing_count: int, incoming_count: int) -> int: + """한 task에 누적 가능한 이미지 수를 검증하고 예상 총 개수를 반환합니다.""" + total_count = existing_count + incoming_count + max_task_images = azure_blob_settings.IMAGE_UPLOAD_MAX_TASK_IMAGES + if total_count > max_task_images: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"한 작업에는 이미지를 최대 {max_task_images}개까지 추가할 수 있습니다.", + ) + return total_count + + +@asynccontextmanager +async def _image_upload_lock_slot( + lock_name: str, timeout_seconds: int +) -> AsyncIterator[None]: + """DB 락 연결 슬롯 대기에도 동일한 제한 시간을 적용합니다.""" + try: + await asyncio.wait_for( + _image_upload_lock_slots.acquire(), + timeout=timeout_seconds, + ) + except TimeoutError as exc: + raise ImageUploadLockTimeoutError(lock_name) from exc + + try: + yield + finally: + _image_upload_lock_slots.release() + + +@asynccontextmanager +async def image_upload_task_lock(task_id: str) -> AsyncIterator[None]: + """동일 task 요청 전체를 MySQL advisory lock으로 직렬화합니다. + + yield dependency가 Azure 업로드부터 최종 태깅까지 connection을 점유하므로, + worker별 semaphore로 main DB pool의 나머지 connection을 보존합니다. + """ + lock_name = f"image_upload:{task_id}" + timeout_seconds = azure_blob_settings.IMAGE_UPLOAD_LOCK_TIMEOUT_SECONDS + lock_started = time.perf_counter() + + async with _image_upload_lock_slot(lock_name, timeout_seconds): + async with engine.connect() as connection: + lock_result = await connection.execute( + text("SELECT GET_LOCK(:lock_name, :timeout_seconds)"), + { + "lock_name": lock_name, + "timeout_seconds": timeout_seconds, + }, + ) + if lock_result.scalar_one_or_none() != 1: + raise ImageUploadLockTimeoutError(lock_name) + logger.info( + f"[image_upload_task_lock] ACQUIRED - task_id: {task_id}, " + f"wait_ms: {(time.perf_counter() - lock_started) * 1000:.1f}" + ) + + try: + yield + finally: + release_task = asyncio.create_task( + connection.execute( + text("SELECT RELEASE_LOCK(:lock_name)"), + {"lock_name": lock_name}, + ) + ) + try: + release_result = await asyncio.shield(release_task) + if release_result.scalar_one_or_none() != 1: + raise RuntimeError(f"RELEASE_LOCK failed: {lock_name}") + except asyncio.CancelledError: + # shield 바깥 task가 다시 취소돼도 release query는 끝까지 기다립니다. + try: + await release_task + except BaseException as release_exc: + await connection.invalidate(release_exc) + raise + except BaseException as exc: + # 락이 남은 connection이 pool로 복귀하지 않도록 폐기합니다. + await connection.invalidate(exc) + logger.error( + f"[image_upload_task_lock] RELEASE_LOCK failed - " + f"{type(exc).__name__}: {exc}" + ) + else: + logger.info( + f"[image_upload_task_lock] RELEASED - task_id: {task_id}, " + f"held_ms: {(time.perf_counter() - lock_started) * 1000:.1f}" + ) + + +def is_valid_image_extension(filename: str | None) -> bool: + """파일명의 확장자가 지원 이미지 확장자인지 확인합니다.""" + if not filename: + return False + return Path(filename).suffix.lower() in ALLOWED_IMAGE_EXTENSIONS + + +def _detect_image_format(header: bytes) -> str | None: + """신뢰할 수 없는 파일명/MIME 대신 파일 시그니처로 형식을 판별합니다.""" + if header.startswith(b"\xff\xd8\xff"): + return "jpeg" + if header.startswith(b"\x89PNG\r\n\x1a\n"): + return "png" + if len(header) >= 12 and header.startswith(b"RIFF") and header[8:12] == b"WEBP": + return "webp" + if len(header) >= 12 and header[4:8] == b"ftyp": + brands = {header[8:12]} + brands.update( + header[index : index + 4] for index in range(16, len(header) - 3, 4) + ) + if brands & _HEIF_BRANDS: + return "heif" + return None + + +def _extension_matches_format(extension: str, detected_format: str) -> bool: + """동일 포맷의 별칭을 고려해 확장자와 시그니처 일치 여부를 확인합니다.""" + expected_formats = { + ".jpg": "jpeg", + ".jpeg": "jpeg", + ".png": "png", + ".webp": "webp", + ".heic": "heif", + ".heif": "heif", + } + return expected_formats.get(extension) == detected_format + + +async def inspect_upload_file(file: UploadFile) -> tuple[str, str, int]: + """UploadFile을 상수 메모리로 검사하고 실제 바이트 크기를 반환합니다.""" + original_name = file.filename or "" + if len(original_name) > 255: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="파일명은 255자를 초과할 수 없습니다.", + ) + + extension = Path(original_name).suffix.lower() + max_file_size = azure_blob_settings.IMAGE_UPLOAD_MAX_FILE_SIZE_BYTES + validation_chunk_size = min( + azure_blob_settings.AZURE_BLOB_UPLOAD_BLOCK_SIZE_BYTES, + 1024 * 1024, + ) + total_size = 0 + header = bytearray() + + await file.seek(0) + try: + while chunk := await file.read(validation_chunk_size): + total_size += len(chunk) + if total_size > max_file_size: + raise HTTPException( + status_code=status.HTTP_413_CONTENT_TOO_LARGE, + detail=( + f"파일 '{original_name}'이 최대 크기 " + f"{max_file_size // (1024 * 1024)} MiB를 초과합니다." + ), + ) + if len(header) < _IMAGE_SIGNATURE_BYTES: + remaining = _IMAGE_SIGNATURE_BYTES - len(header) + header.extend(chunk[:remaining]) + finally: + await file.seek(0) + + if total_size == 0: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"빈 파일은 업로드할 수 없습니다: {original_name}", + ) + + detected_format = _detect_image_format(bytes(header)) + if detected_format is None or not _extension_matches_format( + extension, detected_format + ): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=( + f"파일 내용과 확장자가 일치하는 지원 이미지가 아닙니다: {original_name}" + ), + ) + + return original_name, extension, total_size + + +def normalize_continuation_task_id(task_id: str) -> str: + """continuation task_id를 canonical UUID7 문자열로 검증합니다.""" + try: + parsed = UUID(task_id) + except (ValueError, AttributeError): + parsed = None + + if ( + parsed is None + or len(task_id) != 36 + or parsed.version != 7 + or str(parsed) != task_id.lower() + ): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="task_id는 올바른 UUID7 형식이어야 합니다.", + ) + return str(parsed) + + +def _blob_url_prefix(user_uuid: str, task_id: str) -> str: + """현재 사용자와 task에 허용된 Azure 이미지 URL prefix를 반환합니다.""" + base_url = azure_blob_settings.AZURE_BLOB_BASE_URL.rstrip("/") + return f"{base_url}/{user_uuid}/{task_id}/image/" + + +def assert_continuation_owner( + images: list[Image], user_uuid: str, task_id: str +) -> None: + """Image에 owner 컬럼이 없어 Blob 경로로 continuation 소유권을 검증합니다.""" + if not images: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="이어 올릴 이미지 작업을 찾을 수 없습니다.", + ) + + base_prefix = f"{azure_blob_settings.AZURE_BLOB_BASE_URL.rstrip('/')}/" + owner_prefix = _blob_url_prefix(user_uuid, task_id) + internal_urls = [ + image.img_url for image in images if image.img_url.startswith(base_prefix) + ] + + if not internal_urls or any( + not image_url.startswith(owner_prefix) for image_url in internal_urls + ): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="이 이미지 업로드 작업을 이어서 수정할 권한이 없습니다.", + ) + + +def image_result_item(image: Image) -> ImageUploadResultItem: + """DB Image를 기존 응답 아이템으로 변환합니다.""" + base_prefix = f"{azure_blob_settings.AZURE_BLOB_BASE_URL.rstrip('/')}/" + source: Literal["url", "blob"] = ( + "blob" if image.img_url.startswith(base_prefix) else "url" + ) + return ImageUploadResultItem( + id=image.id, + img_name=image.img_name, + img_url=image.img_url, + img_order=image.img_order, + source=source, + ) diff --git a/app/utils/upload_blob_as_request.py b/app/utils/upload_blob_as_request.py index 7012f76..f3f9857 100644 --- a/app/utils/upload_blob_as_request.py +++ b/app/utils/upload_blob_as_request.py @@ -32,12 +32,17 @@ URL 경로 형식: """ import asyncio +import base64 +import os import re import time +from collections.abc import AsyncIterator from pathlib import Path +from urllib.parse import urlencode import aiofiles import httpx +from fastapi import UploadFile from app.utils.logger import get_logger from config import azure_blob_settings @@ -45,6 +50,15 @@ from config import azure_blob_settings # 로거 설정 logger = get_logger("blob") + +class BlobUploadTooLargeError(ValueError): + """스트리밍 중 파일 크기 상한을 초과했을 때 발생합니다.""" + + def __init__(self, max_size_bytes: int): + self.max_size_bytes = max_size_bytes + super().__init__(f"업로드 파일은 {max_size_bytes} bytes를 초과할 수 없습니다.") + + # ============================================================================= # 모듈 레벨 공유 HTTP 클라이언트 (싱글톤 패턴) # ============================================================================= @@ -100,6 +114,8 @@ class AzureBlobUploader: ".gif": "image/gif", ".webp": "image/webp", ".bmp": "image/bmp", + ".heic": "image/heic", + ".heif": "image/heif", } def __init__(self, user_uuid: str, task_id: str): @@ -111,7 +127,7 @@ class AzureBlobUploader: """ self._user_uuid = user_uuid self._task_id = task_id - self._base_url = azure_blob_settings.AZURE_BLOB_BASE_URL + self._base_url = azure_blob_settings.AZURE_BLOB_BASE_URL.rstrip("/") self._sas_token = azure_blob_settings.AZURE_BLOB_SAS_TOKEN self._last_public_url: str = "" @@ -204,8 +220,12 @@ class AzureBlobUploader: logger.debug(f"[{log_prefix}] Starting upload... " f"(size: {size} bytes, timeout: {timeout}s)") + request_headers = { + **headers, + "x-ms-version": azure_blob_settings.AZURE_BLOB_API_VERSION, + } response = await asyncio.wait_for( - client.put(upload_url, content=file_content, headers=headers), + client.put(upload_url, content=file_content, headers=request_headers), timeout=timeout, ) upload_time = time.perf_counter() @@ -246,6 +266,170 @@ class AzureBlobUploader: f"{type(e).__name__}: {e}") return False + @staticmethod + def _append_query(upload_url: str, **params: str) -> str: + """SAS 쿼리를 유지하며 Azure REST API 쿼리를 추가합니다.""" + separator = "&" if "?" in upload_url else "?" + return f"{upload_url}{separator}{urlencode(params)}" + + async def _delete_upload_url(self, upload_url: str, log_prefix: str) -> bool: + """실패한 업로드의 커밋/미커밋 Blob을 정리합니다.""" + try: + client = await get_shared_blob_client() + response = await client.delete( + upload_url, + headers={"x-ms-version": azure_blob_settings.AZURE_BLOB_API_VERSION}, + ) + if response.status_code in {202, 404}: + return True + logger.warning( + f"[{log_prefix}] Blob cleanup failed - Status: " + f"{response.status_code}, Response: {response.text[:500]}" + ) + except Exception as exc: + logger.warning( + f"[{log_prefix}] Blob cleanup error - {type(exc).__name__}: {exc}" + ) + return False + + async def _upload_stream( + self, + chunks: AsyncIterator[bytes], + upload_url: str, + content_type: str, + timeout: float, + log_prefix: str, + *, + max_size_bytes: int | None = None, + expected_size_bytes: int | None = None, + cleanup_blob_on_failure: bool = False, + ) -> bool: + """Azure Block Blob API로 비동기 청크 스트림을 업로드합니다. + + 각 블록만 메모리에 유지하므로 파일 전체 크기와 무관하게 메모리 사용량이 + 일정합니다. 커밋 전 오류가 발생하면 업로드 대상 Blob 삭제를 시도합니다. + """ + block_ids: list[str] = [] + block_id_nonce = os.urandom(16) + uploaded_size = 0 + start_time = time.perf_counter() + + async def cleanup_failed_stream() -> None: + # 기존 deterministic key는 미커밋 블록만 TTL 정리되게 두어 정상 Blob을 보존합니다. + if cleanup_blob_on_failure: + await self._delete_upload_url(upload_url, log_prefix) + + try: + client = await get_shared_blob_client() + async with asyncio.timeout(timeout): + async for chunk in chunks: + if not chunk: + continue + + uploaded_size += len(chunk) + if max_size_bytes is not None and uploaded_size > max_size_bytes: + raise BlobUploadTooLargeError(max_size_bytes) + + raw_block_id = block_id_nonce + len(block_ids).to_bytes(4, "big") + block_id = base64.b64encode(raw_block_id).decode("ascii") + block_url = self._append_query( + upload_url, + comp="block", + blockid=block_id, + ) + response = await client.put( + block_url, + content=chunk, + headers={ + "Content-Type": "application/octet-stream", + "x-ms-version": ( + azure_blob_settings.AZURE_BLOB_API_VERSION + ), + }, + ) + if response.status_code != 201: + logger.error( + f"[{log_prefix}] Block upload failed - Status: " + f"{response.status_code}, Response: {response.text[:500]}" + ) + await cleanup_failed_stream() + return False + block_ids.append(block_id) + + if uploaded_size == 0: + logger.warning(f"[{log_prefix}] Empty upload stream") + await cleanup_failed_stream() + return False + + if ( + expected_size_bytes is not None + and uploaded_size != expected_size_bytes + ): + logger.error( + f"[{log_prefix}] Stream size changed - expected: " + f"{expected_size_bytes}, actual: {uploaded_size}" + ) + await cleanup_failed_stream() + return False + + block_list = "".join( + f"{block_id}" for block_id in block_ids + ) + commit_body = ( + f'' + f"{block_list}" + ).encode("utf-8") + commit_url = self._append_query(upload_url, comp="blocklist") + response = await client.put( + commit_url, + content=commit_body, + headers={ + "Content-Type": "application/xml; charset=utf-8", + "x-ms-blob-content-type": content_type, + "x-ms-version": azure_blob_settings.AZURE_BLOB_API_VERSION, + }, + ) + if response.status_code not in {200, 201}: + logger.error( + f"[{log_prefix}] Block list commit failed - Status: " + f"{response.status_code}, Response: {response.text[:500]}" + ) + await cleanup_failed_stream() + return False + + duration_ms = (time.perf_counter() - start_time) * 1000 + logger.info( + f"[{log_prefix}] SUCCESS - blocks: {len(block_ids)}, " + f"size: {uploaded_size} bytes, Duration: {duration_ms:.1f}ms" + ) + return True + + except BlobUploadTooLargeError: + await cleanup_failed_stream() + raise + except asyncio.CancelledError: + # 클라이언트 연결 종료 중에도 가능한 범위에서 staged block을 정리합니다. + await asyncio.shield(cleanup_failed_stream()) + raise + except TimeoutError: + elapsed = time.perf_counter() - start_time + logger.error(f"[{log_prefix}] TIMEOUT after {elapsed:.1f}s") + except httpx.HTTPError as exc: + elapsed = time.perf_counter() - start_time + logger.error( + f"[{log_prefix}] HTTP_ERROR after {elapsed:.1f}s - " + f"{type(exc).__name__}: {exc}" + ) + except Exception as exc: + elapsed = time.perf_counter() - start_time + logger.error( + f"[{log_prefix}] ERROR after {elapsed:.1f}s - " + f"{type(exc).__name__}: {exc}" + ) + + await cleanup_failed_stream() + return False + async def _upload_file( self, file_path: str, @@ -273,17 +457,20 @@ class AzureBlobUploader: self._last_public_url = self._build_public_url(category, file_name) logger.debug(f"[{log_prefix}] URL (without SAS): {self._last_public_url}") - headers = {"Content-Type": content_type, "x-ms-blob-type": "BlockBlob"} + async def iter_file() -> AsyncIterator[bytes]: + async with aiofiles.open(file_path, "rb") as file: + while chunk := await file.read( + azure_blob_settings.AZURE_BLOB_UPLOAD_BLOCK_SIZE_BYTES + ): + yield chunk - async with aiofiles.open(file_path, "rb") as file: - file_content = await file.read() - - return await self._upload_bytes( - file_content=file_content, + return await self._upload_stream( + chunks=iter_file(), upload_url=upload_url, - headers=headers, + content_type=content_type, timeout=timeout, log_prefix=log_prefix, + expected_size_bytes=Path(file_path).stat().st_size, ) async def upload_music(self, file_path: str) -> bool: @@ -482,6 +669,51 @@ class AzureBlobUploader: log_prefix=log_prefix, ) + async def upload_image_stream( + self, + file: UploadFile, + file_name: str, + *, + expected_size_bytes: int | None = None, + max_size_bytes: int | None = None, + ) -> bool: + """FastAPI UploadFile을 Azure Block Blob으로 청크 업로드합니다.""" + extension = Path(file_name).suffix.lower() + content_type = self.IMAGE_CONTENT_TYPES.get(extension, "image/jpeg") + file_name = self._sanitize_filename(file_name) + + upload_url = self._build_upload_url("image", file_name) + self._last_public_url = self._build_public_url("image", file_name) + log_prefix = "upload_image_stream" + chunk_size = azure_blob_settings.AZURE_BLOB_UPLOAD_BLOCK_SIZE_BYTES + max_size = ( + max_size_bytes + if max_size_bytes is not None + else azure_blob_settings.IMAGE_UPLOAD_MAX_FILE_SIZE_BYTES + ) + + async def iter_upload() -> AsyncIterator[bytes]: + await file.seek(0) + while chunk := await file.read(chunk_size): + yield chunk + + return await self._upload_stream( + chunks=iter_upload(), + upload_url=upload_url, + content_type=content_type, + timeout=60.0, + log_prefix=log_prefix, + max_size_bytes=max_size, + expected_size_bytes=expected_size_bytes, + cleanup_blob_on_failure=True, + ) + + async def delete_image(self, file_name: str) -> bool: + """이미지 Blob을 삭제합니다. DB 저장 실패 보상 처리용입니다.""" + sanitized_name = self._sanitize_filename(file_name) + upload_url = self._build_upload_url("image", sanitized_name) + return await self._delete_upload_url(upload_url, "delete_image") + # 사용 예시: # import asyncio diff --git a/config.py b/config.py index c9a4e9b..b8853d0 100644 --- a/config.py +++ b/config.py @@ -145,6 +145,40 @@ class AzureBlobSettings(BaseSettings): default="https://ado2mediastoragepublic.blob.core.windows.net/ado2-media-public-access/ado2-media-original", description="Azure Blob Storage 기본 URL", ) + AZURE_BLOB_UPLOAD_BLOCK_SIZE_BYTES: int = Field( + default=4 * 1024 * 1024, + gt=0, + description="Azure Block Blob 업로드 블록 크기 (bytes)", + ) + AZURE_BLOB_API_VERSION: str = Field( + default="2023-11-03", + description="Azure Blob Storage REST API x-ms-version", + ) + IMAGE_UPLOAD_MAX_FILE_SIZE_BYTES: int = Field( + default=15 * 1024 * 1024, + gt=0, + description="이미지 업로드 파일 1개당 최대 크기 (bytes)", + ) + IMAGE_UPLOAD_MAX_REQUEST_SIZE_BYTES: int = Field( + default=20 * 1024 * 1024, + gt=0, + description="한 이미지 업로드 요청에 포함할 수 있는 파일 합계 최대 크기 (bytes)", + ) + IMAGE_UPLOAD_LOCK_TIMEOUT_SECONDS: int = Field( + default=15, + ge=1, + description="동일 task 이미지 append 직렬화 락 대기 시간 (초)", + ) + IMAGE_UPLOAD_MAX_CONCURRENT_LOCKS: int = Field( + default=10, + ge=1, + description="worker별 이미지 upload named lock 동시 점유 상한", + ) + IMAGE_UPLOAD_MAX_TASK_IMAGES: int = Field( + default=100, + ge=1, + description="한 task에 누적할 수 있는 활성 이미지 최대 개수", + ) model_config = _base_config @@ -175,10 +209,7 @@ class CreatomateSettings(BaseSettings): default=False, description="Creatomate 자체 자동 가사 생성 기능 사용 여부", ) - LYRIC_SUBTITLE: bool = Field( - default=False, - description="영상 가사 표기 여부" - ) + LYRIC_SUBTITLE: bool = Field(default=False, description="영상 가사 표기 여부") model_config = _base_config