Merge branch 'main' into feature-ssulbox

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
+testAc

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,
+    request_body: CrawlingRequest, session: AsyncSession = Depends(get_session)
-    session: AsyncSession = Depends(get_session)):
+):
     return await _crawling_logic(request_body.url, session)
 
+
 @router.post(
     "/autocomplete",
     summary="네이버 자동완성 크롤링",
@@ -200,21 +250,30 @@ async def crawling(
     tags=["Crawling"],
 )
 async def autocomplete_crawling(
-    request_body: AutoCompleteRequest,
+    request_body: AutoCompleteRequest, session: AsyncSession = Depends(get_session)
-    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,
+async def _crawling_logic(url: str, session: AsyncSession):
-        session: AsyncSession):
     request_start = time.perf_counter()
-    logger.info("[crawling] ========== START ==========")
+
-    logger.info(f"[crawling] URL: {url[:80]}...")
+    # 요청당 1줄 요약 로그에 쓰이는 값들. 각 Step 이 실제로 실행될 때 채워진다.
+    customer_name = ""
+    region = ""
+    category = ""
+    industry = ""
+    owner_count = 0
+    extra_count = 0
+    filter_summary = "n/a"
+    gpt_status = "completed"
+    step2_elapsed = 0.0
+    step3_elapsed = 0.0
+    step4_elapsed = 0.0
 
     # ========== Step 1: 네이버 지도 크롤링 ==========
     step1_start = time.perf_counter()
-    logger.info("[crawling] Step 1: 네이버 지도 크롤링 시작...")
 
     try:
         scraper = NvMapScraper(url)
@@ -249,14 +308,11 @@ async def _crawling_logic(
         )
 
     step1_elapsed = (time.perf_counter() - step1_start) * 1000
-    logger.info(
+    owner_count = len(scraper.owner_images or [])
-        f"[crawling] Step 1 완료 - 업체사진 {len(scraper.owner_images or [])}개, "
+    extra_count = len(scraper.extra_photo_urls or [])
-        f"보충사진 {len(scraper.extra_photo_urls or [])}개 ({step1_elapsed:.1f}ms)"
-    )
 
     # ========== Step 2: 정보 가공 (industry 선행 계산) ==========
     step2_start = time.perf_counter()
-    logger.info("[crawling] Step 2: 정보 가공 시작...")
 
     processed_info = None
     marketing_analysis = None
@@ -291,10 +347,6 @@ async def _crawling_logic(
         )
 
         step2_elapsed = (time.perf_counter() - step2_start) * 1000
-        logger.info(
-            f"[crawling] Step 2 완료 - {customer_name}, {region}, "
-            f"category={category!r}, industry={industry!r} ({step2_elapsed:.1f}ms)"
-        )
 
         # ========== Step 3: 이미지 마케팅 적합성 필터링 ==========
         # 업체 사진이 SUPPLEMENT_THRESHOLD(30장) 이상이면 보충이 불필요하므로
@@ -309,37 +361,34 @@ async def _crawling_logic(
             #  수집 자체는 NvMapScraper.BIZ_MAX_PAGES가 상한이며 도달 시 scraper가 warning을 남긴다).
             scraper.image_link_list = list(owner_images)
             step3_elapsed = (time.perf_counter() - step3_start) * 1000
-            logger.info(
+            filter_summary = f"skip(owner>={NvMapScraper.SUPPLEMENT_THRESHOLD})"
-                f"[crawling] Step 3 SKIP - 업체 사진 {len(owner_images)}장 "
-                f"≥ {NvMapScraper.SUPPLEMENT_THRESHOLD}장 → 방문자 사진 필터링 생략, 업체 사진만 사용"
-            )
         else:
             try:
                 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
-            logger.info(
+            filter_summary = f"{len(extra_photo_urls)}→{passed_count}"
-                f"[crawling] Step 3 완료 - 방문자 사진 {len(extra_photo_urls)}장 중 "
-                f"{passed_count}장 통과 → 최종 이미지 {len(scraper.image_link_list)}장 "
-                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()
-        logger.info("[crawling] Step 4: ChatGPT 마케팅 분석 시작...")
 
         try:
             # Step 4-1: ChatGPT 서비스 초기화 및 입력 데이터 구성
@@ -376,12 +425,11 @@ 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(
-                f"[crawling] Step 4 완료 - 마케팅 분석 성공 ({step4_elapsed:.1f}ms)"
-            )
 
         except ChatGPTResponseError as e:
             step4_elapsed = (time.perf_counter() - step4_start) * 1000
@@ -408,28 +456,26 @@ async def _crawling_logic(
 
     # ========== 완료 ==========
     total_elapsed = (time.perf_counter() - request_start) * 1000
-    logger.info("[crawling] ========== COMPLETE ==========")
+    logger.info(
-    logger.info(f"[crawling] 총 소요시간: {total_elapsed:.1f}ms")
+        f"[crawling] SUCCESS - url: {url[:80]}, name: {customer_name!r}, "
-    logger.info(f"[crawling]   - Step 1 (크롤링): {step1_elapsed:.1f}ms")
+        f"region: {region!r}, category: {category!r}, industry: {industry!r}, "
-    if scraper.base_info:
+        f"owner: {owner_count}, extra: {extra_count}, filter: {filter_summary}, "
-        logger.info(f"[crawling]   - Step 2 (정보가공): {step2_elapsed:.1f}ms")
+        f"images: {len(scraper.image_link_list or [])}, gpt: {gpt_status}, "
-        if "step3_elapsed" in locals():
+        f"timing(ms): s1={step1_elapsed:.1f} s2={step2_elapsed:.1f} "
-            logger.info(f"[crawling]   - Step 3 (이미지 필터링): {step3_elapsed:.1f}ms")
+        f"s3={step3_elapsed:.1f} s4={step4_elapsed:.1f} total={total_elapsed:.1f}"
-        if "step4_elapsed" in locals():
+    )
-            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,
         "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 +561,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 +589,7 @@ async def _autocomplete_logic(autocomplete_item:dict):
 
     return new_url
 
+
 def _extract_image_name(url: str, index: int) -> str:
     """URL에서 이미지 이름 추출 또는 기본 이름 생성"""
     try:
@@ -555,30 +602,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 +610,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 Storage에 업로드하고 task_id를 생성하거나 기존 작업에 이어 붙입니다.
-바이너리 파일은 로컬 서버에 저장하지 않고 Azure Blob에 직접 업로드됩니다.
+바이너리 파일은 로컬 서버 경로에 복사하지 않고 Azure Blob에 청크 업로드됩니다.
 
 ## 인증
 **Bearer 토큰 필수** - `Authorization: Bearer {access_token}` 헤더를 포함해야 합니다.
@@ -603,8 +627,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 +664,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 +690,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 +723,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,126 +745,288 @@ 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 생성
+    if requested_task_id:
-    task_id = await generate_task_id()
+        task_id = _normalize_continuation_task_id(requested_task_id)
-    logger.info(f"[upload_images_blob] START - task_id: {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: 입력 검증 및 파일 데이터 준비 (세션 없음) ==========
+    # ========== 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 and not (is_continuation and finalize):
-    if not has_images_json and not has_files:
         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):
+            if not isinstance(parsed, list):
-                url_images = [ImageUrlItem(**item) for item in parsed if item]
+                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)}",
             )
 
-    # 유효한 파일만 필터링 및 파일 내용 미리 읽기
+    upload_files = files or []
-    valid_files_data: list[tuple[str, str, bytes]] = []  # (original_name, ext, content)
+    declared_total_size = sum(
-    skipped_files: list[str] = []
+        file.size for file in upload_files if file.size is not None and file.size > 0
-    if has_files and files:
+    )
-        for f in files:
+    max_request_size = azure_blob_settings.IMAGE_UPLOAD_MAX_REQUEST_SIZE_BYTES
-            is_valid_ext = _is_valid_image_extension(f.filename)
+    if declared_total_size > max_request_size:
-            is_not_empty = f.size is None or f.size > 0
+        raise HTTPException(
-            is_real_file = f.filename and f.filename != "filename"
+            status_code=status.HTTP_413_CONTENT_TOO_LARGE,
-
+            detail=(
-            if f and is_real_file and is_valid_ext and is_not_empty:
+                "한 요청의 파일 합계가 최대 크기 "
-                # 파일 내용을 미리 읽어둠
+                f"{max_request_size // (1024 * 1024)} MiB를 초과합니다."
-                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}"
         )
+
+    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="분할 업로드의 첫 요청에는 소유권 확인용 이미지 파일이 필요합니다.",
         )
 
