diff --git a/app/home/api/routers/v1/home.py b/app/home/api/routers/v1/home.py index fa6a6de..701a354 100644 --- a/app/home/api/routers/v1/home.py +++ b/app/home/api/routers/v1/home.py @@ -1,863 +1,932 @@ -import json -import time -from datetime import date -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 -from sqlalchemy import func, select - -from app.database.session import get_session, AsyncSessionLocal -from app.home.models import Image, MarketingIntel, ImageTag -from app.user.dependencies.auth import get_current_user -from app.user.models import User -from app.home.schemas.home_schema import ( - AutoCompleteRequest, - AccommodationSearchItem, - AccommodationSearchResponse, - CrawlingRequest, - 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.utils.prompts.chatgpt_prompt import ChatgptService, ChatGPTResponseError -from app.utils.common import generate_task_id -from app.utils.logger import get_logger -from app.utils.nvMapScraper import NvMapScraper, GraphQLException, URLNotFoundException -from app.utils.nvMapPwScraper import NvMapPwScraper -from app.utils.prompts.prompts import marketing_prompt -from app.utils.address_parser import extract_region_from_address -from app.utils.autotag import autotag_images -from config import MEDIA_ROOT - -logger = get_logger("home") - -router = APIRouter() - - -@router.get( - "/search/accommodation", - summary="장소 자동완성 검색 (숙박/음식점 등)", - description=""" -네이버 지역 검색 API를 이용한 장소 자동완성 검색입니다. - -## 요청 파라미터 -- **query**: 검색어 (필수) -- **category**: 카테고리 - -## 반환 정보 -- **query**: 검색어 -- **count**: 검색 결과 수 (최대 10개) -- **items**: 검색 결과 목록 - - **title**: 장소명 (HTML 태그 포함 가능) - - **address**: 지번 주소 - - **roadAddress**: 도로명 주소 - """, - response_model=AccommodationSearchResponse, - responses={ - 200: {"description": "검색 성공", "model": AccommodationSearchResponse}, - }, - tags=["Search"], -) -async def search_accommodation( - query: str, -) -> AccommodationSearchResponse: - """장소 자동완성 검색""" - results = await naver_search_client.search_accommodation( - query=query, - display=10, - ) - - items = [AccommodationSearchItem(**item) for item in results] - - return AccommodationSearchResponse( - query=query, - count=len(items), - items=items, - ) - - -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"] - - -async def _resolve_industry(category: str, customer_name: str = "") -> str: - """업체를 통합 프롬프트의 8개 industry enum 중 하나로 AI 분류. - - 분류 근거는 카테고리를 우선 사용하고, 카테고리가 없으면 업체명으로 분류한다. - 근거가 둘 다 없거나 분류 실패 시 빈 문자열 반환(프롬프트가 범용 처리). - """ - if category: - basis_label, basis_value = "네이버 지도 카테고리", category - elif customer_name: - basis_label, basis_value = "업체명", customer_name - else: - return "" - try: - chatgpt = ChatgptService() - prompt = ( - f"{basis_label}: '{basis_value}'\n" - "위 정보를 바탕으로 이 업체를 다음 8개 업종 중 가장 적합한 하나로 분류하세요: " - "stay(숙박/펜션/호텔), restaurant(음식점), cafe(카페/디저트), " - "salon(미용실/네일/뷰티), clinic(병원/의원/치과), fitness(헬스/필라테스/요가), " - "academy(학원/교습소), attraction(관광/체험/액티비티)." - ) - result = await chatgpt._call_pydantic_output_chat_completion( - prompt=prompt, - output_format=_IndustryOutput, - model="gpt-4o-mini", - img_url=None, - image_detail_high=False, - ) - return result.industry - except Exception as e: - logger.warning(f"[industry] AI 분류 실패, 빈 값 반환: {e}") - 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, - }, - 502: { - "description": "크롤링 실패", - "model": ErrorResponse, - }, - }, - tags=["Crawling"], -) -async def crawling( - request_body: CrawlingRequest, - session: AsyncSession = Depends(get_session)): - return await _crawling_logic(request_body.url, session) - -@router.post( - "/autocomplete", - summary="네이버 자동완성 크롤링", - description=""" -네이버 검색 API 정보를 활용하여 Place ID를 추출한 뒤 자동으로 크롤링합니다. - -## 요청 필드 -- **title**: 네이버 검색 API Place 결과물 title (필수) -- **address**: 네이버 검색 API Place 결과물 지번주소 (필수) -- **roadAddress**:네이버 검색 API Place 결과물 도로명주소 - -## 반환 정보 -- **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, - }, - 502: { - "description": "크롤링 실패", - "model": ErrorResponse, - }, - }, - tags=["Crawling"], -) -async def autocomplete_crawling( - 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): - request_start = time.perf_counter() - logger.info("[crawling] ========== START ==========") - logger.info(f"[crawling] URL: {url[:80]}...") - - # ========== Step 1: 네이버 지도 크롤링 ========== - step1_start = time.perf_counter() - logger.info("[crawling] Step 1: 네이버 지도 크롤링 시작...") - - try: - scraper = NvMapScraper(url) - await scraper.scrap() - except GraphQLException as e: - step1_elapsed = (time.perf_counter() - step1_start) * 1000 - logger.error( - f"[crawling] Step 1 FAILED - GraphQL 크롤링 실패: {e} ({step1_elapsed:.1f}ms)" - ) - raise HTTPException( - status_code=status.HTTP_502_BAD_GATEWAY, - detail=f"네이버 지도 크롤링에 실패했습니다: {e}", - ) - except URLNotFoundException as e: - step1_elapsed = (time.perf_counter() - step1_start) * 1000 - logger.error( - f"[crawling] Step 1 FAILED - 크롤링 실패: {e} ({step1_elapsed:.1f}ms)" - ) - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"Place ID를 확인할 수 없습니다. URL을 확인하세요. : {e}", - ) - except Exception as e: - step1_elapsed = (time.perf_counter() - step1_start) * 1000 - logger.error( - f"[crawling] Step 1 FAILED - 크롤링 중 예기치 않은 오류: {e} ({step1_elapsed:.1f}ms)" - ) - logger.exception("[crawling] Step 1 상세 오류:") - raise HTTPException( - status_code=status.HTTP_502_BAD_GATEWAY, - detail="네이버 지도 크롤링 중 오류가 발생했습니다.", - ) - - step1_elapsed = (time.perf_counter() - step1_start) * 1000 - image_count = len(scraper.image_link_list) if scraper.image_link_list else 0 - logger.info( - f"[crawling] Step 1 완료 - 이미지 {image_count}개 ({step1_elapsed:.1f}ms)" - ) - - # ========== Step 2: 정보 가공 ========== - step2_start = time.perf_counter() - logger.info("[crawling] Step 2: 정보 가공 시작...") - - processed_info = None - marketing_analysis = None - - if scraper.base_info: - road_address = scraper.base_info.get("roadAddress", "") - jibun_address = scraper.base_info.get("address", "") - customer_name = scraper.base_info.get("name", "") - category = scraper.base_info.get("category", "") - region = _extract_region_from_address(road_address, jibun_address) - industry = await _resolve_industry(category, customer_name) - - processed_info = ProcessedInfo( - customer_name=customer_name, - region=region, - detail_region_info=road_address or jibun_address or "", - industry=industry, - ) - - step2_elapsed = (time.perf_counter() - step2_start) * 1000 - logger.info( - f"[crawling] Step 2 완료 - {customer_name}, {region}, " - f"category={category!r}, place_type={scraper.place_type!r}, industry={industry!r} ({step2_elapsed:.1f}ms)" - ) - - # ========== Step 3: ChatGPT 마케팅 분석 ========== - step3_start = time.perf_counter() - logger.info("[crawling] Step 3: ChatGPT 마케팅 분석 시작...") - - try: - # Step 3-1: ChatGPT 서비스 초기화 및 입력 데이터 구성 - chatgpt_service = ChatgptService() - input_marketing_data = { - "customer_name": customer_name, - "region": region, - "detail_region_info": road_address or "", - "industry": industry, # 통합 프롬프트 내부 분기용 업종 enum - } - - # Step 3-2: GPT API 호출 → 구조화된 마케팅 분석 결과 반환 - marketing_analysis = await chatgpt_service.generate_structured_output( - marketing_prompt, input_marketing_data - ) - - # Step 3-3: 분석 결과 DB 저장 (industry는 Project로 흐르므로 여기엔 미저장) - marketing_intel = MarketingIntel( - place_id=scraper.place_id, - intel_result=marketing_analysis.model_dump(), - ) - session.add(marketing_intel) - 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}") - - step3_elapsed = (time.perf_counter() - step3_start) * 1000 - logger.info( - f"[crawling] Step 3 완료 - 마케팅 분석 성공 ({step3_elapsed:.1f}ms)" - ) - - except ChatGPTResponseError as e: - step3_elapsed = (time.perf_counter() - step3_start) * 1000 - logger.error( - f"[crawling] Step 3 FAILED - ChatGPT Error: status={e.status}, " - f"code={e.error_code}, message={e.error_message} ({step3_elapsed:.1f}ms)" - ) - marketing_analysis = None - gpt_status = "failed" - - except Exception as e: - step3_elapsed = (time.perf_counter() - step3_start) * 1000 - logger.error( - f"[crawling] Step 3 FAILED - GPT 마케팅 분석 중 오류: {e} ({step3_elapsed:.1f}ms)" - ) - logger.exception("[crawling] Step 3 상세 오류:") - marketing_analysis = None - gpt_status = "failed" - else: - step2_elapsed = (time.perf_counter() - step2_start) * 1000 - logger.warning( - f"[crawling] Step 2 - base_info 없음, 마케팅 분석 스킵 ({step2_elapsed:.1f}ms)" - ) - - # ========== 완료 ========== - total_elapsed = (time.perf_counter() - request_start) * 1000 - logger.info("[crawling] ========== COMPLETE ==========") - logger.info(f"[crawling] 총 소요시간: {total_elapsed:.1f}ms") - logger.info(f"[crawling] - Step 1 (크롤링): {step1_elapsed:.1f}ms") - if scraper.base_info: - logger.info(f"[crawling] - Step 2 (정보가공): {step2_elapsed:.1f}ms") - if "step3_elapsed" in locals(): - logger.info(f"[crawling] - Step 3 (GPT 분석): {step3_elapsed:.1f}ms") - - return { - "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 "", - } - - - -@router.post( - "/marketing", - summary="업체명+주소 직접 입력 마케팅 분석", - description=""" -네이버 크롤링 없이 업체명과 주소를 직접 입력받아 마케팅 분석을 수행합니다. - -## 요청 필드 -- **customer_name**: 업체명 / 브랜드명 (필수) -- **address**: 도로명 또는 지번 주소 (필수) - -## 반환 정보 -- **processed_info**: 가공된 장소 정보 (customer_name, region, detail_region_info) -- **marketing_analysis**: ChatGPT 마케팅 분석 결과 -- **m_id**: 마케팅 분석 결과 ID (이후 영상생성 파이프라인에 사용) - """, - response_model=CrawlingResponse, - tags=["Marketing"], -) -async def manual_marketing( - request_body: ManualMarketingRequest, - session: AsyncSession = Depends(get_session), -): - # Step 1: 주소에서 지역명 추출 및 processed_info 구성 - region = _extract_region_from_address(request_body.address) - processed_info = ProcessedInfo( - customer_name=request_body.store_name, - region=region, - detail_region_info=request_body.address, - ) - try: - # Step 2: GPT API 호출 → 마케팅 분석 결과 생성 - # place_id 없이 업체명+주소만으로 분석 (크롤링 없이 직접 입력된 경우) - chatgpt_service = ChatgptService() - # 요청에 industry가 있으면 사용, 없으면 업체명 기반 AI 분류 - industry = request_body.industry or await _resolve_industry( - "", request_body.store_name - ) - processed_info.industry = industry - input_marketing_data = { - "customer_name": request_body.store_name, - "region": region, - "detail_region_info": request_body.address, - "industry": industry, # 통합 프롬프트 내부 분기용 업종 enum - } - marketing_analysis = await chatgpt_service.generate_structured_output( - marketing_prompt, input_marketing_data - ) - - # Step 3: 분석 결과 DB 저장 (place_id=None — 네이버 장소와 연결되지 않음) - marketing_intel = MarketingIntel( - place_id=None, - intel_result=marketing_analysis.model_dump(), - ) - session.add(marketing_intel) - await session.commit() - await session.refresh(marketing_intel) - m_id = marketing_intel.id - logger.debug(f"[MarketingPrompt] INSERT id={marketing_intel.id}") - except ChatGPTResponseError as e: - logger.error( - f"[marketing] ChatGPT Error: status={e.status}, " - f"code={e.error_code}, message={e.error_message}" - ) - raise HTTPException( - status_code=status.HTTP_502_BAD_GATEWAY, - detail=f"마케팅 분석 중 ChatGPT 오류가 발생했습니다: {e.error_message}", - ) - except Exception as e: - logger.error(f"[marketing] 마케팅 분석 중 오류: {e}") - logger.exception("[marketing] 상세 오류:") - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail="마케팅 분석 중 오류가 발생했습니다.", - ) - return CrawlingResponse( - status="completed", - processed_info=processed_info, - marketing_analysis=marketing_analysis, - m_id=m_id, - industry=industry, - ) - - -async def _autocomplete_logic(autocomplete_item:dict): - step1_start = time.perf_counter() - try: - async with NvMapPwScraper() as pw_scraper: - new_url = await pw_scraper.get_place_id_url(autocomplete_item) - except Exception as e: - step1_elapsed = (time.perf_counter() - step1_start) * 1000 - logger.error( - f"[crawling] Autocomplete FAILED - 자동완성 예기치 않은 오류: {e} ({step1_elapsed:.1f}ms)" - ) - logger.exception("[crawling] Autocomplete 상세 오류:") - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail="자동완성 place id 추출 실패", - ) - - if not new_url: - step1_elapsed = (time.perf_counter() - step1_start) * 1000 - logger.error( - f"[crawling] Autocomplete FAILED - URL을 찾을 수 없음 ({step1_elapsed:.1f}ms)" - ) - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail="해당 장소의 네이버 지도 URL을 찾을 수 없습니다.", - ) - - return new_url - -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/blob", - summary="이미지 업로드 (Azure Blob Storage)", - description=""" -이미지를 Azure Blob Storage에 업로드하고 새로운 task_id를 생성합니다. -바이너리 파일은 로컬 서버에 저장하지 않고 Azure Blob에 직접 업로드됩니다. - -## 인증 -**Bearer 토큰 필수** - `Authorization: Bearer {access_token}` 헤더를 포함해야 합니다. - -## 요청 방식 -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" \\ - -H "Authorization: Bearer {access_token}" \\ - -F "files=@/path/to/image1.jpg" \\ - -F "files=@/path/to/image2.png" - -# URL + 바이너리 파일 동시 업로드 -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**: 새로 생성된 작업 고유 식별자 -- **total_count**: 총 업로드된 이미지 개수 -- **url_count**: URL로 등록된 이미지 개수 (Image 테이블에 외부 URL 그대로 저장) -- **file_count**: 파일로 업로드된 이미지 개수 (Azure Blob Storage에 저장) -- **saved_count**: Image 테이블에 저장된 row 수 -- **images**: 업로드된 이미지 목록 - - **source**: "url" (외부 URL) 또는 "blob" (Azure Blob Storage) -- **image_urls**: Image 테이블에 저장된 현재 task_id의 이미지 URL 목록 - -## 저장 경로 -- 바이너리 파일: Azure Blob Storage ({BASE_URL}/{task_id}/image/{파일명}) -- URL 이미지: 외부 URL 그대로 Image 테이블에 저장 - """, - response_model=ImageUploadResponse, - responses={ - 200: {"description": "이미지 업로드 성공"}, - 400: {"description": "이미지가 제공되지 않음", "model": ErrorResponse}, - 401: {"description": "인증 실패 (토큰 없음/만료)"}, - }, - tags=["Image-Blob"], - openapi_extra={ - "requestBody": { - "content": { - "multipart/form-data": { - "encoding": {"files": {"contentType": "application/octet-stream"}} - } - } - } - }, -) -async def upload_images_blob( - images_json: Optional[str] = Form( - default=None, - description="외부 이미지 URL 목록 (JSON 문자열)", - examples=[IMAGES_JSON_EXAMPLE], - ), - files: Optional[list[UploadFile]] = File( - default=None, - description="이미지 바이너리 파일 목록", - ), - current_user: User = Depends(get_current_user), -) -> ImageUploadResponse: - """이미지 업로드 (URL + Azure Blob Storage) - - 3단계로 분리하여 세션 점유 시간 최소화: - - Stage 1: 입력 검증 및 파일 데이터 준비 (세션 없음) - - Stage 2: Azure Blob 업로드 (세션 없음) - - Stage 3: DB 저장 (새 세션으로 빠르게 처리) - """ - request_start = time.perf_counter() - - # task_id 생성 - task_id = await generate_task_id() - logger.info(f"[upload_images_blob] START - task_id: {task_id}") - - # ========== 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: - 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] - 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}" - ) - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=detail, - ) - - 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"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 이미지 다음 순서부터 시작 - - 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 - ) - 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)" - ) - - # Azure Blob Storage에 직접 업로드 - upload_success = await uploader.upload_image_bytes(file_content, filename) - - if upload_success: - blob_url = uploader.public_url - blob_upload_results.append((original_name, blob_url)) - img_order += 1 - logger.debug( - f"[upload_images_blob] File {idx + 1}/{total_files} SUCCESS" - ) - else: - skipped_files.append(filename) - logger.warning( - f"[upload_images_blob] File {idx + 1}/{total_files} FAILED" - ) - - 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...") - result_images: list[ImageUploadResultItem] = [] - img_order = 0 - - try: - async with AsyncSessionLocal() as session: - # 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() - - 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: - image = Image( - task_id=task_id, - img_name=img_name, - img_url=blob_url, - 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", - ) - ) - img_order += 1 - - await session.commit() - stage3_time = time.perf_counter() - logger.info( - f"[upload_images_blob] Stage 3 done - " - f"saved: {len(result_images)}, " - f"elapsed: {(stage3_time - stage2_time) * 1000:.1f}ms" - ) - - except SQLAlchemyError as e: - logger.error(f"[upload_images_blob] DB Error - task_id: {task_id}, error: {e}") - logger.exception("[upload_images_blob] DB 상세 오류:") - raise HTTPException( - status_code=status.HTTP_503_SERVICE_UNAVAILABLE, - detail="이미지 저장 중 데이터베이스 오류가 발생했습니다.", - ) - except Exception as e: - logger.error( - f"[upload_images_blob] Stage 3 EXCEPTION - " - f"task_id: {task_id}, error: {type(e).__name__}: {e}" - ) - logger.exception("[upload_images_blob] Stage 3 상세 오류:") - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail="이미지 업로드 중 오류가 발생했습니다.", - ) - - 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, clear_old_tags=True) - logger.info(f"[image_tagging] Done - task_id: {task_id}") - - 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" - ) - - return ImageUploadResponse( - task_id=task_id, - total_count=len(result_images), - url_count=len(url_images), - file_count=len(blob_upload_results), - saved_count=saved_count, - images=result_images, - image_urls=image_urls, - ) - - -async def tagging_images( - image_urls : list[str], - clear_old_tags : bool = False - ) -> None: - # 1. 조회 - async with AsyncSessionLocal() as session: - stmt = ( - select(ImageTag) - .where(ImageTag.img_url_hash.in_([func.crc32(url) for url in image_urls])) - .where(ImageTag.img_url.in_(image_urls)) - ) - image_tags_query_results = await session.execute(stmt) - image_tags = image_tags_query_results.scalars().all() - existing_urls = {tag.img_url for tag in image_tags} - new_imt = [ - ImageTag(img_url=url, img_tag=None) - for url in image_urls - if url not in existing_urls - ] - if clear_old_tags: - for tag in image_tags: - tag.img_tag = None - 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]) - print(tag_datas) - - async with AsyncSessionLocal() as session: - for tag, tag_data in zip(null_imts, tag_datas): - if isinstance(tag_data, Exception): - continue - tag.img_tag = tag_data.model_dump(mode="json") - session.add(tag) - await session.commit() +import json +import time +from datetime import date +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 +from sqlalchemy import func, select + +from app.database.session import get_session, AsyncSessionLocal +from app.home.models import Image, MarketingIntel, ImageTag +from app.user.dependencies.auth import get_current_user +from app.user.models import User +from app.home.schemas.home_schema import ( + AutoCompleteRequest, + AccommodationSearchItem, + AccommodationSearchResponse, + CrawlingRequest, + 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.utils.prompts.chatgpt_prompt import ChatgptService, ChatGPTResponseError +from app.utils.common import generate_task_id +from app.utils.logger import get_logger +from app.utils.nvMapScraper import NvMapScraper, GraphQLException, URLNotFoundException +from app.utils.nvMapPwScraper import NvMapPwScraper +from app.utils.prompts.prompts import marketing_prompt +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 + +logger = get_logger("home") + +router = APIRouter() + + +@router.get( + "/search/accommodation", + summary="장소 자동완성 검색 (숙박/음식점 등)", + description=""" +네이버 지역 검색 API를 이용한 장소 자동완성 검색입니다. + +## 요청 파라미터 +- **query**: 검색어 (필수) +- **category**: 카테고리 + +## 반환 정보 +- **query**: 검색어 +- **count**: 검색 결과 수 (최대 10개) +- **items**: 검색 결과 목록 + - **title**: 장소명 (HTML 태그 포함 가능) + - **address**: 지번 주소 + - **roadAddress**: 도로명 주소 + """, + response_model=AccommodationSearchResponse, + responses={ + 200: {"description": "검색 성공", "model": AccommodationSearchResponse}, + }, + tags=["Search"], +) +async def search_accommodation( + query: str, +) -> AccommodationSearchResponse: + """장소 자동완성 검색""" + results = await naver_search_client.search_accommodation( + query=query, + display=10, + ) + + items = [AccommodationSearchItem(**item) for item in results] + + return AccommodationSearchResponse( + query=query, + count=len(items), + items=items, + ) + + +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"] + + +async def _resolve_industry(category: str, customer_name: str = "") -> str: + """업체를 통합 프롬프트의 8개 industry enum 중 하나로 AI 분류. + + 분류 근거는 카테고리를 우선 사용하고, 카테고리가 없으면 업체명으로 분류한다. + 근거가 둘 다 없으면 빈 문자열 반환. API 장애 시 예외 전파(하위 분석도 동일 API 의존). + """ + if category: + basis_label, basis_value = "네이버 지도 카테고리", category + elif customer_name: + basis_label, basis_value = "업체명", customer_name + else: + return "" + chatgpt = ChatgptService() + prompt = ( + f"{basis_label}: '{basis_value}'\n" + "위 정보를 바탕으로 이 업체를 다음 8개 업종 중 가장 적합한 하나로 분류하세요: " + "stay(숙박/펜션/호텔), restaurant(음식점), cafe(카페/디저트), " + "salon(미용실/네일/뷰티), clinic(병원/의원/치과), fitness(헬스/필라테스/요가), " + "academy(학원/교습소), attraction(관광/체험/액티비티)." + ) + result = await chatgpt._call_pydantic_output_chat_completion( + prompt=prompt, + output_format=_IndustryOutput, + model="gpt-4o-mini", + img_url=None, + image_detail_high=False, + ) + return result.industry + + +@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, + }, + 502: { + "description": "크롤링 실패", + "model": ErrorResponse, + }, + }, + tags=["Crawling"], +) +async def crawling( + request_body: CrawlingRequest, + session: AsyncSession = Depends(get_session)): + return await _crawling_logic(request_body.url, session) + +@router.post( + "/autocomplete", + summary="네이버 자동완성 크롤링", + description=""" +네이버 검색 API 정보를 활용하여 Place ID를 추출한 뒤 자동으로 크롤링합니다. + +## 요청 필드 +- **title**: 네이버 검색 API Place 결과물 title (필수) +- **address**: 네이버 검색 API Place 결과물 지번주소 (필수) +- **roadAddress**:네이버 검색 API Place 결과물 도로명주소 + +## 반환 정보 +- **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, + }, + 502: { + "description": "크롤링 실패", + "model": ErrorResponse, + }, + }, + tags=["Crawling"], +) +async def autocomplete_crawling( + 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): + request_start = time.perf_counter() + logger.info("[crawling] ========== START ==========") + logger.info(f"[crawling] URL: {url[:80]}...") + + # ========== Step 1: 네이버 지도 크롤링 ========== + step1_start = time.perf_counter() + logger.info("[crawling] Step 1: 네이버 지도 크롤링 시작...") + + try: + scraper = NvMapScraper(url) + await scraper.scrap() + except GraphQLException as e: + step1_elapsed = (time.perf_counter() - step1_start) * 1000 + logger.error( + f"[crawling] Step 1 FAILED - GraphQL 크롤링 실패: {e} ({step1_elapsed:.1f}ms)" + ) + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, + detail=f"네이버 지도 크롤링에 실패했습니다: {e}", + ) + except URLNotFoundException as e: + step1_elapsed = (time.perf_counter() - step1_start) * 1000 + logger.error( + f"[crawling] Step 1 FAILED - 크롤링 실패: {e} ({step1_elapsed:.1f}ms)" + ) + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Place ID를 확인할 수 없습니다. URL을 확인하세요. : {e}", + ) + except Exception as e: + step1_elapsed = (time.perf_counter() - step1_start) * 1000 + logger.error( + f"[crawling] Step 1 FAILED - 크롤링 중 예기치 않은 오류: {e} ({step1_elapsed:.1f}ms)" + ) + logger.exception("[crawling] Step 1 상세 오류:") + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, + detail="네이버 지도 크롤링 중 오류가 발생했습니다.", + ) + + step1_elapsed = (time.perf_counter() - step1_start) * 1000 + logger.info( + f"[crawling] Step 1 완료 - 업체사진 {len(scraper.owner_images 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 + + if scraper.base_info: + road_address = scraper.base_info.get("roadAddress", "") + jibun_address = scraper.base_info.get("address", "") + customer_name = scraper.base_info.get("name", "") + category = scraper.base_info.get("category", "") + region = _extract_region_from_address(road_address, jibun_address) + try: + # industry는 마케팅 적합성 필터(Step 3)와 ProcessedInfo 양쪽에 필요하므로 + # 이미지 필터링보다 먼저 계산한다. + industry = await _resolve_industry(category, customer_name) + except ChatGPTResponseError as e: + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, + detail=f"업종 분류 중 ChatGPT 오류가 발생했습니다: {e.error_message}", + ) + except Exception as e: + logger.error(f"[crawling] Step 2 FAILED - 업종 분류 오류: {e}") + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, + detail="업종 분류 중 오류가 발생했습니다.", + ) + + processed_info = ProcessedInfo( + customer_name=customer_name, + region=region, + detail_region_info=road_address or jibun_address or "", + industry=industry, + ) + + 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장) 이상이면 보충이 불필요하므로 + # 방문자 사진 필터링(Gemini 호출) 자체를 건너뛴다. + step3_start = time.perf_counter() + owner_images = scraper.owner_images or [] + extra_photo_urls = scraper.extra_photo_urls or [] + + if len(owner_images) >= NvMapScraper.SUPPLEMENT_THRESHOLD: + scraper.image_link_list = owner_images[: NvMapScraper.MAX_IMAGES] + step3_elapsed = (time.perf_counter() - step3_start) * 1000 + logger.info( + 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: + 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 + ) + passed_count = sum(extra_pass_flags) + step3_elapsed = (time.perf_counter() - step3_start) * 1000 + logger.info( + 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장입니다.") + + # ========== Step 4: ChatGPT 마케팅 분석 ========== + step4_start = time.perf_counter() + logger.info("[crawling] Step 4: ChatGPT 마케팅 분석 시작...") + + try: + # Step 4-1: ChatGPT 서비스 초기화 및 입력 데이터 구성 + chatgpt_service = ChatgptService() + input_marketing_data = { + "customer_name": customer_name, + "region": region, + "detail_region_info": road_address or "", + "industry": industry, # 통합 프롬프트 내부 분기용 업종 enum + } + + # Step 4-2: GPT API 호출 → 구조화된 마케팅 분석 결과 반환 + marketing_analysis = await chatgpt_service.generate_structured_output( + marketing_prompt, input_marketing_data + ) + + # Step 4-3: 분석 결과 DB 저장 (industry는 Project로 흐르므로 여기엔 미저장) + marketing_intel = MarketingIntel( + place_id=scraper.place_id, + intel_result=marketing_analysis.model_dump(), + ) + session.add(marketing_intel) + 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}") + + 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 + logger.error( + f"[crawling] Step 4 FAILED - ChatGPT Error: status={e.status}, " + f"code={e.error_code}, message={e.error_message} ({step4_elapsed:.1f}ms)" + ) + marketing_analysis = None + gpt_status = "failed" + + except Exception as e: + step4_elapsed = (time.perf_counter() - step4_start) * 1000 + logger.error( + f"[crawling] Step 4 FAILED - GPT 마케팅 분석 중 오류: {e} ({step4_elapsed:.1f}ms)" + ) + logger.exception("[crawling] Step 4 상세 오류:") + marketing_analysis = None + gpt_status = "failed" + else: + step2_elapsed = (time.perf_counter() - step2_start) * 1000 + logger.warning( + f"[crawling] Step 2 - base_info 없음, 마케팅 분석 스킵 ({step2_elapsed:.1f}ms)" + ) + + # ========== 완료 ========== + total_elapsed = (time.perf_counter() - request_start) * 1000 + logger.info("[crawling] ========== COMPLETE ==========") + logger.info(f"[crawling] 총 소요시간: {total_elapsed:.1f}ms") + logger.info(f"[crawling] - Step 1 (크롤링): {step1_elapsed:.1f}ms") + if scraper.base_info: + logger.info(f"[crawling] - Step 2 (정보가공): {step2_elapsed:.1f}ms") + if "step3_elapsed" in locals(): + logger.info(f"[crawling] - Step 3 (이미지 필터링): {step3_elapsed:.1f}ms") + 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", + "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 "", + } + + + +@router.post( + "/marketing", + summary="업체명+주소 직접 입력 마케팅 분석", + description=""" +네이버 크롤링 없이 업체명과 주소를 직접 입력받아 마케팅 분석을 수행합니다. + +## 요청 필드 +- **customer_name**: 업체명 / 브랜드명 (필수) +- **address**: 도로명 또는 지번 주소 (필수) + +## 반환 정보 +- **processed_info**: 가공된 장소 정보 (customer_name, region, detail_region_info) +- **marketing_analysis**: ChatGPT 마케팅 분석 결과 +- **m_id**: 마케팅 분석 결과 ID (이후 영상생성 파이프라인에 사용) + """, + response_model=CrawlingResponse, + tags=["Marketing"], +) +async def manual_marketing( + request_body: ManualMarketingRequest, + session: AsyncSession = Depends(get_session), +): + # Step 1: 주소에서 지역명 추출 및 processed_info 구성 + region = _extract_region_from_address(request_body.address) + processed_info = ProcessedInfo( + customer_name=request_body.store_name, + region=region, + detail_region_info=request_body.address, + ) + try: + # Step 2: GPT API 호출 → 마케팅 분석 결과 생성 + # place_id 없이 업체명+주소만으로 분석 (크롤링 없이 직접 입력된 경우) + chatgpt_service = ChatgptService() + # 요청에 industry가 있으면 사용, 없으면 업체명 기반 AI 분류 + industry = request_body.industry or await _resolve_industry( + "", request_body.store_name + ) + processed_info.industry = industry + input_marketing_data = { + "customer_name": request_body.store_name, + "region": region, + "detail_region_info": request_body.address, + "industry": industry, # 통합 프롬프트 내부 분기용 업종 enum + } + marketing_analysis = await chatgpt_service.generate_structured_output( + marketing_prompt, input_marketing_data + ) + + # Step 3: 분석 결과 DB 저장 (place_id=None — 네이버 장소와 연결되지 않음) + marketing_intel = MarketingIntel( + place_id=None, + intel_result=marketing_analysis.model_dump(), + ) + session.add(marketing_intel) + await session.commit() + await session.refresh(marketing_intel) + m_id = marketing_intel.id + logger.debug(f"[MarketingPrompt] INSERT id={marketing_intel.id}") + except ChatGPTResponseError as e: + logger.error( + f"[marketing] ChatGPT Error: status={e.status}, " + f"code={e.error_code}, message={e.error_message}" + ) + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, + detail=f"마케팅 분석 중 ChatGPT 오류가 발생했습니다: {e.error_message}", + ) + except Exception as e: + logger.error(f"[marketing] 마케팅 분석 중 오류: {e}") + logger.exception("[marketing] 상세 오류:") + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="마케팅 분석 중 오류가 발생했습니다.", + ) + return CrawlingResponse( + status="completed", + processed_info=processed_info, + marketing_analysis=marketing_analysis, + m_id=m_id, + industry=industry, + ) + + +async def _autocomplete_logic(autocomplete_item:dict): + step1_start = time.perf_counter() + try: + async with NvMapPwScraper() as pw_scraper: + new_url = await pw_scraper.get_place_id_url(autocomplete_item) + except Exception as e: + step1_elapsed = (time.perf_counter() - step1_start) * 1000 + logger.error( + f"[crawling] Autocomplete FAILED - 자동완성 예기치 않은 오류: {e} ({step1_elapsed:.1f}ms)" + ) + logger.exception("[crawling] Autocomplete 상세 오류:") + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="자동완성 place id 추출 실패", + ) + + if not new_url: + step1_elapsed = (time.perf_counter() - step1_start) * 1000 + logger.error( + f"[crawling] Autocomplete FAILED - URL을 찾을 수 없음 ({step1_elapsed:.1f}ms)" + ) + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="해당 장소의 네이버 지도 URL을 찾을 수 없습니다.", + ) + + return new_url + +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/blob", + summary="이미지 업로드 (Azure Blob Storage)", + description=""" +이미지를 Azure Blob Storage에 업로드하고 새로운 task_id를 생성합니다. +바이너리 파일은 로컬 서버에 저장하지 않고 Azure Blob에 직접 업로드됩니다. + +## 인증 +**Bearer 토큰 필수** - `Authorization: Bearer {access_token}` 헤더를 포함해야 합니다. + +## 요청 방식 +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" \\ + -H "Authorization: Bearer {access_token}" \\ + -F "files=@/path/to/image1.jpg" \\ + -F "files=@/path/to/image2.png" + +# URL + 바이너리 파일 동시 업로드 +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**: 새로 생성된 작업 고유 식별자 +- **total_count**: 총 업로드된 이미지 개수 +- **url_count**: URL로 등록된 이미지 개수 (Image 테이블에 외부 URL 그대로 저장) +- **file_count**: 파일로 업로드된 이미지 개수 (Azure Blob Storage에 저장) +- **saved_count**: Image 테이블에 저장된 row 수 +- **images**: 업로드된 이미지 목록 + - **source**: "url" (외부 URL) 또는 "blob" (Azure Blob Storage) +- **image_urls**: Image 테이블에 저장된 현재 task_id의 이미지 URL 목록 + +## 저장 경로 +- 바이너리 파일: Azure Blob Storage ({BASE_URL}/{task_id}/image/{파일명}) +- URL 이미지: 외부 URL 그대로 Image 테이블에 저장 + """, + response_model=ImageUploadResponse, + responses={ + 200: {"description": "이미지 업로드 성공"}, + 400: {"description": "이미지가 제공되지 않음", "model": ErrorResponse}, + 401: {"description": "인증 실패 (토큰 없음/만료)"}, + }, + tags=["Image-Blob"], + openapi_extra={ + "requestBody": { + "content": { + "multipart/form-data": { + "encoding": {"files": {"contentType": "application/octet-stream"}} + } + } + } + }, +) +async def upload_images_blob( + images_json: Optional[str] = Form( + default=None, + description="외부 이미지 URL 목록 (JSON 문자열)", + examples=[IMAGES_JSON_EXAMPLE], + ), + files: Optional[list[UploadFile]] = File( + default=None, + description="이미지 바이너리 파일 목록", + ), + industry: str = Form( + default="", + description="업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction). 크롤링 응답의 industry 값을 그대로 전달", + ), + current_user: User = Depends(get_current_user), +) -> ImageUploadResponse: + """이미지 업로드 (URL + Azure Blob Storage) + + 3단계로 분리하여 세션 점유 시간 최소화: + - Stage 1: 입력 검증 및 파일 데이터 준비 (세션 없음) + - Stage 2: Azure Blob 업로드 (세션 없음) + - Stage 3: DB 저장 (새 세션으로 빠르게 처리) + """ + request_start = time.perf_counter() + + # task_id 생성 + task_id = await generate_task_id() + logger.info(f"[upload_images_blob] START - task_id: {task_id}") + + # ========== 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: + 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] + 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}" + ) + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=detail, + ) + + 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"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 이미지 다음 순서부터 시작 + + 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 + ) + 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)" + ) + + # Azure Blob Storage에 직접 업로드 + upload_success = await uploader.upload_image_bytes(file_content, filename) + + if upload_success: + blob_url = uploader.public_url + blob_upload_results.append((original_name, blob_url)) + img_order += 1 + logger.debug( + f"[upload_images_blob] File {idx + 1}/{total_files} SUCCESS" + ) + else: + skipped_files.append(filename) + logger.warning( + f"[upload_images_blob] File {idx + 1}/{total_files} FAILED" + ) + + 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...") + result_images: list[ImageUploadResultItem] = [] + img_order = 0 + + try: + async with AsyncSessionLocal() as session: + # 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() + + 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: + image = Image( + task_id=task_id, + img_name=img_name, + img_url=blob_url, + 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", + ) + ) + img_order += 1 + + await session.commit() + stage3_time = time.perf_counter() + logger.info( + f"[upload_images_blob] Stage 3 done - " + f"saved: {len(result_images)}, " + f"elapsed: {(stage3_time - stage2_time) * 1000:.1f}ms" + ) + + except SQLAlchemyError as e: + logger.error(f"[upload_images_blob] DB Error - task_id: {task_id}, error: {e}") + logger.exception("[upload_images_blob] DB 상세 오류:") + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="이미지 저장 중 데이터베이스 오류가 발생했습니다.", + ) + except Exception as e: + logger.error( + f"[upload_images_blob] Stage 3 EXCEPTION - " + f"task_id: {task_id}, error: {type(e).__name__}: {e}" + ) + logger.exception("[upload_images_blob] Stage 3 상세 오류:") + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="이미지 업로드 중 오류가 발생했습니다.", + ) + + 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}") + + # 태깅 직후 영상 생성에 사용 가능한 이미지가 하나도 없으면 조기에 실패시킨다. + # (여기서 걸러지지 않으면 훨씬 나중인 영상 생성 단계에서야 슬롯 미배정으로 발견됨) + # 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="영상 생성에 적합한 이미지가 없습니다. 다른 이미지로 다시 업로드해주세요.", + ) + + 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" + ) + + return ImageUploadResponse( + task_id=task_id, + total_count=len(result_images), + url_count=len(url_images), + file_count=len(blob_upload_results), + saved_count=saved_count, + images=result_images, + image_urls=image_urls, + ) + + +async def tagging_images( + image_urls : list[str], + industry: str = "", + clear_old_tags : bool = False + ) -> None: + # 1. 조회 + async with AsyncSessionLocal() as session: + stmt = ( + select(ImageTag) + .where(ImageTag.img_url_hash.in_([func.crc32(url) for url in image_urls])) + .where(ImageTag.img_url.in_(image_urls)) + ) + image_tags_query_results = await session.execute(stmt) + image_tags = image_tags_query_results.scalars().all() + existing_urls = {tag.img_url for tag in image_tags} + new_imt = [ + ImageTag(img_url=url, img_tag=None) + for url in image_urls + if url not in existing_urls + ] + if clear_old_tags: + for tag in image_tags: + tag.img_tag = None + 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) + # print(tag_datas) + + async with AsyncSessionLocal() as session: + for tag, tag_data in zip(null_imts, tag_datas): + if isinstance(tag_data, Exception): + continue + tag.img_tag = tag_data.model_dump(mode="json") + session.add(tag) + await session.commit() diff --git a/app/home/models.py b/app/home/models.py index 73e216b..6a0d9d7 100644 --- a/app/home/models.py +++ b/app/home/models.py @@ -116,7 +116,7 @@ class Project(Base): industry: Mapped[str] = mapped_column( String(50), - nullable=True, + nullable=False, default="", comment="업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction)", ) @@ -314,6 +314,12 @@ class MarketingIntel(Base): comment="자막 정보 생성 결과물", ) + image_match : Mapped[dict[str, Any]] = mapped_column( + JSON, + nullable=True, + comment="이미지 슬롯 배정 결과물 — {슬롯명: image_url, 'audio-music': music_url, ...}", + ) + created_at: Mapped[datetime] = mapped_column( DateTime, nullable=False, diff --git a/app/home/schemas/home_schema.py b/app/home/schemas/home_schema.py index a3d317e..f7c014b 100644 --- a/app/home/schemas/home_schema.py +++ b/app/home/schemas/home_schema.py @@ -98,6 +98,12 @@ class ProcessedInfo(BaseModel): # selling_points: list[str] = Field(default_factory=list, description="추천 부대시설 목록") +class ImageListItem(BaseModel): + """크롤링 이미지 아이템 (미리보기 URL + 원본 URL)""" + preview: str = Field(..., description="미리보기용 CDN URL") + original: str = Field(..., description="업로드용 원본 URL") + + class CrawlingResponse(BaseModel): """크롤링 응답 스키마""" @@ -105,7 +111,7 @@ class CrawlingResponse(BaseModel): json_schema_extra={ "example": { "status": "completed", - "image_list": ["https://example.com/image1.jpg", "https://example.com/image2.jpg"], + "image_list": [{"preview": "https://example.com/image1.jpg", "original": "https://example.com/image1.jpg"}], "image_count": 2, "processed_info": { "customer_name": "스테이 머뭄", @@ -234,7 +240,7 @@ class CrawlingResponse(BaseModel): default="completed", description="처리 상태 (completed: 성공, failed: ChatGPT 분석 실패)" ) - image_list: Optional[list[str]] = Field(None, description="이미지 URL 목록") + image_list: Optional[list[ImageListItem]] = Field(None, description="이미지 목록 (preview/original URL)") image_count: Optional[int] = Field(None, description="이미지 개수") processed_info: Optional[ProcessedInfo] = Field( None, description="가공된 장소 정보 (customer_name, region, detail_region_info)" diff --git a/app/lyric/api/routers/v1/lyric.py b/app/lyric/api/routers/v1/lyric.py index c8ef911..2fbd3ce 100644 --- a/app/lyric/api/routers/v1/lyric.py +++ b/app/lyric/api/routers/v1/lyric.py @@ -42,7 +42,8 @@ from app.lyric.schemas.lyric import ( LyricStatusResponse, SubtitleStatusResponse, ) -from app.lyric.worker.lyric_task import generate_lyric_background, generate_subtitle_background +from app.lyric.worker.lyric_task import generate_lyric_background +from app.video.worker.creative_assets_task import generate_subtitle_background from app.utils.prompts.chatgpt_prompt import ChatgptService from app.utils.logger import get_logger from app.utils.pagination import PaginatedResponse, get_paginated @@ -160,6 +161,7 @@ async def get_lyric_by_task_id( project_id=lyric.project_id, status=lyric.status, lyric_result=lyric.lyric_result, + genre=lyric.genre, created_at=lyric.created_at, ) @@ -269,23 +271,30 @@ async def generate_lyric( step1_start = time.perf_counter() logger.debug(f"[generate_lyric] Step 1: 서비스 초기화 및 프롬프트 생성...") - timing_rules = { - "60s" : """ - 8–12 lines - Full verse flow, immersive mood - """ - } marketing_intel_result = await session.execute(select(MarketingIntel).where(MarketingIntel.id == request_body.m_id)) marketing_intel = marketing_intel_result.scalar_one_or_none() - - + + # 자동 선택(genre 미지정) 재생성 시, 직전에 사용된 장르를 조회해 이번엔 다른 장르가 나오도록 제외 처리 + exclude_genre = "" + if not request_body.genre: + previous_lyric_result = await session.execute( + select(Lyric) + .where(Lyric.task_id == task_id, Lyric.genre.is_not(None)) + .order_by(Lyric.created_at.desc()) + .limit(1) + ) + previous_lyric = previous_lyric_result.scalar_one_or_none() + if previous_lyric: + exclude_genre = previous_lyric.genre + lyric_input_data = { "customer_name" : request_body.customer_name, "region" : request_body.region, "detail_region_info" : request_body.detail_region_info or "", "marketing_intelligence_summary" : json.dumps(marketing_intel.intel_result, ensure_ascii = False), "language" : request_body.language, - "timing_rules" : timing_rules["60s"], # 아직은 선택지 하나 + "genre": request_body.genre or "", # 비어있으면 GPT가 가사 무드에 맞춰 장르를 직접 추천 + "exclude_genre": exclude_genre, # 자동 선택 재생성 시 직전 장르 제외 "industry": request_body.industry, # 크롤 응답에서 받아 전달된 업종 enum } @@ -340,6 +349,7 @@ async def generate_lyric( lyric_prompt=estimated_prompt, lyric_result=None, language=request_body.language, + genre=request_body.genre or None, # 자동 선택인 경우 GPT 응답 완료 후 채워짐 ) session.add(lyric) await session.commit() @@ -575,19 +585,29 @@ async def get_subtitle_status( detail=f"task_id '{task_id}'에 해당하는 마케팅 인텔리전스를 찾을 수 없습니다.", ) - if intel.subtitle: - logger.info(f"[get_subtitle_status] completed - task_id: {task_id}") + # 자막과 이미지 배정 모두 완료돼야 영상 생성 가능 + subtitle_done = bool(intel.subtitle) + image_match_done = bool(intel.image_match) + + if subtitle_done and image_match_done: + logger.info(f"[get_subtitle_status] completed (subtitle+image_match) - task_id: {task_id}") return SubtitleStatusResponse( task_id=task_id, status="completed", - message="자막 생성이 완료되었습니다.", + message="자막 생성 및 이미지 배정이 완료되었습니다.", ) - logger.info(f"[get_subtitle_status] pending - task_id: {task_id}") + pending_parts = [] + if not subtitle_done: + pending_parts.append("자막") + if not image_match_done: + pending_parts.append("이미지 배정") + pending_msg = ", ".join(pending_parts) + logger.info(f"[get_subtitle_status] pending ({pending_msg}) - task_id: {task_id}") return SubtitleStatusResponse( task_id=task_id, status="pending", - message="자막 생성이 진행 중입니다. 잠시 후 다시 확인해주세요.", + message=f"{pending_msg} 처리가 진행 중입니다. 잠시 후 다시 확인해주세요.", ) diff --git a/app/lyric/models.py b/app/lyric/models.py index 75f14aa..25d38c5 100644 --- a/app/lyric/models.py +++ b/app/lyric/models.py @@ -93,6 +93,13 @@ class Lyric(Base): comment="가사 출력 언어 (Korean, English, Chinese, Japanese, Thai, Vietnamese)", ) + genre: Mapped[str | None] = mapped_column( + String(20), + nullable=True, + comment="음악 장르 (kpop, pop, ballad, hip-hop, rnb, edm, jazz, rock). " + "수동 선택 시 요청값, 자동 선택 시 GPT 추천값", + ) + is_deleted: Mapped[bool] = mapped_column( Boolean, nullable=False, diff --git a/app/lyric/schemas/lyric.py b/app/lyric/schemas/lyric.py index 10479bb..59f31bf 100644 --- a/app/lyric/schemas/lyric.py +++ b/app/lyric/schemas/lyric.py @@ -78,6 +78,16 @@ class GenerateLyricRequest(BaseModel): m_id : Optional[int] = Field(None, description="마케팅 인텔리전스 ID 값") industry: str = Field(default="", description="업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction)") instrumental: bool = Field(default=False, description="BGM 전용 모드 (가사 생성 안 함)") + genre: Optional[str] = Field( + None, + description="사용자가 수동으로 고른 음악 장르 (kpop|pop|ballad|hip-hop|rnb|edm|jazz|rock). " + "'자동 선택'인 경우 None으로 전달하면 GPT가 가사 무드에 맞춰 추천", + ) + genre: Optional[str] = Field( + None, + description="사용자가 수동으로 고른 음악 장르 (kpop|pop|ballad|hip-hop|rnb|edm|jazz|rock). " + "'자동 선택'인 경우 None으로 전달하면 GPT가 가사 무드에 맞춰 추천", + ) class GenerateLyricResponse(BaseModel): @@ -200,6 +210,10 @@ class LyricDetailResponse(BaseModel): project_id: int = Field(..., description="프로젝트 ID") status: str = Field(..., description="처리 상태 (processing, completed, failed)") lyric_result: Optional[str] = Field(None, description="생성된 가사 또는 에러 메시지 (실패 시)") + genre: Optional[str] = Field( + None, + description="확정된 음악 장르. 수동 선택 시 요청값, 자동 선택 시 GPT 추천값 (완료 전에는 None)", + ) created_at: Optional[datetime] = Field(None, description="생성 일시") diff --git a/app/lyric/worker/lyric_task.py b/app/lyric/worker/lyric_task.py index 722e47a..5a5521a 100644 --- a/app/lyric/worker/lyric_task.py +++ b/app/lyric/worker/lyric_task.py @@ -4,30 +4,24 @@ Lyric Background Tasks 가사 생성 관련 백그라운드 태스크를 정의합니다. """ -import traceback - from sqlalchemy import select -from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.exc import SQLAlchemyError from app.database.session import BackgroundSessionLocal -from app.home.models import Image, Project, MarketingIntel from app.lyric.models import Lyric from app.utils.prompts.chatgpt_prompt import ChatgptService, ChatGPTResponseError -from app.utils.subtitles import SubtitleContentsGenerator -from app.utils.creatomate import CreatomateService from app.utils.prompts.prompts import Prompt from app.utils.logger import get_logger # 로거 설정 logger = get_logger("lyric") - async def _update_lyric_status( task_id: str, status: str, result: str | None = None, lyric_id: int | None = None, + genre: str | None = None, ) -> bool: """Lyric 테이블의 상태를 업데이트합니다. @@ -36,6 +30,7 @@ async def _update_lyric_status( status: 변경할 상태 ("processing", "completed", "failed") result: 가사 결과 또는 에러 메시지 lyric_id: 특정 Lyric 레코드 ID (재생성 시 정확한 레코드 식별용) + genre: 확정된 음악 장르 (자동 선택이었던 경우 GPT 추천값으로 갱신) Returns: bool: 업데이트 성공 여부 @@ -61,6 +56,8 @@ async def _update_lyric_status( lyric.status = status if result is not None: lyric.lyric_result = result + if genre is not None: + lyric.genre = genre await session.commit() logger.info(f"[Lyric] Status updated - task_id: {task_id}, lyric_id: {lyric_id}, status: {status}") return True @@ -117,14 +114,18 @@ async def generate_lyric_background( #result = await service.generate(prompt=prompt) result_response = await chatgpt.generate_structured_output(prompt, lyric_input_data) result = result_response.lyric + confirmed_genre = result_response.recommended_genre step2_elapsed = (time.perf_counter() - step2_start) * 1000 - logger.info(f"[generate_lyric_background] Step 2 완료 - 응답 {len(result)}자 ({step2_elapsed:.1f}ms)") + logger.info( + f"[generate_lyric_background] Step 2 완료 - 응답 {len(result)}자, " + f"genre: {confirmed_genre} ({step2_elapsed:.1f}ms)" + ) # ========== Step 3: DB 상태 업데이트 ========== step3_start = time.perf_counter() logger.debug(f"[generate_lyric_background] Step 3: DB 상태 업데이트...") - await _update_lyric_status(task_id, "completed", result, lyric_id) + await _update_lyric_status(task_id, "completed", result, lyric_id, genre=confirmed_genre) step3_elapsed = (time.perf_counter() - step3_start) * 1000 logger.debug(f"[generate_lyric_background] Step 3 완료 ({step3_elapsed:.1f}ms)") @@ -155,73 +156,3 @@ async def generate_lyric_background( elapsed = (time.perf_counter() - task_start) * 1000 logger.error(f"[generate_lyric_background] EXCEPTION - task_id: {task_id}, error: {e} ({elapsed:.1f}ms)", exc_info=True) await _update_lyric_status(task_id, "failed", f"Error: {str(e)}", lyric_id) - -async def generate_subtitle_background( - orientation: str, - task_id: str, - max_retries: int = 3, -) -> None: - logger.info(f"[generate_subtitle_background] START - task_id: {task_id}, orientation: {orientation}") - - for attempt in range(1, max_retries + 1): - try: - creatomate_service = CreatomateService(orientation=orientation) - template = await creatomate_service.get_one_template_data(creatomate_service.template_id) - pitchings = creatomate_service.extract_text_format_from_template(template) - - subtitle_generator = SubtitleContentsGenerator() - - async with BackgroundSessionLocal() as session: - project_result = await session.execute( - select(Project) - .where(Project.task_id == task_id) - .order_by(Project.created_at.desc()) - .limit(1) - ) - project = project_result.scalar_one_or_none() - marketing_result = await session.execute( - select(MarketingIntel).where(MarketingIntel.id == project.marketing_intelligence) - ) - marketing_intelligence = marketing_result.scalar_one_or_none() - - store_address = project.detail_region_info - customer_name = project.store_name - language = project.language or "Korean" - logger.info(f"[generate_subtitle_background] customer_name: {customer_name}, store_address: {store_address}, language: {language}") - - generated_subtitles = await subtitle_generator.generate_subtitle_contents( - marketing_intelligence=marketing_intelligence.intel_result, - pitching_label_list=pitchings, - customer_name=customer_name, - detail_region_info=store_address, - language=language, - industry=project.industry, - ) - pitching_output_list = generated_subtitles.pitching_results - - subtitle_modifications = { - pitching_output.pitching_tag: pitching_output.pitching_data - for pitching_output in pitching_output_list - } - logger.info(f"[generate_subtitle_background] subtitle_modifications: {subtitle_modifications}") - - async with BackgroundSessionLocal() as session: - marketing_result = await session.execute( - select(MarketingIntel).where(MarketingIntel.id == project.marketing_intelligence) - ) - marketing_intelligence = marketing_result.scalar_one_or_none() - marketing_intelligence.subtitle = subtitle_modifications - await session.commit() - - logger.info(f"[generate_subtitle_background] DONE - task_id: {task_id} (attempt {attempt}/{max_retries})") - return - - except Exception as e: - logger.error( - f"[generate_subtitle_background] FAILED (attempt {attempt}/{max_retries}) - task_id: {task_id}, error: {e}", - exc_info=True, - ) - if attempt < max_retries: - logger.info(f"[generate_subtitle_background] 재시도 중... ({attempt + 1}/{max_retries}) - task_id: {task_id}") - - logger.error(f"[generate_subtitle_background] 모든 재시도 실패 - task_id: {task_id}") diff --git a/app/utils/autotag.py b/app/utils/autotag.py index 7b65909..aa526cf 100644 --- a/app/utils/autotag.py +++ b/app/utils/autotag.py @@ -6,10 +6,11 @@ from app.utils.prompts.schemas import SpaceType, Subject, Camera, MotionRecommen import asyncio -async def autotag_image(image_url : str) -> list[str]: #tag_list +async def autotag_image(image_url : str, industry: str = "") -> list[str]: #tag_list chatgpt = ChatgptService(model_type="gemini") image_input_data = { "img_url" : image_url, + "industry" : industry, "space_type" : list(SpaceType), "subject" : list(Subject), "camera" : list(Camera), @@ -19,10 +20,11 @@ async def autotag_image(image_url : str) -> list[str]: #tag_list image_result = await chatgpt.generate_structured_output(image_autotag_prompt, image_input_data, image_url, False) return image_result -async def autotag_images(image_url_list : list[str]) -> list[dict]: #tag_list +async def autotag_images(image_url_list : list[str], industry: str = "") -> list[dict]: #tag_list chatgpt = ChatgptService(model_type="gemini") - image_input_data_list = [{ + image_input_data_list = [{ "img_url" : image_url, + "industry" : industry, "space_type" : list(SpaceType), "subject" : list(Subject), "camera" : list(Camera), diff --git a/app/utils/creatomate.py b/app/utils/creatomate.py index 776e090..26cee72 100644 --- a/app/utils/creatomate.py +++ b/app/utils/creatomate.py @@ -233,7 +233,7 @@ DHST0002 = "3f194cc7-464e-4581-9db2-179d42d3e40f" DHST0003 = "f45df555-2956-4a13-9004-ead047070b3d" DVST0001T = "fe11aeab-ff29-4bc8-9f75-c695c7e243e6" HST_LIST = [DHST0001,DHST0002,DHST0003] -VST_LIST = [DVST0001,DVST0002,DVST0003, DVST0001T] +VST_LIST = [DVST0001T] SCENE_TRACK = 1 AUDIO_TRACK = 2 @@ -242,11 +242,13 @@ KEYWORD_TRACK = 4 def select_template(orientation:OrientationType): if orientation == "horizontal": - return DHST0001 + template_id = DHST0001 elif orientation == "vertical": - return DVST0001T + template_id = random.choice(VST_LIST) else: raise + logger.info(f"[select_template] orientation={orientation}, template_id={template_id}") + return template_id async def get_shared_client() -> httpx.AsyncClient: """공유 HTTP 클라이언트를 반환합니다. 없으면 생성합니다.""" @@ -451,56 +453,161 @@ class CreatomateService: for _, (_, template_type) in enumerate(template_component_data.items()): if template_type == target_template_type: count += 1 - return count - + return count + def template_matching_taged_image( self, - template : dict, - taged_image_list : list, # [{"image_name" : str , "image_tag" : dict}] + template: dict, + taged_image_list: list, # [{"image_url": str, "image_tag": dict}] — 이미 마케팅 적합성 필터를 통과한 이미지만 전달할 것 music_url: str, - address : str, - duplicate : bool = False - ) -> list: + address: str, + duplicate: bool = False + ) -> tuple[dict, dict]: + """템플릿 슬롯에 이미지를 배정합니다. + + Returns: + (modifications, assigned) 튜플 + - modifications: {슬롯명: image_url, "audio-music": music_url, ...} — modify_element에 전달 + - assigned: {슬롯명: image_tag} — 자막 생성 컨텍스트에 전달 (modify_element에 넣지 않음) + + 배정 전략 (duplicate=False): + - 순수 greedy pop 대신 "난이도 우선 greedy" 적용 + - 사전에 전체 슬롯 점수를 계산해 최고 달성 점수가 낮은(=까다로운) 슬롯을 먼저 배정 + - 이미지 배정 직전 점수를 재계산 (pop 이후 인덱스 변동 대응) + + 배정 전략 (duplicate=True, 이미지 < 슬롯): + - 1단계(최소 커버리지 보장): 이미지별로 가장 잘 맞는 슬롯을 찾아 pool의 모든 이미지를 + 서로 다른 슬롯에 먼저 배정한다 (이미지 수 <= 슬롯 수이므로 전량 배정 가능). + 슬롯을 먼저 소진시키는 것이 아니라 "이미지"를 기준으로 난이도 우선 배정하여, + 특정 이미지 몇 장이 여러 슬롯을 독점하는 것을 방지한다. + - 2단계: 1단계에서 못 채운 나머지 슬롯은 기존처럼 pool 전체에서 최고점 이미지를 + 배정한다 (여기서부터는 중복 재사용 불가피). + + Note: + marketing_acceptable 필터링은 크롤링 단계에서 이미 완료되므로 여기서는 재필터링하지 않는다. + taged_image_list를 그대로 pool로 사용한다. + """ source_elements = template["source"]["elements"] template_component_data = self.parse_template_component_name(source_elements) - taged_image_list = [img for img in taged_image_list if img.get("image_tag") is not None] - modifications = {} + pool = taged_image_list + modifications: dict = {} + assigned: dict = {} # {슬롯명: image_tag} — 자막 컨텍스트용 - for slot_idx, (template_component_name, template_type) in enumerate(template_component_data.items()): - match template_type: - case "image": - image_score_list = self.calculate_image_slot_score_multi(taged_image_list, template_component_name) - if duplicate: - # 이미지 부족(슬롯 수 > 이미지 수): 재사용이 불가피하므로 슬롯에 가장 적합한 이미지(최고점) 선택 - maximum_idx = image_score_list.index(max(image_score_list)) - selected = taged_image_list[maximum_idx] - else: - # 이미지 충분(슬롯 수 <= 이미지 수): 상위 3장 중 점수 비례 확률로 선택해 영상 다양화 - # 동일 요청을 반복할 때 매번 같은 영상이 생성되는 문제 방지 - sorted_indices = sorted(range(len(image_score_list)), key=lambda i: image_score_list[i], reverse=True) - top_indices = sorted_indices[:min(3, len(sorted_indices))] - weights = [image_score_list[i] for i in top_indices] - if sum(weights) > 0: - maximum_idx = random.choices(top_indices, weights=weights, k=1)[0] - else: - maximum_idx = random.choice(top_indices) - selected = taged_image_list.pop(maximum_idx) - image_name = selected["image_url"] - modifications[template_component_name] =image_name - pass - case "text": - if "address_input" in template_component_name: - modifications[template_component_name] = address + # 이미지 슬롯과 텍스트 슬롯 분리 + image_slots = [name for name, t in template_component_data.items() if t == "image"] + text_slots = [(name, t) for name, t in template_component_data.items() if t == "text"] + + if not pool: + logger.warning("[template_matching_taged_image] 태그된 이미지가 없음 — 이미지 슬롯 배정 불가") + elif duplicate: + # 이미지 부족(슬롯 수 > 이미지 수) + # Step 1: 최소 커버리지 보장 — 이미지마다 가장 잘 맞는 슬롯을 찾아 서로 다른 슬롯에 배정 + remaining_slots = list(image_slots) + + def _best_slot_for_image(image: dict, slots: list) -> tuple[str | None, float]: + best_slot, best_score = None, -1.0 + for slot in slots: + score = self.calculate_image_slot_score_multi([image], slot)[0] + if score > best_score: + best_slot, best_score = slot, score + return best_slot, best_score + + # 사전 점수 계산 (최고 달성 점수가 낮을수록 배정처가 마땅찮은=까다로운 이미지) + prelim_best_score = [ + _best_slot_for_image(image, remaining_slots)[1] for image in pool + ] + ordered_indices = sorted( + range(len(pool)), + key=lambda i: prelim_best_score[i] + ) + + for idx in ordered_indices: + if not remaining_slots: + break + image = pool[idx] + # 슬롯이 이전 배정으로 줄어들었으므로 현재 remaining_slots 기준 재계산 + best_slot, _ = _best_slot_for_image(image, remaining_slots) + if best_slot is None: + continue + modifications[best_slot] = image["image_url"] + assigned[best_slot] = image["image_tag"] + remaining_slots.remove(best_slot) + + # Step 2: 커버리지 배정 후 남은 슬롯은 pool 전체에서 최고점 이미지 배정 (재사용 불가피) + for slot in remaining_slots: + scores = self.calculate_image_slot_score_multi(pool, slot) + if not scores: + continue + best_idx = scores.index(max(scores)) + selected = pool[best_idx] + modifications[slot] = selected["image_url"] + assigned[slot] = selected["image_tag"] + else: + # 이미지 충분(슬롯 수 <= 이미지 수): 난이도 우선 greedy 배정 + # Step 1: 정렬용 사전 점수 계산 (최고 달성 점수가 낮을수록 배정이 어려운 슬롯) + prelim = {slot: self.calculate_image_slot_score_multi(pool, slot) for slot in image_slots} + ordered_slots = sorted( + image_slots, + key=lambda s: max(prelim[s]) if prelim[s] else 0.0 + # 오름차순 → 달성 최대값이 낮은(어려운) 슬롯이 앞으로 + ) + + # Step 2: 난이도 순서로 배정, 배정마다 pool에서 pop + 다음 슬롯은 점수 재계산 + for slot in ordered_slots: + if not pool: + logger.warning(f"[template_matching_taged_image] 이미지 풀 소진 — 남은 슬롯 배정 불가: {slot}") + break + # pop 후 인덱스 변동이 있으므로 현재 pool로 재계산 (사전 점수 재사용 금지) + scores = self.calculate_image_slot_score_multi(pool, slot) + if not scores: + continue + + # 상위 3장 중 점수 비례 확률로 선택 (영상 다양화) + sorted_indices = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True) + top_indices = sorted_indices[:min(3, len(sorted_indices))] + weights = [scores[i] for i in top_indices] + if sum(weights) > 0: + chosen_pool_idx = random.choices(top_indices, weights=weights, k=1)[0] + else: + chosen_pool_idx = random.choice(top_indices) + + selected = pool.pop(chosen_pool_idx) + modifications[slot] = selected["image_url"] + assigned[slot] = selected["image_tag"] + + # 텍스트 슬롯 처리 + for name, _ in text_slots: + if "address_input" in name: + modifications[name] = address modifications["audio-music"] = music_url - return modifications + return modifications, assigned + # 점수 계산 상수 (모듈 레벨로 빼면 더 좋지만 기존 코드 스타일 따라 클래스 내부에 위치) + _NARR_FLOOR = 0.25 # narrative 최솟값 modulator: base 점수가 완전히 0이 되는 것을 방지 + _NARR_BONUS = 1.5 # narrative 독립 보너스: base=0이어도 narrative 신호가 점수에 남음 + _NARR_MIN = 0.0 # narrative_preference 값 clamp 하한 (무경계 스키마 방어) + _NARR_MAX = 100.0 # narrative_preference 값 clamp 상한. NarrativePreference 스키마(0.0~100.0)와 스케일 일치 + def calculate_image_slot_score_multi(self, taged_image_list : list[dict], slot_name : str): + """이미지 슬롯에 대한 각 이미지의 매칭 점수를 계산합니다. + + 점수 = base * narr_mod + NARR_BONUS * narr + - base: 태그 카테고리 매칭 합산 (최대 5.5) + - narr_mod: [NARR_FLOOR, 1.0] 구간으로 완화된 narrative 계수 + - NARR_BONUS * narr: narrative 독립 보너스 (base=0이어도 신호 유지) + 이 구조로 "base=0, narrative가 높음"과 "완전히 틀린 이미지(0,0)"를 구별할 수 있음. + """ image_tag_list = [taged_image["image_tag"] for taged_image in taged_image_list] slot_tag_dict = self.parse_slot_name_to_tag(slot_name) - image_score_list = [0] * len(image_tag_list) - + if slot_tag_dict is None: + # parse 실패 시 전부 0점 반환 (슬롯 skip) + return [0.0] * len(image_tag_list) + + base_score_list = [0.0] * len(image_tag_list) + # slot_tag_narrative = NarrativePhase.accent # 기본값 (narrative 토큰 없는 슬롯 대비) + for slot_tag_cate, slot_tag_item in slot_tag_dict.items(): if slot_tag_cate == "narrative_preference": slot_tag_narrative = slot_tag_item @@ -508,41 +615,61 @@ class CreatomateService: match slot_tag_cate: case "space_type": - weight = 2 - case "subject" : - weight = 2 + weight = 2.0 + case "subject": + weight = 2.0 case "camera": - weight = 1 - case "motion_recommended" : + weight = 1.0 + case "motion_recommended": weight = 0.5 case _: - raise + raise ValueError(f"[calculate_image_slot_score_multi] unknown slot tag category: {slot_tag_cate}") for idx, image_tag in enumerate(image_tag_list): - if slot_tag_item.value in image_tag[slot_tag_cate]: #collect! - image_score_list[idx] += weight + if slot_tag_item.value in image_tag[slot_tag_cate]: # collect! + base_score_list[idx] += weight + # narrative 점수 적용: 순수 곱셈 대신 "완화 modulator + 독립 보너스" + # - narr_mod: base를 흐릴 수 있지만 NARR_FLOOR 이하로는 안 떨어짐 + # - NARR_BONUS * narr: base=0이어도 narrative가 높으면 양수 점수 확보 + image_score_list = [] for idx, image_tag in enumerate(image_tag_list): - image_narrative_score = image_tag["narrative_preference"][slot_tag_narrative] - image_score_list[idx] = image_score_list[idx] * image_narrative_score + raw_narr = image_tag.get("narrative_preference", {}).get(slot_tag_narrative, 0.0) + clamped_narr = min(self._NARR_MAX, max(self._NARR_MIN, float(raw_narr))) + narr = clamped_narr / self._NARR_MAX # 0.0~1.0로 정규화 + narr_mod = self._NARR_FLOOR + (1.0 - self._NARR_FLOOR) * narr + score = base_score_list[idx] * narr_mod + self._NARR_BONUS * narr + image_score_list.append(score) return image_score_list - def parse_slot_name_to_tag(self, slot_name : str) -> dict[str, StrEnum]: - tag_list = slot_name.split("-") - space_type = SpaceType(tag_list[0]) - subject = Subject(tag_list[1]) - camera = Camera(tag_list[2]) - motion = MotionRecommended(tag_list[3]) - narrative = NarrativePhase(tag_list[4]) - tag_dict = { - "space_type" : space_type, - "subject" : subject, - "camera" : camera, - "motion_recommended" : motion, - "narrative_preference" : narrative, - } - return tag_dict + def parse_slot_name_to_tag(self, slot_name: str) -> dict[str, StrEnum] | None: + """슬롯 이름을 파싱하여 태그 딕셔너리를 반환합니다. + + 슬롯 이름 형식: {space_type}-{subject}-{camera}-{motion}-{narrative} + 파싱 실패 시 None을 반환합니다 (호출자가 해당 슬롯을 skip+log 처리). + """ + try: + tag_list = slot_name.split("-") + if len(tag_list) < 5: + logger.warning(f"[parse_slot_name_to_tag] 슬롯명 토큰 부족 ({len(tag_list)}개): '{slot_name}' — 슬롯 skip") + return None + space_type = SpaceType(tag_list[0]) + subject = Subject(tag_list[1]) + camera = Camera(tag_list[2]) + motion = MotionRecommended(tag_list[3]) + narrative = NarrativePhase(tag_list[4]) + tag_dict = { + "space_type": space_type, + "subject": subject, + "camera": camera, + "motion_recommended": motion, + "narrative_preference": narrative, + } + return tag_dict + except (ValueError, IndexError) as e: + logger.warning(f"[parse_slot_name_to_tag] 슬롯명 파싱 실패: '{slot_name}' — {e} — 슬롯 skip") + return None def elements_connect_resource_blackbox( self, @@ -788,8 +915,7 @@ class CreatomateService: for animation in elem["animations"]: if "reversed" in animation: continue - assert animation.get("time",0) == 0 # 0이 아닌 경우 확인 필요 - if "transition" in animation and animation["transition"]: + if "transition" in animation and animation["transition"] and animation.get("time", 0) == 0: track_maximum_duration[elem["track"]] -= animation["duration"] else: track_maximum_duration[elem["track"]] = max(track_maximum_duration[elem["track"]], elem["time"] + elem["duration"]) @@ -825,7 +951,8 @@ class CreatomateService: if "animations" not in elem: continue for animation in elem["animations"]: - assert animation["time"] == 0 # 0이 아닌 경우 확인 필요 + if animation.get("time", 0) != 0: + animation["time"] = animation["time"] * extend_rate animation["duration"] = animation["duration"] * extend_rate except Exception as e: logger.error( @@ -890,7 +1017,7 @@ class CreatomateService: len_keywords = len(target_keywords) if len(target_keywords) < 3 else 3 - hashtaged_target_keywords = [f"#{tk}" for tk in target_keywords[len_keywords]] + hashtaged_target_keywords = [f"#{tk}" for tk in target_keywords[:len_keywords]] mod_dict = { "thumb-hashtag-primary" : ' '.join(hashtaged_target_keywords), diff --git a/app/utils/image_filter.py b/app/utils/image_filter.py new file mode 100644 index 0000000..4db8e58 --- /dev/null +++ b/app/utils/image_filter.py @@ -0,0 +1,125 @@ +"""크롤링 단계 이미지 마케팅 적합성 필터. + +마케팅 적합성(marketing_acceptable) 판정은 이 모듈이 전담한다. 크롤링 시점에 +방문자/AI View 사진(extra_photo_urls)만을 대상으로 판정하며, 업로드 이후 단계의 +태깅(app/utils/autotag.py의 autotag_images(), sheet 기반 image_autotag_prompt)은 +공간/피사체/카메라/모션/내러티브 태그만 다루고 marketing_acceptable은 포함하지 않는다 +(이미 이 단계에서 걸러진 이미지만 태깅 대상이 되기 때문). 프롬프트가 짧아 +시트(Prompt) 대신 코드에 하드코딩한다. + +업체 등록 사진(owner_images)은 이 필터의 대상이 아니며, 호출측(라우터)에서 +owner_images 개수가 NvMapScraper.SUPPLEMENT_THRESHOLD(30) 이상이면 이 모듈을 +아예 호출하지 않는 것이 원칙이다. +""" + +import asyncio + +from app.utils.logger import get_logger +from app.utils.prompts.chatgpt_prompt import ChatgptService +from app.utils.prompts.schemas import MarketingFilterOutput + +logger = get_logger("image_filter") + +# 크롤링 단계 필터링에 사용할 Gemini 모델. 시트(prompts.py)와 달리 코드에 고정한다. +MARKETING_FILTER_MODEL = "gemini-3.5-flash" + +MAX_RETRY = 2 + +MARKETING_FILTER_PROMPT_TEMPLATE = """\ +당신은 광고 영상 제작을 위한 이미지 심사자입니다. 아래 업종의 광고 영상에 이 이미지를 \ +사용해도 되는지 판정하세요. + +업종(industry): {industry} + +## MARKETING SUITABILITY CHECK (industry-aware) +- marketing_acceptable: true | false +- reject_reason: short reason if false (e.g., 저화질, 인물 신원 노출, 업종 부적합, 민감/의료 장면, 미성년자 식별 노출). Empty if acceptable. +- Person identity protection: 식별 가능한 얼굴이 주피사체이고 동의 여부가 불확실하면 보수적으로 처리. + +주어진 스키마에 맞춰 marketing_acceptable과 reject_reason만 반환하세요.\ +""" + + +def _build_prompt(industry: str) -> str: + return MARKETING_FILTER_PROMPT_TEMPLATE.format(industry=industry or "미상") + + +async def filter_marketing_images(image_url_list: list[str], industry: str = "") -> list[bool]: + """이미지 URL마다 마케팅 적합성(marketing_acceptable)만 판정해 통과 여부 리스트를 반환한다. + + 이미지 1장당 API 호출 1회를 asyncio.gather로 병렬 실행한다(배치 호출 아님). + 실패한 이미지는 최대 MAX_RETRY회 재시도하며, 그래도 실패하면 보수적으로 + False(제외)로 처리한다. + + Args: + image_url_list: 판정할 이미지 URL 목록 (extra_photo_urls의 original만 전달할 것) + industry: 업종 분류값 (_resolve_industry 결과). 필터 프롬프트 앞에서 먼저 계산되어야 함. + + Returns: + image_url_list와 같은 순서의 통과 여부(bool) 리스트. + """ + if not image_url_list: + return [] + + chatgpt = ChatgptService(model_type="gemini") + prompt_text = _build_prompt(industry) + + async def _call(url: str): + return await chatgpt._call_pydantic_output_chat_completion( + prompt=prompt_text, + output_format=MarketingFilterOutput, + model=MARKETING_FILTER_MODEL, + img_url=url, + image_detail_high=False, + ) + + results: list[MarketingFilterOutput | BaseException] = await asyncio.gather( + *[_call(url) for url in image_url_list], return_exceptions=True + ) + + for _ in range(MAX_RETRY): + failed_idx = [i for i, r in enumerate(results) if isinstance(r, Exception)] + if not failed_idx: + break + logger.warning(f"[image_filter] 재시도 대상 인덱스: {failed_idx}") + retried = await asyncio.gather( + *[_call(image_url_list[i]) for i in failed_idx], return_exceptions=True + ) + for i, result in zip(failed_idx, retried): + results[i] = result + + final_failed = [i for i, r in enumerate(results) if isinstance(r, Exception)] + if final_failed: + logger.warning( + f"[image_filter] {len(final_failed)}건 최종 실패 → 보수적으로 제외 처리: {final_failed}" + ) + + return [ + (not isinstance(r, Exception)) and r.marketing_acceptable + for r in results + ] + + +def assemble_images( + owner_images: list[dict], + extra_images: list[dict], + extra_pass_flags: list[bool], + max_images: int, +) -> list[dict]: + """owner_images 전량 + 필터 통과한 extra_images를 합쳐 최대 max_images장을 만든다. + + owner_images는 무조건 포함(필터링 대상 아님). extra_images는 extra_pass_flags와 + 같은 순서로 매칭되며, 통과분만 순서대로 이어붙인다. original URL 기준으로 중복을 + 제거한다(nvMapScraper._extract_origins의 dedup 방식과 동일). + """ + passed_extra = [img for img, ok in zip(extra_images, extra_pass_flags) if ok] + + seen: set[str] = set() + combined: list[dict] = [] + for img in owner_images + passed_extra: + original = img.get("original") + if original and original not in seen: + seen.add(original) + combined.append(img) + + return combined[:max_images] diff --git a/app/utils/nvMapPwScraper.py b/app/utils/nvMapPwScraper.py index 0872dbe..e369b96 100644 --- a/app/utils/nvMapPwScraper.py +++ b/app/utils/nvMapPwScraper.py @@ -52,14 +52,12 @@ class NvMapPwScraper(): cls.is_ready = True GRAPHQL_URL = "https://pcmap-api.place.naver.com/graphql" - _PLACE_TYPES = ("restaurant", "accommodation") @classmethod async def fetch_graphql( cls, place_id: str, payloads: list[dict], - place_type: str | None = None, ) -> list[dict | None] | None: """실제 브라우저로 네이버 WTM 안티봇 캡차를 통과해 GraphQL 쿼리를 실행한다. @@ -73,8 +71,6 @@ class NvMapPwScraper(): Args: place_id: 네이버 place ID payloads: GraphQL POST 본문 목록 - place_type: "restaurant" | "accommodation" (없으면 둘 다 시도) - Returns: payload별 파싱 JSON 목록(실패 항목은 None). 토큰 캡처 실패 시 None. """ @@ -95,8 +91,7 @@ class NvMapPwScraper(): page.on("request", on_request) try: - types = [place_type] if place_type in cls._PLACE_TYPES else list(cls._PLACE_TYPES) - for t in types: + for t in ("place", "restaurant", "accommodation"): try: await page.goto( f"https://pcmap.place.naver.com/{t}/{place_id}/home", diff --git a/app/utils/nvMapScraper.py b/app/utils/nvMapScraper.py index ee6985e..99fe1ba 100644 --- a/app/utils/nvMapScraper.py +++ b/app/utils/nvMapScraper.py @@ -35,6 +35,8 @@ class NvMapScraper: GRAPHQL_URL: str = "https://pcmap-api.place.naver.com/graphql" REQUEST_TIMEOUT = 120 # 초 data_source_identifier = "nv" + SUPPLEMENT_THRESHOLD = 30 # 업체 사진이 이 수 미만일 때 방문자 사진으로 보충 + MAX_IMAGES = 50 # 업체+방문자 사진 합산 최대 장수 OVERVIEW_QUERY: str = """ query getAccommodation($id: String!, $deviceType: String) { business: placeDetail(input: {id: $id, isNx: true, deviceType: $deviceType}) { @@ -51,7 +53,15 @@ query getAccommodation($id: String!, $deviceType: String) { visitorReviewsTotal } images { images { origin url } } - cpImages(source: [ugcImage]) { images { origin url } } + } +}""" + + PHOTO_VIEWER_QUERY: str = """ +query getPhotoViewerItems($input: PhotoViewerInput) { + photoViewer(input: $input) { + photos { + originalUrl + } } }""" @@ -84,10 +94,11 @@ query getVisitorReviewStats($id: String!) { ) self.scrap_type: str | None = None self.rawdata: dict | None = None - self.image_link_list: list[str] | None = None + self.image_link_list: list[dict] | None = None # [{"preview": str, "original": str}] + self.owner_images: list[dict] | None = None # 업체 등록 사진 (마케팅 필터 제외 대상) + self.extra_photo_urls: list[dict] | None = None # 방문자/AI View 보충 사진 (마케팅 필터 대상) self.base_info: dict | None = None self.facility_info: str | None = None - self.place_type: str | None = None # 크롤링으로 감지된 업종 (accommodation | restaurant) self.voted_keyword_stats: list[dict] | None = None # 키워드 투표 집계 (displayName, count) def _get_request_headers(self) -> dict: @@ -96,6 +107,62 @@ query getVisitorReviewStats($id: String!) { headers["Cookie"] = self.cookies return headers + _GIF_PATTERN = re.compile(r"\.gif(?:/|$)", re.IGNORECASE) + + @classmethod + def _is_gif_url(cls, url: str) -> bool: + """URL의 확장자가 gif인지 확인한다. + + 네이버 CDN은 리사이즈 파라미터를 확장자 뒤에 경로 세그먼트로 붙인다 + (예: ".../3982000924.gif/500x500") — endswith(".gif")로는 잡히지 않으므로 + 경로 세그먼트 단위로 ".gif" 뒤에 "/" 또는 문자열 끝이 오는지 검사한다. + """ + return bool(cls._GIF_PATTERN.search(url)) + + @staticmethod + def _extract_origins(image_node: dict) -> list[dict]: + """GraphQL 이미지 노드({ images: [{origin, url}, ...] })에서 {preview, original} 목록을 추출한다. + origin = 원본(업로드용), url = CDN 리사이즈(미리보기용) + """ + items = image_node.get("images") or [] + return [ + {"preview": item.get("url") or item["origin"], "original": item["origin"]} + for item in items + if item.get("origin") and not NvMapScraper._is_gif_url(item["origin"]) + ] + + @staticmethod + def _extract_photo_viewer_urls(photo_viewer_raw: dict | None) -> list[dict]: + """getPhotoViewerItems 응답에서 {preview, original} 목록을 추출한다. + photoViewer는 썸네일 필드가 없으므로 preview = original 동일하게 사용. + """ + photos = ((photo_viewer_raw or {}).get("data") or {}).get("photoViewer") or {} + items = photos.get("photos") or [] + return [ + {"preview": p["originalUrl"], "original": p["originalUrl"]} + for p in items + if p.get("originalUrl") and not NvMapScraper._is_gif_url(p["originalUrl"]) + ] + + @staticmethod + def _dedup_by_original(images: list[dict]) -> list[dict]: + """{"preview","original"} dict 리스트를 original URL 기준으로 순서를 유지한 채 중복 제거한다.""" + seen: set[str] = set() + result = [] + for img in images: + original = img.get("original") + if original and original not in seen: + seen.add(original) + result.append(img) + return result + + @staticmethod + def _interleave(a: list, b: list) -> list: + """두 리스트를 1:1 교차 병합한다. 남은 항목은 뒤에 붙인다.""" + result = [x for pair in zip(a, b) for x in pair] + longer = a[len(b):] if len(a) > len(b) else b[len(a):] + return result + longer + async def parse_url(self) -> str: """URL에서 place ID를 추출합니다. 단축 URL인 경우 실제 URL로 변환합니다.""" place_pattern = r"/place/(\d+)" @@ -114,9 +181,6 @@ query getVisitorReviewStats($id: String!) { # place.naver.com/{type}/{id} 형식 (예: m.place.naver.com/restaurant/11667161) type_match = re.search(r"place\.naver\.com/([a-zA-Z]+)/(\d+)", self.url) if type_match: - url_place_type = type_match.group(1) - if url_place_type in ("accommodation", "restaurant"): - self.place_type = url_place_type return type_match.group(2) if not match: raise URLNotFoundException("Failed to parse place ID from URL") @@ -144,7 +208,7 @@ query getVisitorReviewStats($id: String!) { # self.scrap_type = "GraphQL-Browser" # ── 실제 브라우저로 WTM 캡차 우회 ── - data, stats_data = await self._scrap_via_browser(place_id) + data, stats_data, extra_photo_urls = await self._scrap_via_browser(place_id) fac_data = None # 편의시설(HTML)은 브라우저 경로에서 생략 (best-effort) self.scrap_type = "GraphQL-Browser" @@ -152,21 +216,41 @@ query getVisitorReviewStats($id: String!) { # Naver 기준임, 구글 등 다른 데이터 소스의 경우 고유 Identifier 사용할 것. self.place_id = self.data_source_identifier + place_id self.rawdata["facilities"] = fac_data - self.image_link_list = [ - nv_image["origin"] - for nv_image in data["data"]["business"]["images"]["images"] - ] + business = data["data"]["business"] + + # 업체 등록 이미지 (마케팅 적합성 필터 제외 대상 — 자체 dedup) + self.owner_images = self._dedup_by_original(self._extract_origins(business.get("images") or {})) + + # 방문자/AI View 보충 사진 (마케팅 적합성 필터 대상). owner에 이미 있는 원본은 제외. + owner_originals = {img["original"] for img in self.owner_images} + self.extra_photo_urls = self._dedup_by_original( + [img for img in extra_photo_urls if img["original"] not in owner_originals] + ) + + # 업체 사진이 임계값 미만이면 내부/외부/리뷰 사진으로 보충 + # (필터링 없는 기본 조립. 마케팅 적합성 필터를 적용하려면 호출측에서 + # owner_images / extra_photo_urls를 직접 사용해 image_filter.assemble_images로 재조립할 것.) + if len(self.owner_images) < self.SUPPLEMENT_THRESHOLD: + combined = self._dedup_by_original(self.owner_images + self.extra_photo_urls) + logger.info( + f"[NvMapScraper] 업체 사진 {len(self.owner_images)}장 < {self.SUPPLEMENT_THRESHOLD}장 " + f"→ 보충 사진 {len(self.extra_photo_urls)}장 추가 (합산 {len(combined)}장, 상한 {self.MAX_IMAGES}장)" + ) + else: + combined = self.owner_images + + self.image_link_list = combined[: self.MAX_IMAGES] self.base_info = data["data"]["business"]["base"] self.facility_info = fac_data self.voted_keyword_stats = stats_data return - async def _scrap_via_browser(self, place_id: str) -> tuple[dict, list[dict] | None]: + async def _scrap_via_browser(self, place_id: str) -> tuple[dict, list[dict] | None, list[str]]: """직접 호출이 WTM 캡차에 막힌 경우, 실제 브라우저로 GraphQL을 호출한다. Returns: - (overview_data, review_stats_details) + (overview_data, review_stats_details, extra_photo_urls) Raises: GraphQLException: 브라우저 폴백마저 실패한 경우 @@ -183,32 +267,97 @@ query getVisitorReviewStats($id: String!) { "variables": {"id": place_id}, "query": self.REVIEW_STATS_QUERY, } + interior_payload = { + "operationName": "getPhotoViewerItems", + "variables": { + "input": { + "businessId": place_id, + "cursors": [{"id": "aiView"}], + "filter": "AI View", + "subFilter": "INTERIOR", + "dateRange": "", + "excludeAuthorIds": [], + "excludeClipIds": [], + "excludeSection": [], + } + }, + "query": self.PHOTO_VIEWER_QUERY, + } + exterior_payload = { + "operationName": "getPhotoViewerItems", + "variables": { + "input": { + "businessId": place_id, + "cursors": [{"id": "aiView"}], + "filter": "AI View", + "subFilter": "EXTERIOR", + "dateRange": "", + "excludeAuthorIds": [], + "excludeClipIds": [], + "excludeSection": [], + } + }, + "query": self.PHOTO_VIEWER_QUERY, + } + review_payload = { + "operationName": "getPhotoViewerItems", + "variables": { + "input": { + "businessId": place_id, + "cursors": [{"id": "placeReview"}], + "dateRange": "", + "excludeAuthorIds": [], + "excludeClipIds": [], + "excludeSection": [], + } + }, + "query": self.PHOTO_VIEWER_QUERY, + } - try: - results = await NvMapPwScraper.fetch_graphql( - place_id, [overview_payload, stats_payload], self.place_type - ) - except Exception as e: - logger.error(f"[NvMapScraper] 브라우저 폴백 오류: {e}") - raise GraphQLException(f"브라우저 폴백 실패: {e}") + payloads = [overview_payload, stats_payload, interior_payload, exterior_payload, review_payload] + MAX_RETRY = 3 + results = None + last_error: Exception | None = None + for attempt in range(1, MAX_RETRY + 1): + try: + results = await NvMapPwScraper.fetch_graphql(place_id, payloads) + except Exception as e: + last_error = e + logger.warning(f"[NvMapScraper] 브라우저 폴백 시도 {attempt}/{MAX_RETRY} 오류: {e}") + if attempt < MAX_RETRY: + await asyncio.sleep(1) + continue + if results and results[0] is not None and "data" in results[0]: + break + logger.warning(f"[NvMapScraper] 브라우저 폴백 시도 {attempt}/{MAX_RETRY} 실패(405 등), 재시도") + results = None + if attempt < MAX_RETRY: + await asyncio.sleep(1) if not results or results[0] is None or "data" not in results[0]: + if last_error: + raise GraphQLException(f"브라우저 폴백 실패: {last_error}") raise GraphQLException("브라우저 폴백 크롤링 실패 (WTM 토큰 또는 응답 없음)") data = results[0] stats_data = None stats_raw = results[1] if len(results) > 1 else None if stats_raw: - stats_data = ( - stats_raw.get("data", {}) - .get("visitorReviewStats", {}) - .get("analysis", {}) - .get("votedKeyword", {}) - .get("details") - ) or None + _vrs = (stats_raw.get("data") or {}).get("visitorReviewStats") or {} + _analysis = _vrs.get("analysis") or {} + _voted = _analysis.get("votedKeyword") or {} + stats_data = _voted.get("details") or None + + interior_urls = self._extract_photo_viewer_urls(results[2] if len(results) > 2 else None) + exterior_urls = self._extract_photo_viewer_urls(results[3] if len(results) > 3 else None) + review_urls = self._extract_photo_viewer_urls(results[4] if len(results) > 4 else None) + extra_photo_urls = self._interleave(interior_urls, exterior_urls) + review_urls + logger.info( + f"[NvMapScraper] 보충 이미지 - 내부:{len(interior_urls)} 외부:{len(exterior_urls)} 리뷰:{len(review_urls)}" + ) logger.info(f"[NvMapScraper] 브라우저 폴백 SUCCESS - place_id: {place_id}") - return data, stats_data + return data, stats_data, extra_photo_urls async def _call_get_accommodation(self, place_id: str) -> dict: """GraphQL API를 호출하여 숙소 정보를 가져옵니다. @@ -281,14 +430,10 @@ query getVisitorReviewStats($id: String!) { logger.warning(f"[NvMapScraper] review stats failed: {response.status}") return None result = await response.json() - details = ( - result.get("data", {}) - .get("visitorReviewStats", {}) - .get("analysis", {}) - .get("votedKeyword", {}) - .get("details") - ) - return details or None + _vrs = (result.get("data") or {}).get("visitorReviewStats") or {} + _analysis = _vrs.get("analysis") or {} + _voted = _analysis.get("votedKeyword") or {} + return _voted.get("details") or None except Exception as e: logger.warning(f"[NvMapScraper] Failed to get review stats: {e}") return None @@ -302,7 +447,7 @@ query getVisitorReviewStats($id: String!) { Returns: 편의시설 정보 문자열 또는 None """ - place_types = ["accommodation", "restaurant"] + place_types = ["place", "accommodation", "restaurant"] try: async with aiohttp.ClientSession() as session: for place_type in place_types: diff --git a/app/utils/prompts/chatgpt_prompt.py b/app/utils/prompts/chatgpt_prompt.py index 27cb984..6a3f898 100644 --- a/app/utils/prompts/chatgpt_prompt.py +++ b/app/utils/prompts/chatgpt_prompt.py @@ -176,7 +176,7 @@ class ChatgptService: input_data : dict, img_url : Optional[str] = None, img_detail_high : bool = False, - silent : bool = False + silent : bool = True ) -> BaseModel: prompt_text = prompt.build_prompt(input_data, silent) diff --git a/app/utils/prompts/schemas/image.py b/app/utils/prompts/schemas/image.py index 4c38a7d..e08cc97 100644 --- a/app/utils/prompts/schemas/image.py +++ b/app/utils/prompts/schemas/image.py @@ -3,6 +3,7 @@ from typing import List, Optional from enum import StrEnum, auto class SpaceType(StrEnum): + """이미지가 촬영된 공간·장소 유형. LLM이 이미지를 보고 해당하는 공간 태그를 선택하는 데 사용.""" exterior_front = auto() exterior_night = auto() exterior_aerial = auto() @@ -43,6 +44,7 @@ class SpaceType(StrEnum): detail_tableware = auto() class Subject(StrEnum): + """이미지 내 주요 피사체 유형. 화면에 무엇이 담겨 있는지를 분류하는 태그.""" empty_space = auto() exterior_building = auto() architecture_detail = auto() @@ -55,6 +57,7 @@ class Subject(StrEnum): person = auto() class Camera(StrEnum): + """이미지의 촬영 기법·구도·조명 특성. 영상 편집 시 컷의 시각적 스타일을 구분하는 태그.""" wide_angle = auto() tight_crop = auto() panoramic = auto() @@ -68,6 +71,7 @@ class Camera(StrEnum): has_face = auto() class MotionRecommended(StrEnum): + """해당 이미지에 적용 가능한 카메라 모션 유형. 영상 제작 시 각 컷에 어울리는 움직임을 추천하는 데 사용.""" static = auto() slow_pan = auto() slow_zoom_in = auto() @@ -76,23 +80,41 @@ class MotionRecommended(StrEnum): dolly = auto() class NarrativePhase(StrEnum): + """광고 영상 내러티브 단계. 이미지가 스토리 흐름에서 어느 위치에 배치되어야 하는지를 나타냄.""" intro = auto() welcome = auto() core = auto() highlight = auto() support = auto() accent = auto() + cta = auto() + +NARRATIVE_PHASE_LABEL: dict[NarrativePhase, str] = { + NarrativePhase.intro: "첫인상(intro)", + NarrativePhase.welcome: "진입(welcome)", + NarrativePhase.core: "핵심(core)", + NarrativePhase.highlight: "차별화(highlight)", + NarrativePhase.support: "보조(support)", + NarrativePhase.accent: "감성(accent)", + NarrativePhase.cta: "행동유도(cta)", +} class NarrativePreference(BaseModel): + """이미지가 각 내러티브 단계에 얼마나 적합한지를 나타내는 점수 모델 (0.0 ~ 100.0). + LLM이 이미지를 분석한 뒤 각 단계별 적합도를 수치로 반환하며, + 영상 편집기가 이 점수를 기반으로 컷 배치 순서를 자동 결정하는 데 사용.""" intro: float = Field(..., description="첫인상 — 여기가 어디인가 | 장소의 정체성과 위치를 전달하는 이미지. 영상 첫 1~2초에 어떤 곳인지 즉시 인지시키는 역할. 건물 외관, 간판, 정원 등 **장소 자체를 보여주는** 컷") welcome: float = Field(..., description="진입/환영 — 어떻게 들어가나 | 도착 후 내부로 들어가는 경험을 전달하는 이미지. 공간의 첫 분위기와 동선을 보여줘 들어가고 싶다는 기대감을 만드는 역할. **문을 열고 들어갔을 때 보이는** 컷.") core: float = Field(..., description="핵심 가치 — 무엇을 경험하나 | **고객이 이 장소를 찾는 본질적 이유.** 이 이미지가 없으면 영상 자체가 성립하지 않음. 질문: 이 비즈니스에서 돈을 지불하는 대상이 뭔가? → 그 답이 core.") highlight: float = Field(..., description="차별화 — 뭐가 특별한가 | **같은 카테고리의 경쟁사 대비 이곳을 선택하게 만드는 이유.** core가 왜 왔는가라면, highlight는 왜 **여기**인가에 대한 답.") support: float = Field(..., description="보조/부대 — 그 외에 뭐가 있나 | 핵심은 아니지만 전체 경험을 풍성하게 하는 부가 요소. 없어도 영상은 성립하지만, 있으면 설득력이 올라감. **이것도 있어요** 라고 말하는 컷.") accent: float = Field(..., description="감성/마무리 — 어떤 느낌인가 | 공간의 분위기와 톤을 전달하는 감성 디테일 컷. 직접적 정보 전달보다 **느낌과 무드**를 제공. 영상 사이사이에 삽입되어 완성도를 높이는 역할.") + cta: float = Field(..., description="행동 유도 — 지금 예약/방문하고 싶게 만드는 컷 | 가격·혜택·한정 오퍼 등 직접적 행동을 촉구하거나, 마지막 장면에서 강한 인상을 남기는 이미지.") # Input 정의 class ImageTagPromptInput(BaseModel): + """이미지 태깅 프롬프트에 전달되는 입력 스키마. + 분석할 이미지 URL과 업종 정보, LLM이 선택 가능한 태그 후보 리스트를 포함.""" img_url : str = Field(..., description="이미지 URL") industry : str = Field(default="", description = "업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction). 통합 프롬프트가 이 값으로 내부 분기") space_type: list[str] = Field(list(SpaceType), description="공간적 정보를 가지는 태그 리스트") @@ -100,9 +122,18 @@ class ImageTagPromptInput(BaseModel): camera: list[str] = Field(list(Camera), description="카메라 정보를 가지는 태그 리스트") motion_recommended: list[str] = Field(list(MotionRecommended), description="가능한 카메라 모션 리스트") +class MarketingFilterOutput(BaseModel): + """크롤링 단계에서 이미지의 마케팅(광고 영상) 사용 가능 여부만 판정하는 프롬프트의 출력 스키마. + ImageTagPromptOutput과 달리 태깅(공간/피사체/카메라/모션/내러티브)은 포함하지 않는다.""" + marketing_acceptable: bool = Field(..., description="광고 영상 사용 가능 여부") + reject_reason: str = Field(default="", description="거부 사유 (marketing_acceptable=false 시). 예: 저화질, 인물 신원 노출, 업종 부적합, 민감/의료 장면, 미성년자 식별 노출") + + # Output 정의 class ImageTagPromptOutput(BaseModel): - #ad_avaliable : bool = Field(..., description="광고 영상 사용 가능 이미지 여부") + """이미지 태깅 프롬프트의 LLM 출력 스키마. + 공간·피사체·카메라·모션 태그와 내러티브 적합도 점수를 담아 반환. + 마케팅 사용 가능 여부는 크롤링 단계(MarketingFilterOutput)에서 이미 판정되므로 포함하지 않는다.""" space_type: list[SpaceType] = Field(..., description="공간적 정보를 가지는 태그 리스트") subject: list[Subject] = Field(..., description="피사체 정보를 가지는 태그 리스트") camera: list[Camera] = Field(..., description="카메라 정보를 가지는 태그 리스트") diff --git a/app/utils/prompts/schemas/lyric.py b/app/utils/prompts/schemas/lyric.py index c2cef87..2cd9aa4 100644 --- a/app/utils/prompts/schemas/lyric.py +++ b/app/utils/prompts/schemas/lyric.py @@ -1,5 +1,7 @@ from pydantic import BaseModel, Field -from typing import List, Optional +from typing import List, Literal, Optional + +GENRE_VALUES = Literal["kpop", "pop", "ballad", "hip-hop", "rnb", "edm", "jazz", "rock"] # Input 정의 class LyricPromptInput(BaseModel): @@ -8,10 +10,16 @@ class LyricPromptInput(BaseModel): detail_region_info : str = Field(..., description = "마케팅 대상 지역 상세") marketing_intelligence_summary : Optional[str] = Field(None, description = "마케팅 분석 정보 보고서") language : str= Field(..., description = "가사 언어") - timing_rules : str = Field(..., description = "시간 제어문") + genre : str = Field(default="", description = "지정된 음악 장르 (kpop|pop|ballad|hip-hop|rnb|edm|jazz|rock). 비어있으면 GPT가 무드에 맞는 장르를 자유롭게 추천") + exclude_genre : str = Field(default="", description = "재생성 시 직전에 사용된 장르. genre가 비어있는 경우(자동 선택) 이 장르는 추천에서 제외") industry : str = Field(default="", description = "업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction). 통합 프롬프트가 이 값으로 내부 분기") # Output 정의 class LyricPromptOutput(BaseModel): lyric: str = Field(..., description="생성된 가사") - suno_prompt: str = Field(..., description="Suno AI용 프롬프트") \ No newline at end of file + suno_prompt: str = Field(..., description="Suno AI용 프롬프트") + recommended_genre: GENRE_VALUES = Field( + ..., + description="가사에 사용된(또는 가장 잘 맞는) 음악 장르. genre 입력이 주어졌다면 그 값을 그대로 반환하고, " + "비어있었다면 exclude_genre를 제외한 나머지 중 가사 무드에 가장 잘 맞는 장르를 직접 선택해 반환", + ) \ No newline at end of file diff --git a/app/utils/suno.py b/app/utils/suno.py index 45f36ba..9a9f082 100644 --- a/app/utils/suno.py +++ b/app/utils/suno.py @@ -114,7 +114,7 @@ class SunoService: Args: prompt: 가사 (customMode=true일 때 가사로 사용) 1분 이내 길이의 노래에 적합한 가사여야 함 - genre: 음악 장르 (예: "K-Pop", "Pop", "R&B", "Hip-Hop", "Ballad", "EDM") + genre: 음악 장르 (예: "K-Pop", "Pop", "R&B", "Hip-Hop", "Ballad", "EDM", "Rock", "Jazz") None일 경우 style 파라미터를 전송하지 않음 callback_url: 생성 완료 시 알림 받을 URL (None일 경우 config에서 기본값 사용) instrumental: True이면 BGM 전용 — 더미 가사로 60초 길이를 유도하고 보컬 없이 생성 @@ -137,7 +137,7 @@ class SunoService: logger.info(f"[Suno] BGM 더미 가사 장르 {normalized_genre} 선택됨") else: formatted_prompt = ( - f"[Song Duration: Exactly 1 minute - Must be around 60 seconds]\n{prompt}" + f"[Song Duration: Around 1 minute - Must be around 60 seconds]\n{prompt}" ) payload: dict[str, Any] = { diff --git a/app/video/api/routers/v1/video.py b/app/video/api/routers/v1/video.py index 5b71bde..8e2265d 100644 --- a/app/video/api/routers/v1/video.py +++ b/app/video/api/routers/v1/video.py @@ -26,13 +26,12 @@ from app.dependencies.pagination import PaginationParams, get_pagination_params from app.user.dependencies.auth import get_current_user, get_current_user_optional from app.user.models import User from app.utils.pagination import PaginatedResponse -from app.home.models import Image, Project, MarketingIntel, ImageTag +from app.home.models import Image, Project, MarketingIntel from app.home.api.routers.v1.home import _extract_region_from_address from app.utils.address_parser import SIDO_CITIES, SIDO_SEARCH_ALIASES from app.lyric.models import Lyric from app.song.models import Song, SongTimestamp from app.utils.creatomate import CreatomateService -from app.utils.subtitles import SubtitleContentsGenerator from app.comment.models import Comment from app.database.like_cache import ( @@ -59,7 +58,7 @@ from app.video.schemas.video_schema import ( VideoThumbnailItem, ) from app.video.worker.video_task import download_and_upload_video_to_blob -from app.video.services.video import get_image_tags_by_task_id + from config import creatomate_settings @@ -199,16 +198,22 @@ async def generate_video( ) marketing_intelligence: MarketingIntel = marketing_result.scalar_one_or_none() - # subtitle 미완료 시 즉시 반환 — Lyric/Song/Image 쿼리 전에 체크하여 불필요한 조회 방지 + # 자막 + 이미지 배정 미완료 시 즉시 반환 — Lyric/Song/Image 쿼리 전에 체크하여 불필요한 조회 방지 # 클라이언트가 /lyric/subtitle/status/{task_id} 폴링 후 재시도 - if not marketing_intelligence.subtitle: - logger.info(f"[generate_video] subtitle pending - task_id: {task_id}") + if not marketing_intelligence.subtitle or not marketing_intelligence.image_match: + pending_what = [] + if not marketing_intelligence.subtitle: + pending_what.append("자막") + if not marketing_intelligence.image_match: + pending_what.append("이미지 배정") + pending_msg = ", ".join(pending_what) + logger.info(f"[generate_video] 사전 준비 미완료 ({pending_msg}) - task_id: {task_id}") return GenerateVideoResponse( success=False, status="subtitle_pending", task_id=task_id, creatomate_render_id=None, - message="자막 생성이 아직 완료되지 않았습니다. /lyric/subtitle/status/{task_id}로 완료 확인 후 재요청하세요.", + message=f"{pending_msg} 생성이 아직 완료되지 않았습니다. /lyric/subtitle/status/{{task_id}}로 완료 확인 후 재요청하세요.", error_message=None, ) @@ -365,25 +370,15 @@ async def generate_video( logger.debug(f"[generate_video] Template fetched - task_id: {task_id}") # 6-2. elements에서 리소스 매핑 생성 - # modifications = creatomate_service.elements_connect_resource_blackbox( - # elements=template["source"]["elements"], - # image_url_list=image_urls, - # music_url=music_url, - # address=store_address - taged_image_list = await get_image_tags_by_task_id(task_id) - min_image_num = creatomate_service.counting_component( - template = template, - target_template_type = "image" - ) - duplicate = bool(len(taged_image_list) < min_image_num) - logger.info(f"[generate_video] Duplicate : {duplicate} | length of taged_image {len(taged_image_list)}, min_len {min_image_num},- task_id: {task_id}") - modifications = creatomate_service.template_matching_taged_image( - template = template, - taged_image_list = taged_image_list, - music_url = music_url, - address = store_address, - duplicate = duplicate, - ) + # 이미지 배정은 /lyric/generate 사전 단계에서 이미 수행되어 marketing_intelligence.image_match에 저장됨. + # 여기서는 사전 산출물을 읽어 음악 URL과 주소만 보완한다. + modifications: dict = dict(marketing_intelligence.image_match) # 슬롯→image_url 사전 배정 결과 + modifications["audio-music"] = music_url + # address_input 슬롯은 사전 단계에서도 채워지지만 영상 시점에 재확인 + for key in list(modifications.keys()): + if "address_input" in key: + modifications[key] = store_address + logger.info(f"[generate_video] image_match loaded from DB (slots: {len(modifications)}) - task_id: {task_id}") logger.debug(f"[generate_video] Modifications created - task_id: {task_id}") subtitle_modifications = marketing_intelligence.subtitle @@ -391,14 +386,14 @@ async def generate_video( modifications.update(subtitle_modifications) # revert thumbnail scene - # thumbnail_modifications = creatomate_service.make_thumbnail_modification( - # brand_name =brand_name, - # region = region, - # brand_concept = brand_concept, - # category_definition= category_definition, - # target_keywords=target_keywords) + thumbnail_modifications = creatomate_service.make_thumbnail_modification( + brand_name =brand_name, + region = region, + brand_concept = brand_concept, + category_definition= category_definition, + target_keywords=target_keywords) - # modifications.update(thumbnail_modifications) + modifications.update(thumbnail_modifications) # 6-3. elements 수정 new_elements = creatomate_service.modify_element( diff --git a/app/video/worker/creative_assets_task.py b/app/video/worker/creative_assets_task.py new file mode 100644 index 0000000..9eedfd4 --- /dev/null +++ b/app/video/worker/creative_assets_task.py @@ -0,0 +1,218 @@ +""" +Creative Assets Background Tasks + +영상 생성 전 사전 준비(이미지 배정 + 자막 생성)를 수행하는 백그라운드 태스크를 정의합니다. +""" + +from sqlalchemy import select + +from app.database.session import BackgroundSessionLocal +from app.home.models import Project, MarketingIntel +from app.utils.subtitles import SubtitleContentsGenerator +from app.utils.creatomate import CreatomateService +from app.utils.logger import get_logger +from app.video.services.video import get_image_tags_by_task_id +from app.utils.prompts.schemas.image import NARRATIVE_PHASE_LABEL + +# 로거 설정 +logger = get_logger("video") + + +async def generate_subtitle_background( + orientation: str, + task_id: str, + max_retries: int = 3, +) -> None: + """자막 + 이미지 배정을 사전에 수행하는 백그라운드 태스크. + + 내부적으로 `generate_creative_assets_background`를 호출한다. + """ + await generate_creative_assets_background( + orientation=orientation, + task_id=task_id, + max_retries=max_retries, + ) + + +async def generate_creative_assets_background( + orientation: str, + task_id: str, + max_retries: int = 3, +) -> None: + """이미지 배정 + 자막 생성을 사전에 수행하는 백그라운드 태스크. + + 실행 순서: + 1. 템플릿 조회 (읽기 전용, 렌더 없음) + 2. 이미지 매칭 — get_image_tags_by_task_id → template_matching_taged_image + → MarketingIntel.image_match 저장 + 3. 자막 생성 — 배정된 이미지 컨텍스트(슬롯별 공간/단계 정보)를 pitching 리스트에 포함 + → MarketingIntel.subtitle 저장 + """ + logger.info(f"[generate_creative_assets_background] START - task_id: {task_id}, orientation: {orientation}") + + for attempt in range(1, max_retries + 1): + try: + creatomate_service = CreatomateService(orientation=orientation) + template = await creatomate_service.get_one_template_data(creatomate_service.template_id) + + # ── 프로젝트 / 마케팅 인텔 로드 ────────────────────────────────── + async with BackgroundSessionLocal() as session: + project_result = await session.execute( + select(Project) + .where(Project.task_id == task_id) + .order_by(Project.created_at.desc()) + .limit(1) + ) + project = project_result.scalar_one_or_none() + marketing_result = await session.execute( + select(MarketingIntel).where(MarketingIntel.id == project.marketing_intelligence) + ) + marketing_intelligence = marketing_result.scalar_one_or_none() + + store_address = project.detail_region_info + customer_name = project.store_name + language = project.language or "Korean" + logger.info( + f"[generate_creative_assets_background] customer: {customer_name}, " + f"address: {store_address}, language: {language} - task_id: {task_id}" + ) + + # ── Step 1: 이미지 매칭 ───────────────────────────────────────── + # marketing_acceptable 필터링은 크롤링 단계에서 이미 완료되었으므로 재필터링하지 않는다. + taged_image_list = await get_image_tags_by_task_id(task_id) + min_image_num = creatomate_service.counting_component( + template=template, + target_template_type="image", + ) + duplicate = len(taged_image_list) < min_image_num + logger.info( + f"[generate_creative_assets_background] image matching — " + f"pool={len(taged_image_list)}, slots={min_image_num}, " + f"duplicate={duplicate} - task_id: {task_id}" + ) + + image_modifications, assigned_image_tags = creatomate_service.template_matching_taged_image( + template=template, + taged_image_list=taged_image_list, + music_url="", # 음악 URL은 영상 생성 시점에 결정 — 사전 단계에서는 빈 값 + address=store_address, + duplicate=duplicate, + ) + # audio-music은 영상 생성 시점에 덮어씌워지므로 image_match에서 제거 + image_match = {k: v for k, v in image_modifications.items() if k != "audio-music"} + + logger.info(f"[generate_creative_assets_background] image_match: {list(image_match.keys())} - task_id: {task_id}") + + # image_match 저장 + async with BackgroundSessionLocal() as session: + marketing_result = await session.execute( + select(MarketingIntel).where(MarketingIntel.id == project.marketing_intelligence) + ) + mi = marketing_result.scalar_one_or_none() + mi.image_match = image_match + await session.commit() + + logger.info(f"[generate_creative_assets_background] image_match saved - task_id: {task_id}") + + # ── Step 2: 자막 생성 (이미지 컨텍스트 포함) ────────────────────── + # 슬롯별 배정된 이미지의 공간/내러티브 정보를 pitching 레이블에 포함하여 + # LLM이 "이 자막이 나올 때 화면에 보이는 이미지"를 인지하고 자막을 작성할 수 있게 함. + pitchings_raw = creatomate_service.extract_text_format_from_template(template) + pitchings_with_context = _build_pitching_list_with_image_context( + pitching_label_list=pitchings_raw, + assigned_image_tags=assigned_image_tags, + ) + + subtitle_generator = SubtitleContentsGenerator() + generated_subtitles = await subtitle_generator.generate_subtitle_contents( + marketing_intelligence=marketing_intelligence.intel_result, + pitching_label_list=pitchings_with_context, + customer_name=customer_name, + detail_region_info=store_address, + language=language, + industry=project.industry, + ) + pitching_output_list = generated_subtitles.pitching_results + + subtitle_modifications = { + pitching_output.pitching_tag: pitching_output.pitching_data + for pitching_output in pitching_output_list + } + logger.info(f"[generate_creative_assets_background] subtitle_modifications: {subtitle_modifications}") + + # subtitle 저장 + async with BackgroundSessionLocal() as session: + marketing_result = await session.execute( + select(MarketingIntel).where(MarketingIntel.id == project.marketing_intelligence) + ) + mi = marketing_result.scalar_one_or_none() + mi.subtitle = subtitle_modifications + await session.commit() + + logger.info( + f"[generate_creative_assets_background] DONE - task_id: {task_id} " + f"(attempt {attempt}/{max_retries})" + ) + return + + except Exception as e: + logger.error( + f"[generate_creative_assets_background] FAILED (attempt {attempt}/{max_retries}) " + f"- task_id: {task_id}, error: {e}", + exc_info=True, + ) + if attempt < max_retries: + logger.info( + f"[generate_creative_assets_background] 재시도 중... " + f"({attempt + 1}/{max_retries}) - task_id: {task_id}" + ) + + logger.error(f"[generate_creative_assets_background] 모든 재시도 실패 - task_id: {task_id}") + + +def _build_pitching_list_with_image_context( + pitching_label_list: list[str], + assigned_image_tags: dict[str, dict], +) -> list[str]: + """pitching 레이블 리스트에 배정 이미지 컨텍스트를 포함하여 반환합니다. + + 슬롯명(`bedroom-furniture-tight_crop-slow_zoom_in-welcome`)의 5번째 토큰과 + pitching 레이어명(`subtitle-welcome-emotion_cue-empathic-001`)의 2번째 토큰이 + 동일한 narrative_phase이므로 시간 겹침 계산 없이 phase로 직접 매칭합니다. + + 출력 예시: + "subtitle-welcome-emotion_cue-empathic-001 | 화면: lobby / furniture / 진입(welcome) 단계" + """ + # narrative_phase → 해당 phase에 배정된 image_tag 매핑 구성 + # 슬롯명 형식: {space_type}-{subject}-{camera}-{motion}-{narrative_phase} + phase_to_tag: dict[str, dict] = {} + for slot_name, tag in assigned_image_tags.items(): + tokens = slot_name.split("-") + if len(tokens) >= 5: + phase = tokens[4] + # 같은 phase에 슬롯이 여러 개면 첫 번째만 사용 + if phase not in phase_to_tag: + phase_to_tag[phase] = tag + + if not phase_to_tag: + return pitching_label_list + + result = [] + for label in pitching_label_list: + # pitching 레이어명 형식: {track_role}-{narrative_phase}-{content_type}-{tone}-{pair_id} + tokens = label.split("-") + context_str = "" + if len(tokens) >= 2: + phase = tokens[1] + tag = phase_to_tag.get(phase) + if tag: + space_types = tag.get("space_type", []) + subjects = tag.get("subject", []) + space_str = "/".join(space_types[:2]) if space_types else "?" + subject_str = subjects[0] if subjects else "?" + phase_str = NARRATIVE_PHASE_LABEL.get(phase, phase) + context_str = f" | 화면: {space_str} / {subject_str} / {phase_str} 단계" + + result.append(label + context_str) + + return result diff --git a/docs/database-schema/migration_2026_06_17_biz_type.sql b/docs/database-schema/migration_2026_06_17_biz_type.sql index af1a4ad..d295fc3 100644 --- a/docs/database-schema/migration_2026_06_17_biz_type.sql +++ b/docs/database-schema/migration_2026_06_17_biz_type.sql @@ -8,4 +8,5 @@ ALTER TABLE project ADD COLUMN biz_type VARCHAR(50) NULL DEFAULT 'place' COMMENT '업종 분류 (place: 기본, accommodation: 숙박, restaurant: 음식점)' AFTER id; -ALTER TABLE project CHANGE biz_type industry VARCHAR(50) DEFAULT ''; +ALTER TABLE project CHANGE biz_type industry VARCHAR(50) DEFAULT '' + COMMENT '업종 분류' AFTER id; diff --git a/docs/database-schema/migration_2026_07_02_add_lyric_genre.sql b/docs/database-schema/migration_2026_07_02_add_lyric_genre.sql new file mode 100644 index 0000000..44706d8 --- /dev/null +++ b/docs/database-schema/migration_2026_07_02_add_lyric_genre.sql @@ -0,0 +1,12 @@ +-- ============================================================ +-- Migration: lyric 테이블에 genre 컬럼 추가 +-- Date: 2026-07-02 +-- Description: 가사 생성 시점에 확정된 음악 장르 저장 +-- - 수동 선택: 사용자가 고른 장르 +-- - 자동 선택: GPT가 가사 무드에 맞춰 추천한 장르 +-- 장르-BPM 불일치로 인한 곡 길이 편차 문제 해결을 위해 도입 +-- ============================================================ + +ALTER TABLE lyric + ADD COLUMN genre VARCHAR(20) NULL + COMMENT '음악 장르 (kpop, pop, ballad, hip-hop, rnb, edm, jazz, rock). 수동 선택 시 요청값, 자동 선택 시 GPT 추천값' AFTER language;