646 lines
23 KiB
Python
646 lines
23 KiB
Python
import json
|
|
from datetime import date
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
from urllib.parse import unquote, urlparse
|
|
|
|
import aiofiles
|
|
from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile, status
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.database.session import get_session
|
|
from app.home.models import Image
|
|
from app.home.schemas.home_schema import (
|
|
CrawlingRequest,
|
|
CrawlingResponse,
|
|
ErrorResponse,
|
|
ImageUploadResponse,
|
|
ImageUploadResultItem,
|
|
ImageUrlItem,
|
|
MarketingAnalysis,
|
|
ProcessedInfo,
|
|
)
|
|
from app.utils.upload_blob_as_request import AzureBlobUploader
|
|
from app.utils.chatgpt_prompt import ChatgptService
|
|
from app.utils.common import generate_task_id
|
|
from app.utils.nvMapScraper import NvMapScraper
|
|
|
|
MEDIA_ROOT = Path("media")
|
|
|
|
# 전국 시 이름 목록 (roadAddress에서 region 추출용)
|
|
# fmt: off
|
|
KOREAN_CITIES = [
|
|
# 특별시/광역시
|
|
"서울시", "부산시", "대구시", "인천시", "광주시", "대전시", "울산시", "세종시",
|
|
# 경기도
|
|
"수원시", "성남시", "고양시", "용인시", "부천시", "안산시", "안양시", "남양주시",
|
|
"화성시", "평택시", "의정부시", "시흥시", "파주시", "광명시", "김포시", "군포시",
|
|
"광주시", "이천시", "양주시", "오산시", "구리시", "안성시", "포천시", "의왕시",
|
|
"하남시", "여주시", "동두천시", "과천시",
|
|
# 강원도
|
|
"춘천시", "원주시", "강릉시", "동해시", "태백시", "속초시", "삼척시",
|
|
# 충청북도
|
|
"청주시", "충주시", "제천시",
|
|
# 충청남도
|
|
"천안시", "공주시", "보령시", "아산시", "서산시", "논산시", "계룡시", "당진시",
|
|
# 전라북도
|
|
"전주시", "군산시", "익산시", "정읍시", "남원시", "김제시",
|
|
# 전라남도
|
|
"목포시", "여수시", "순천시", "나주시", "광양시",
|
|
# 경상북도
|
|
"포항시", "경주시", "김천시", "안동시", "구미시", "영주시", "영천시", "상주시", "문경시", "경산시",
|
|
# 경상남도
|
|
"창원시", "진주시", "통영시", "사천시", "김해시", "밀양시", "거제시", "양산시",
|
|
# 제주도
|
|
"제주시", "서귀포시",
|
|
]
|
|
# fmt: on
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def _extract_region_from_address(road_address: str | None) -> str:
|
|
"""roadAddress에서 시 이름 추출"""
|
|
if not road_address:
|
|
return ""
|
|
for city in KOREAN_CITIES:
|
|
if city in road_address:
|
|
return city
|
|
return ""
|
|
|
|
|
|
@router.post(
|
|
"/crawling",
|
|
summary="네이버 지도 크롤링",
|
|
description="""
|
|
네이버 지도 장소 URL을 입력받아 이미지 목록과 기본 정보를 크롤링합니다.
|
|
|
|
## 요청 필드
|
|
- **url**: 네이버 지도 장소 URL (필수)
|
|
|
|
## 반환 정보
|
|
- **image_list**: 장소 이미지 URL 목록
|
|
- **image_count**: 이미지 개수
|
|
- **processed_info**: 가공된 장소 정보 (customer_name, region, detail_region_info)
|
|
""",
|
|
response_model=CrawlingResponse,
|
|
response_description="크롤링 결과",
|
|
responses={
|
|
200: {"description": "크롤링 성공", "model": CrawlingResponse},
|
|
400: {
|
|
"description": "잘못된 URL",
|
|
"model": ErrorResponse,
|
|
},
|
|
},
|
|
tags=["crawling"],
|
|
)
|
|
async def crawling(request_body: CrawlingRequest):
|
|
"""네이버 지도 장소 크롤링"""
|
|
scraper = NvMapScraper(request_body.url)
|
|
await scraper.scrap()
|
|
|
|
# 가공된 정보 생성
|
|
processed_info = None
|
|
marketing_analysis = None
|
|
|
|
if scraper.base_info:
|
|
road_address = scraper.base_info.get("roadAddress", "")
|
|
customer_name = scraper.base_info.get("name", "")
|
|
region = _extract_region_from_address(road_address)
|
|
|
|
processed_info = ProcessedInfo(
|
|
customer_name=customer_name,
|
|
region=region,
|
|
detail_region_info=road_address or "",
|
|
)
|
|
|
|
# ChatGPT를 이용한 마케팅 분석
|
|
chatgpt_service = ChatgptService(
|
|
customer_name=customer_name,
|
|
region=region,
|
|
detail_region_info=road_address or "",
|
|
)
|
|
prompt = chatgpt_service.build_market_analysis_prompt()
|
|
raw_response = await chatgpt_service.generate(prompt)
|
|
parsed = await chatgpt_service.parse_marketing_analysis(raw_response)
|
|
marketing_analysis = MarketingAnalysis(**parsed)
|
|
|
|
return {
|
|
"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,
|
|
}
|
|
|
|
|
|
def _extract_image_name(url: str, index: int) -> str:
|
|
"""URL에서 이미지 이름 추출 또는 기본 이름 생성"""
|
|
try:
|
|
path = urlparse(url).path
|
|
filename = path.split("/")[-1] if path else ""
|
|
if filename:
|
|
return unquote(filename)
|
|
except Exception:
|
|
pass
|
|
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"},
|
|
{"url": "https://naverbooking-phinf.pstatic.net/20240514_92/17156880307484bvpH_JPEG/3.jpg"},
|
|
{"url": "https://naverbooking-phinf.pstatic.net/20240514_7/1715688031000y8Y5q_JPEG/4.jpg"},
|
|
{"url": "https://naverbooking-phinf.pstatic.net/20240514_259/17156880311809wCnY_JPEG/5.jpg", "name": "외관"}
|
|
]"""
|
|
|
|
|
|
@router.post(
|
|
"/image/upload/server/{task_id}",
|
|
include_in_schema=False,
|
|
summary="이미지 업로드",
|
|
description="""
|
|
task_id에 연결된 이미지를 서버에 업로드합니다.
|
|
|
|
## 요청 방식
|
|
multipart/form-data 형식으로 전송합니다.
|
|
|
|
## 요청 필드
|
|
- **images_json**: 외부 이미지 URL 목록 (JSON 문자열, 선택)
|
|
- **files**: 이미지 바이너리 파일 목록 (선택)
|
|
|
|
**주의**: images_json 또는 files 중 최소 하나는 반드시 전달해야 합니다.
|
|
|
|
## 지원 이미지 확장자
|
|
jpg, jpeg, png, webp, heic, heif
|
|
|
|
## images_json 예시
|
|
```json
|
|
[
|
|
{"url": "https://naverbooking-phinf.pstatic.net/20240514_189/1715688030436xT14o_JPEG/1.jpg"},
|
|
{"url": "https://naverbooking-phinf.pstatic.net/20240514_48/1715688030574wTtQd_JPEG/2.jpg"},
|
|
{"url": "https://naverbooking-phinf.pstatic.net/20240514_92/17156880307484bvpH_JPEG/3.jpg"},
|
|
{"url": "https://naverbooking-phinf.pstatic.net/20240514_7/1715688031000y8Y5q_JPEG/4.jpg"},
|
|
{"url": "https://naverbooking-phinf.pstatic.net/20240514_259/17156880311809wCnY_JPEG/5.jpg", "name": "외관"}
|
|
]
|
|
```
|
|
|
|
## 바이너리 파일 업로드 테스트 방법
|
|
|
|
### 1. Swagger UI에서 테스트
|
|
1. 이 엔드포인트의 "Try it out" 버튼 클릭
|
|
2. task_id 입력 (예: test-task-001)
|
|
3. files 항목에서 "Add item" 클릭하여 로컬 이미지 파일 선택
|
|
4. (선택) images_json에 URL 목록 JSON 입력
|
|
5. "Execute" 버튼 클릭
|
|
|
|
### 2. cURL로 테스트
|
|
```bash
|
|
# 바이너리 파일만 업로드
|
|
curl -X POST "http://localhost:8000/image/upload/server/test-task-001" \\
|
|
-F "files=@/path/to/image1.jpg" \\
|
|
-F "files=@/path/to/image2.png"
|
|
|
|
# URL + 바이너리 파일 동시 업로드
|
|
curl -X POST "http://localhost:8000/image/upload/server/test-task-001" \\
|
|
-F 'images_json=[{"url":"https://example.com/image.jpg"}]' \\
|
|
-F "files=@/path/to/local_image.jpg"
|
|
```
|
|
|
|
### 3. Python requests로 테스트
|
|
```python
|
|
import requests
|
|
|
|
url = "http://localhost:8000/image/upload/server/test-task-001"
|
|
files = [
|
|
("files", ("image1.jpg", open("image1.jpg", "rb"), "image/jpeg")),
|
|
("files", ("image2.png", open("image2.png", "rb"), "image/png")),
|
|
]
|
|
data = {
|
|
"images_json": '[{"url": "https://example.com/image.jpg"}]'
|
|
}
|
|
response = requests.post(url, files=files, data=data)
|
|
print(response.json())
|
|
```
|
|
|
|
## 반환 정보
|
|
- **task_id**: 작업 고유 식별자
|
|
- **total_count**: 총 업로드된 이미지 개수
|
|
- **url_count**: URL로 등록된 이미지 개수
|
|
- **file_count**: 파일로 업로드된 이미지 개수
|
|
- **images**: 업로드된 이미지 목록
|
|
|
|
## 저장 경로
|
|
- 바이너리 파일: /media/image/{날짜}/{uuid7}/{파일명}
|
|
""",
|
|
response_model=ImageUploadResponse,
|
|
responses={
|
|
200: {"description": "이미지 업로드 성공"},
|
|
400: {"description": "이미지가 제공되지 않음", "model": ErrorResponse},
|
|
},
|
|
tags=["image"],
|
|
)
|
|
async def upload_images(
|
|
task_id: str,
|
|
images_json: Optional[str] = Form(
|
|
default=None,
|
|
description="외부 이미지 URL 목록 (JSON 문자열)",
|
|
example=IMAGES_JSON_EXAMPLE,
|
|
),
|
|
files: Optional[list[UploadFile]] = File(
|
|
default=None, description="이미지 바이너리 파일 목록"
|
|
),
|
|
session: AsyncSession = Depends(get_session),
|
|
) -> ImageUploadResponse:
|
|
"""이미지 업로드 (URL + 바이너리 파일)"""
|
|
print(f"[upload_images] START - task_id: {task_id}")
|
|
print(f"[upload_images] images_json: {images_json}")
|
|
print(f"[upload_images] files: {files}")
|
|
|
|
# 1. 진입 검증: images_json 또는 files 중 하나는 반드시 있어야 함
|
|
has_images_json = images_json is not None and images_json.strip() != ""
|
|
has_files = files is not None and len(files) > 0
|
|
print(f"[upload_images] has_images_json: {has_images_json}, has_files: {has_files}")
|
|
|
|
if has_files and files:
|
|
for idx, f in enumerate(files):
|
|
print(
|
|
f"[upload_images] file[{idx}]: filename={f.filename}, size={f.size}, content_type={f.content_type}"
|
|
)
|
|
|
|
if not has_images_json and not has_files:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="images_json 또는 files 중 하나는 반드시 제공해야 합니다.",
|
|
)
|
|
|
|
# 2. images_json 파싱 (있는 경우만)
|
|
url_images: list[ImageUrlItem] = []
|
|
if has_images_json:
|
|
try:
|
|
parsed = json.loads(images_json)
|
|
if isinstance(parsed, list):
|
|
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)}",
|
|
)
|
|
|
|
# 3. 유효한 파일만 필터링 (빈 파일, 유효한 이미지 확장자가 아닌 경우 제외)
|
|
valid_files: list[UploadFile] = []
|
|
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
|
|
) # size가 None이면 아직 읽지 않은 것
|
|
is_real_file = (
|
|
f.filename and f.filename != "filename"
|
|
) # Swagger 빈 파일 체크
|
|
print(
|
|
f"[upload_images] Checking file: {f.filename}, size={f.size}, is_valid_ext={is_valid_ext}, is_real_file={is_real_file}"
|
|
)
|
|
|
|
if f and is_real_file and is_valid_ext and is_not_empty:
|
|
valid_files.append(f)
|
|
else:
|
|
skipped_files.append(f.filename or "unknown")
|
|
|
|
print(
|
|
f"[upload_images] valid_files count: {len(valid_files)}, skipped: {skipped_files}"
|
|
)
|
|
|
|
# 유효한 데이터가 하나도 없으면 에러
|
|
if not url_images and not valid_files:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"유효한 이미지가 없습니다. 지원 확장자: {', '.join(ALLOWED_IMAGE_EXTENSIONS)}. 건너뛴 파일: {skipped_files}",
|
|
)
|
|
|
|
result_images: list[ImageUploadResultItem] = []
|
|
img_order = 0
|
|
|
|
# 1. URL 이미지 저장
|
|
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,
|
|
img_url=url_item.url,
|
|
img_order=img_order,
|
|
)
|
|
session.add(image)
|
|
await session.flush() # ID 생성을 위해 flush
|
|
print(f"[upload_images] URL image saved - id: {image.id}, img_name: {img_name}")
|
|
|
|
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
|
|
|
|
# 2. 바이너리 파일을 media에 저장
|
|
if valid_files:
|
|
today = date.today().strftime("%Y-%m-%d")
|
|
# 한 번의 요청에서 업로드된 모든 이미지는 같은 폴더에 저장
|
|
batch_uuid = await generate_task_id()
|
|
upload_dir = MEDIA_ROOT / "image" / today / batch_uuid
|
|
upload_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
for file in valid_files:
|
|
# 파일명: 원본 파일명 사용 (중복 방지를 위해 순서 추가)
|
|
original_name = file.filename or "image"
|
|
ext = _get_file_extension(file.filename) # type: ignore[arg-type]
|
|
# 파일명에서 확장자 제거 후 순서 추가
|
|
name_without_ext = (
|
|
original_name.rsplit(".", 1)[0]
|
|
if "." in original_name
|
|
else original_name
|
|
)
|
|
filename = f"{name_without_ext}_{img_order:03d}{ext}"
|
|
|
|
save_path = upload_dir / filename
|
|
|
|
# media에 파일 저장
|
|
await _save_upload_file(file, save_path)
|
|
|
|
# media 기준 URL 생성
|
|
img_url = f"/media/image/{today}/{batch_uuid}/{filename}"
|
|
img_name = file.filename or filename
|
|
print(f"[upload_images] File saved to media - path: {save_path}, url: {img_url}")
|
|
|
|
image = Image(
|
|
task_id=task_id,
|
|
img_name=img_name,
|
|
img_url=img_url, # Media URL을 DB에 저장
|
|
img_order=img_order,
|
|
)
|
|
session.add(image)
|
|
await session.flush()
|
|
|
|
result_images.append(
|
|
ImageUploadResultItem(
|
|
id=image.id,
|
|
img_name=img_name,
|
|
img_url=img_url,
|
|
img_order=img_order,
|
|
source="file",
|
|
)
|
|
)
|
|
img_order += 1
|
|
|
|
saved_count = len(result_images)
|
|
print(f"[upload_images] Committing {saved_count} images to database...")
|
|
await session.commit()
|
|
print("[upload_images] Commit successful!")
|
|
|
|
return ImageUploadResponse(
|
|
task_id=task_id,
|
|
total_count=len(result_images),
|
|
url_count=len(url_images),
|
|
file_count=len(valid_files),
|
|
saved_count=saved_count,
|
|
images=result_images,
|
|
)
|
|
|
|
|
|
@router.post(
|
|
"/image/upload/blob/{task_id}",
|
|
summary="이미지 업로드 (Azure Blob)",
|
|
description="""
|
|
task_id에 연결된 이미지를 Azure Blob Storage에 업로드합니다.
|
|
|
|
## 요청 방식
|
|
multipart/form-data 형식으로 전송합니다.
|
|
|
|
## 요청 필드
|
|
- **images_json**: 외부 이미지 URL 목록 (JSON 문자열, 선택)
|
|
- **files**: 이미지 바이너리 파일 목록 (선택)
|
|
|
|
**주의**: images_json 또는 files 중 최소 하나는 반드시 전달해야 합니다.
|
|
|
|
## 지원 이미지 확장자
|
|
jpg, jpeg, png, webp, heic, heif
|
|
|
|
## images_json 예시
|
|
```json
|
|
[
|
|
{"url": "https://example.com/image1.jpg"},
|
|
{"url": "https://example.com/image2.jpg", "name": "외관"}
|
|
]
|
|
```
|
|
|
|
## 바이너리 파일 업로드 테스트 방법
|
|
|
|
### cURL로 테스트
|
|
```bash
|
|
# 바이너리 파일만 업로드
|
|
curl -X POST "http://localhost:8000/image/upload/blob/test-task-001" \\
|
|
-F "files=@/path/to/image1.jpg" \\
|
|
-F "files=@/path/to/image2.png"
|
|
|
|
# URL + 바이너리 파일 동시 업로드
|
|
curl -X POST "http://localhost:8000/image/upload/blob/test-task-001" \\
|
|
-F 'images_json=[{"url":"https://example.com/image.jpg"}]' \\
|
|
-F "files=@/path/to/local_image.jpg"
|
|
```
|
|
|
|
## 반환 정보
|
|
- **task_id**: 작업 고유 식별자
|
|
- **total_count**: 총 업로드된 이미지 개수
|
|
- **url_count**: URL로 등록된 이미지 개수
|
|
- **file_count**: 파일로 업로드된 이미지 개수 (Blob에 저장됨)
|
|
- **images**: 업로드된 이미지 목록
|
|
|
|
## 저장 경로
|
|
- 바이너리 파일: Azure Blob Storage ({task_id}/{파일명})
|
|
""",
|
|
response_model=ImageUploadResponse,
|
|
responses={
|
|
200: {"description": "이미지 업로드 성공"},
|
|
400: {"description": "이미지가 제공되지 않음", "model": ErrorResponse},
|
|
},
|
|
tags=["image"],
|
|
)
|
|
async def upload_images_blob(
|
|
task_id: str,
|
|
images_json: Optional[str] = Form(
|
|
default=None,
|
|
description="외부 이미지 URL 목록 (JSON 문자열)",
|
|
example=IMAGES_JSON_EXAMPLE,
|
|
),
|
|
files: Optional[list[UploadFile]] = File(
|
|
default=None, description="이미지 바이너리 파일 목록"
|
|
),
|
|
session: AsyncSession = Depends(get_session),
|
|
) -> ImageUploadResponse:
|
|
"""이미지 업로드 (URL + Azure Blob Storage)"""
|
|
print(f"[upload_images_blob] START - task_id: {task_id}")
|
|
|
|
# 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:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="images_json 또는 files 중 하나는 반드시 제공해야 합니다.",
|
|
)
|
|
|
|
# 2. images_json 파싱
|
|
url_images: list[ImageUrlItem] = []
|
|
if has_images_json:
|
|
try:
|
|
parsed = json.loads(images_json)
|
|
if isinstance(parsed, list):
|
|
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)}",
|
|
)
|
|
|
|
# 3. 유효한 파일만 필터링
|
|
valid_files: list[UploadFile] = []
|
|
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:
|
|
valid_files.append(f)
|
|
else:
|
|
skipped_files.append(f.filename or "unknown")
|
|
|
|
print(f"[upload_images_blob] valid_files: {len(valid_files)}, url_images: {len(url_images)}")
|
|
|
|
if not url_images and not valid_files:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"유효한 이미지가 없습니다. 지원 확장자: {', '.join(ALLOWED_IMAGE_EXTENSIONS)}. 건너뛴 파일: {skipped_files}",
|
|
)
|
|
|
|
result_images: list[ImageUploadResultItem] = []
|
|
img_order = 0
|
|
|
|
# 1. URL 이미지 저장
|
|
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,
|
|
img_url=url_item.url,
|
|
img_order=img_order,
|
|
)
|
|
session.add(image)
|
|
await session.flush()
|
|
print(f"[upload_images_blob] URL saved - id: {image.id}, img_name: {img_name}")
|
|
|
|
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
|
|
|
|
# 2. 바이너리 파일을 Azure Blob Storage에 직접 업로드 (media 저장 없음)
|
|
if valid_files:
|
|
uploader = AzureBlobUploader(task_id=task_id)
|
|
|
|
for file in valid_files:
|
|
original_name = file.filename or "image"
|
|
ext = _get_file_extension(file.filename) # type: ignore[arg-type]
|
|
name_without_ext = (
|
|
original_name.rsplit(".", 1)[0]
|
|
if "." in original_name
|
|
else original_name
|
|
)
|
|
filename = f"{name_without_ext}_{img_order:03d}{ext}"
|
|
|
|
# 파일 내용 읽기
|
|
file_content = await file.read()
|
|
print(f"[upload_images_blob] Uploading {filename} ({len(file_content)} bytes) to Blob...")
|
|
|
|
# Azure Blob Storage에 직접 업로드
|
|
upload_success = await uploader.upload_image_bytes(file_content, filename)
|
|
|
|
if upload_success:
|
|
blob_url = uploader.public_url
|
|
img_name = file.filename or filename
|
|
|
|
image = Image(
|
|
task_id=task_id,
|
|
img_name=img_name,
|
|
img_url=blob_url,
|
|
img_order=img_order,
|
|
)
|
|
session.add(image)
|
|
await session.flush()
|
|
print(f"[upload_images_blob] Blob saved - id: {image.id}, blob_url: {blob_url}")
|
|
|
|
result_images.append(
|
|
ImageUploadResultItem(
|
|
id=image.id,
|
|
img_name=img_name,
|
|
img_url=blob_url,
|
|
img_order=img_order,
|
|
source="blob",
|
|
)
|
|
)
|
|
img_order += 1
|
|
else:
|
|
print(f"[upload_images_blob] Failed to upload {filename}")
|
|
skipped_files.append(filename)
|
|
|
|
saved_count = len(result_images)
|
|
print(f"[upload_images_blob] Committing {saved_count} images...")
|
|
await session.commit()
|
|
print(f"[upload_images_blob] Done! saved_count: {saved_count}")
|
|
|
|
return ImageUploadResponse(
|
|
task_id=task_id,
|
|
total_count=len(result_images),
|
|
url_count=len(url_images),
|
|
file_count=len(valid_files) - len(skipped_files),
|
|
saved_count=saved_count,
|
|
images=result_images,
|
|
)
|