-    stage1_time = time.perf_counter()
+    validate_task_image_count(
-    logger.info(
+        existing_count=len(existing_images),
-        f"[upload_images_blob] Stage 1 done - urls: {len(url_images)}, "
+        incoming_count=len(url_images) + len(valid_files_data),
-        f"files: {len(valid_files_data)}, "
-        f"elapsed: {(stage1_time - request_start) * 1000:.1f}ms"
     )
 
-    # ========== Stage 2: Azure Blob 업로드 (세션 없음) ==========
+    stage1_time = time.perf_counter()
-    # 업로드 결과를 저장할 리스트 (나중에 DB에 저장)
+
-    blob_upload_results: list[tuple[str, str]] = []  # (img_name, blob_url)
+    # ========== Stage 2: Azure Blob 청크 업로드 (세션 없음) ==========
-    img_order = len(url_images)  # URL 이미지 다음 순서부터 시작
+    # (원본명, 공개 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):
+        for idx, (file, original_name, extension, actual_size) in enumerate(
-            name_without_ext = (
+            valid_files_data
-                original_name.rsplit(".", 1)[0]
+        ):
-                if "." in original_name
+            name_without_ext = Path(original_name).stem
-                else original_name
+            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에 직접 업로드
+            try:
-            upload_success = await uploader.upload_image_bytes(file_content, filename)
+                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(
-                blob_upload_results.append((original_name, blob_url))
+                    (original_name, uploader.public_url, stored_name)
-                img_order += 1
+                )
                 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: "
-        f"{len(blob_upload_results)}, skipped: {len(skipped_files)}, "
-        f"elapsed: {(stage2_time - stage1_time) * 1000:.1f}ms"
-    )
 
     # ========== Stage 3: DB 저장 (새 세션으로 빠르게 처리) ==========
-    logger.info("[upload_images_blob] Stage 3 starting - DB save...")
+    all_images: list[Image] = []
-    result_images: list[ImageUploadResultItem] = []
+    # 요약 로그용. 커밋이 끝난 뒤 실제 값으로 덮어쓴다.
-    img_order = 0
+    stage3_time = stage2_time
+    added_count = 0
+    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 +1034,10 @@ async def upload_images_blob(
                     img_order=img_order,
                 )
                 session.add(image)
-                await session.flush()
+                new_images.append(image)
-
-                result_images.append(
-                    ImageUploadResultItem(
-                        id=image.id,
-                        img_name=img_name,
-                        img_url=url_item.url,
-                        img_order=img_order,
-                        source="url",
-                    )
-                )
                 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 +1045,27 @@ async def upload_images_blob(
                     img_order=img_order,
                 )
                 session.add(image)
-                await session.flush()
+                new_images.append(image)
-
-                result_images.append(
-                    ImageUploadResultItem(
-                        id=image.id,
-                        img_name=img_name,
-                        img_url=blob_url,
-                        img_order=img_order,
-                        source="blob",
-                    )
-                )
                 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(
+            added_count = len(new_images)
-                f"[upload_images_blob] Stage 3 done - "
-                f"saved: {len(result_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 +1073,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,36 +1084,48 @@ 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}")
+    tagging_summary = "deferred"
-    await tagging_images(image_urls, industry=industry, clear_old_tags=True)
+    if finalize:
-    logger.info(f"[image_tagging] Done - task_id: {task_id}")
+        await tagging_images(image_urls, industry=industry, clear_old_tags=True)
 
-    # 태깅 직후 영상 생성에 사용 가능한 이미지가 하나도 없으면 조기에 실패시킨다.
+        # 마지막 분할 요청에서 누적된 전체 이미지의 적합성을 확인합니다.
-    # (여기서 걸러지지 않으면 훨씬 나중인 영상 생성 단계에서야 슬롯 미배정으로 발견됨)
+        taged_image_list = await get_image_tags_by_task_id(task_id)
-    # marketing_acceptable 필터링은 크롤링 단계에서 이미 완료되었으므로 여기서는 재필터링하지 않는다.
+        tagging_summary = f"tagged={len(taged_image_list)}"
-    taged_image_list = await get_image_tags_by_task_id(task_id)
+        if not taged_image_list:
-    logger.info(f"태깅된 이미지: {len(taged_image_list)}개 - task_id: {task_id}")
+            logger.error(
-    if not taged_image_list:
+                f"[image_tagging] 영상 생성에 적합한 이미지가 없음 - task_id: {task_id}"
-        logger.error(f"[image_tagging] 영상 생성에 적합한 이미지가 없음 - task_id: {task_id}")
+            )
-        raise HTTPException(
+            raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
+                status_code=status.HTTP_400_BAD_REQUEST,
-            detail="영상 생성에 적합한 이미지가 없습니다. 다른 이미지로 다시 업로드해주세요.",
+                detail=(
-        )
+                    "영상 생성에 적합한 이미지가 없습니다. "
+                    "다른 이미지로 다시 업로드해주세요."
+                ),
+            )
 
     total_time = time.perf_counter() - request_start
     logger.info(
         f"[upload_images_blob] SUCCESS - task_id: {task_id}, "
-        f"total: {saved_count}, total_time: {total_time * 1000:.1f}ms"
+        f"cont: {is_continuation}, finalize: {finalize}, "
+        f"urls: {len(url_images)}, files: {len(valid_files_data)}, "
+        f"bytes: {actual_total_size}, blobs: {len(blob_upload_results)}, "
+        f"skipped: {len(skipped_files)}, added: {added_count}, "
+        f"task_total: {saved_count}, tagging: {tagging_summary}, "
+        f"timing(ms): s1={(stage1_time - request_start) * 1000:.1f} "
+        f"blob={(stage2_time - stage1_time) * 1000:.1f} "
+        f"db={(stage3_time - stage2_time) * 1000:.1f} "
+        f"total={total_time * 1000:.1f}"
     )
 
     return ImageUploadResponse(
         task_id=task_id,
-        total_count=len(result_images),
+        total_count=saved_count,
-        url_count=len(url_images),
+        url_count=sum(image.source == "url" for image in result_images),
-        file_count=len(blob_upload_results),
+        file_count=sum(image.source == "blob" for image in result_images),
         saved_count=saved_count,
         images=result_images,
         image_urls=image_urls,
@@ -913,10 +1133,8 @@ async def upload_images_blob(
 
 
 async def tagging_images(
-        image_urls : list[str],
+    image_urls: list[str], industry: str = "", clear_old_tags: bool = False
-        industry: str = "",
+) -> None:
-        clear_old_tags : bool = False
-    ) -> None:
     # 1. 조회
     async with AsyncSessionLocal() as session:
         stmt = (
@@ -938,9 +1156,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:

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,
+    )

app/utils/creatomate.py

@@ -574,8 +574,15 @@ class CreatomateService:
           다운로드 실패를 배제로 오판해 풀 전체가 날아가는 것을 막기 위함.
         일반 씬 슬롯은 태그 점수를 그대로 반환한다.
         """
-        scores = self.calculate_image_slot_score_multi(pool_subset, slot)
+        is_thumbnail = slot.endswith(THUMBNAIL_SLOT_MARKER)
-        if not thumbnail_fitness_map or not slot.endswith(THUMBNAIL_SLOT_MARKER):
+        if is_thumbnail and self.parse_slot_name_to_tag(slot) is None:
+            # 슬롯명이 명명 규칙을 어겨 태그 매칭이 불가능한 썸네일 슬롯.
+            # 태그 점수를 0으로 두면 pool 첫 컷이 뽑혀 사실상 무작위가 되므로
+            # 전 이미지 중립(1.0)으로 두고 아래 픽셀 적합도만으로 순위를 가른다.
+            scores = [1.0] * len(pool_subset)
+        else:
+            scores = self.calculate_image_slot_score_multi(pool_subset, slot)
+        if not thumbnail_fitness_map or not is_thumbnail:
             return scores
 
         adjusted = []
@@ -589,6 +596,31 @@ class CreatomateService:
                 adjusted.append(score * fitness["score"])
         return adjusted
 
+    def _collect_thumbnail_slots(self, template_component_data: dict) -> list[str]:
+        """배정 대상 썸네일 슬롯(-9999)을 수집합니다.
+
+        일반 씬 슬롯과 달리 슬롯명 파싱에 실패해도 제외하지 않는다. 썸네일은
+        노출 면적이 가장 큰 표면이라, 미배정 시 modify_element가 템플릿 원본
+        (샘플 이미지)을 그대로 남겨 완성 영상에 그대로 나가기 때문이다. 태그
+        매칭이 불가능한 슬롯은 _slot_scores_with_fitness가 픽셀 적합도만으로
+        고른다. 단 '-fixed' 고정 자산은 여기서도 제외한다.
+
+        파싱 실패는 템플릿 슬롯명 오타이므로 ERROR로 남겨 드러나게 한다
+        (조용히 넘어가면 샘플 이미지가 나가도 아무도 알아채지 못한다).
+        """
+        slots = [
+            name for name, t in template_component_data.items()
+            if t == "image" and name.endswith(THUMBNAIL_SLOT_MARKER)
+            and not is_fixed_slot_name(name)
+        ]
+        for name in slots:
+            if self.parse_slot_name_to_tag(name) is None:
+                logger.error(
+                    f"[_collect_thumbnail_slots] 썸네일 슬롯명이 명명 규칙 위반 — "
+                    f"'{name}' — 템플릿 슬롯명 수정 필요. 픽셀 적합도 기준으로 폴백 배정합니다."
+                )
+        return slots
+
     def rank_thumbnail_candidates(
         self,
         template: dict,
@@ -607,11 +639,7 @@ class CreatomateService:
             후보가 없는 슬롯은 키 자체를 포함하지 않는다.
         """
         component = self.parse_template_component_name(template["source"]["elements"])
-        thumbnail_slots = [
+        thumbnail_slots = self._collect_thumbnail_slots(component)
-            name for name, t in component.items()
-            if t == "image" and name.endswith(THUMBNAIL_SLOT_MARKER)
-            and not is_fixed_slot_name(name) and self.parse_slot_name_to_tag(name) is not None
-        ]
         result: dict[str, list[dict]] = {}
         for slot in thumbnail_slots:
             scores = self._slot_scores_with_fitness(taged_image_list, slot, thumbnail_fitness_map)
@@ -669,9 +697,13 @@ class CreatomateService:
         # 않는 image 요소는 콘텐츠 슬롯이 아니므로 배정 대상에서 제외 — 그렇지
         # 않으면 파싱 실패로 0점 처리되어 "가장 까다로운 슬롯"으로 취급되고
         # 무작위 이미지로 덮어써진다.
+        # 썸네일 슬롯(-9999)은 파싱 실패해도 배정해야 하므로 여기서 제외하고
+        # _collect_thumbnail_slots가 별도로 수집한다.
         image_slots = [
             name for name, t in template_component_data.items()
-            if t == "image" and not is_fixed_slot_name(name) and self.parse_slot_name_to_tag(name) is not None
+            if t == "image" and not is_fixed_slot_name(name)
+            and not name.endswith(THUMBNAIL_SLOT_MARKER)
+            and self.parse_slot_name_to_tag(name) is not None
         ]
         text_slots  = [(name, t) for name, t in template_component_data.items() if t == "text"]
 
@@ -686,8 +718,7 @@ class CreatomateService:
         # thumbnail_choice(비전 LLM 최종 선택)가 해당 슬롯에 있으면 그 이미지를,
         # 없으면(선택 실패/미제공) 결정론적 최고점 컷을 사용한다 — 폴백 안전.
         thumbnail_choice = thumbnail_choice or {}
-        thumbnail_slots = [s for s in image_slots if s.endswith(THUMBNAIL_SLOT_MARKER)]
+        thumbnail_slots = self._collect_thumbnail_slots(template_component_data)
-        image_slots = [s for s in image_slots if not s.endswith(THUMBNAIL_SLOT_MARKER)]
         for slot in thumbnail_slots:
             if not pool:
                 logger.warning(f"[template_matching_taged_image] 이미지 풀 없음 — 썸네일 슬롯 배정 불가: {slot}")
@@ -885,7 +916,13 @@ class CreatomateService:
         """슬롯 이름을 파싱하여 태그 딕셔너리를 반환합니다.
 
         슬롯 이름 형식: {space_type}-{subject}-{camera}-{motion}-{narrative}
-        파싱 실패 시 None을 반환합니다 (호출자가 해당 슬롯을 skip+log 처리).
+
+        위치 기반 파싱에 실패하면 토큰 위치를 무시한 대조로 한 번 더 시도한다
+        (_parse_slot_name_loosely). 슬롯명 오타로 슬롯이 배정에서 빠지면
+        modify_element가 템플릿 원본(샘플 이미지)을 그대로 남겨 완성 영상에
+        그대로 나가기 때문이다.
+
+        둘 다 실패하면 None을 반환합니다 (호출자가 해당 슬롯을 skip+log 처리).
         """
         try:
             tag_list = slot_name.split("-")
@@ -907,9 +944,64 @@ class CreatomateService:
             }
             return tag_dict
         except (ValueError, IndexError) as e:
