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