+            loose = self._parse_slot_name_loosely(tag_list)
+            if loose is not None:
+                logger.warning(
+                    f"[parse_slot_name_to_tag] 슬롯명이 명명 규칙 위반: '{slot_name}' — {e} — "
+                    f"위치 무시 대조로 복구: { {k: v.value for k, v in loose.items()} } — 템플릿 슬롯명 수정 권장"
+                )
+                return loose
             logger.warning(f"[parse_slot_name_to_tag] 슬롯명 파싱 실패: '{slot_name}' — {e} — 슬롯 skip")
             return None
 
+    def _parse_slot_name_loosely(self, tag_list: list[str]) -> dict[str, StrEnum] | None:
+        """토큰 위치를 무시하고 각 enum에 대조해 태그를 복구합니다.
+
+        위치 기반 파싱이 실패했을 때만 호출한다. 토큰 하나는 한 카테고리에만
+        쓰이며, 앞선 토큰부터 순서대로 소비한다(중복 후보가 있으면 앞선 것 채택
+        — 위치 기반과 같은 값을 고르게 된다).
+
+        space_type/subject/narrative는 필수다. 이 셋을 못 채우면 슬롯명이 아닌
+        것으로 보고 None을 반환한다(고정 자산·비규칙 요소가 배정 대상에 섞여
+        무작위 이미지로 덮어써지는 것을 막기 위함). camera/motion은 선택이며
+        못 찾으면 키 자체를 넣지 않는다 — 점수 계산은 태그 딕셔너리를 순회하므로
+        없는 키는 자연히 가중치에서 빠진다.
+        """
+        used: set[int] = set()
+
+        def take(converter) -> StrEnum | None:
+            for idx, token in enumerate(tag_list):
+                if idx in used:
+                    continue
+                try:
+                    value = converter(token)
+                except ValueError:
+                    continue
+                if value is not None:
+                    used.add(idx)
+                    return value
+            return None
+
+        space_type = take(SpaceType)
+        subject = take(Subject)
+        narrative = take(NarrativePhase)
+        if space_type is None or subject is None or narrative is None:
+            return None
+
+        camera = take(Camera)
+        motion = take(lambda t: MOTION_TOKEN_NORMALIZATION.get(t) or MotionRecommended(t))
+
+        tag_dict: dict[str, StrEnum] = {
+            "space_type": space_type,
+            "subject": subject,
+            "narrative_preference": narrative,
+        }
+        if camera is not None:
+            tag_dict["camera"] = camera
+        if motion is not None:
+            tag_dict["motion_recommended"] = motion
+        return tag_dict
+
     def elements_connect_resource_blackbox(
         self,
         elements: list,

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"<Latest>{block_id}</Latest>" for block_id in block_ids
+                )
+                commit_body = (
+                    f'<?xml version="1.0" encoding="utf-8"?>'
+                    f"<BlockList>{block_list}</BlockList>"
+                ).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:
+        return await self._upload_stream(
-            file_content = await file.read()
+            chunks=iter_file(),
-
-        return await self._upload_bytes(
-            file_content=file_content,
             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

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(
+    LYRIC_SUBTITLE: bool = Field(default=False, description="영상 가사 표기 여부")
-        default=False,
-        description="영상 가사 표기 여부"
-    )
 
     model_config = _base_config