image retagging

app/home/api/routers/v1/home.py

@@ -751,7 +751,7 @@ async def upload_images_blob(
     image_urls = [img.img_url for img in result_images]
 
     logger.info(f"[image_tagging] START - task_id: {task_id}")
-    await tag_images_if_not_exist(image_urls)
+    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
@@ -771,8 +771,9 @@ async def upload_images_blob(
     )
 
 
-async def tag_images_if_not_exist(
-        image_urls : list[str]
+async def tagging_images(
+        image_urls : list[str],
+        clear_old_tags : bool = False
     ) -> None:
     # 1. 조회
     async with AsyncSessionLocal() as session:
@@ -784,21 +785,24 @@ async def tag_images_if_not_exist(
         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_tags = [
+        new_imt = [
             ImageTag(img_url=url, img_tag=None)
             for url in image_urls
             if url not in existing_urls
         ]
-        session.add_all(new_tags)
-
-        null_tags = [tag for tag in image_tags if tag.img_tag is None] + new_tags
-        
-        if null_tags:
-            tag_datas = await autotag_images([img.img_url for img in null_tags])
-
-            print(tag_datas)
-
-            for tag, tag_data in zip(null_tags, tag_datas):
-                tag.img_tag = tag_data.model_dump(mode="json")
-
+        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):
+                tag.img_tag = tag_data.model_dump(mode="json")
+                session.add(tag)
+            await session.commit()

app/utils/common.py

@@ -19,12 +19,16 @@ Note:
 
 import os
 import time
+import re
 from typing import Any, Optional, Type
 
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 
+def normalize_location(name: str) -> str:
+    return re.sub(r'(특별시|광역시|특별자치시|특별자치도|시|군|구|도)$', '', name)
+
 def _generate_uuid7_string() -> str:
     """UUID7 문자열을 생성합니다.
 

app/utils/creatomate.py

@@ -38,6 +38,8 @@ import httpx
 
 from app.utils.logger import get_logger
 from app.utils.prompts.schemas.image import SpaceType,Subject,Camera,MotionRecommended,NarrativePhase
+from app.utils.common import normalize_location
+
 from config import apikey_settings, creatomate_settings, recovery_settings
 
 # 로거 설정
@@ -865,4 +867,20 @@ class CreatomateService:
         except Exception as E:
             logger.error("this template does not have same amount of keyword and subtitle.")
         pitching_list = keyword_list + subtitle_list
-        return pitching_list
+        return pitching_list
+
+
+    def make_thumbnail_modification(self, brand_name : str, region : str, brand_concept : str, category_definition : str, target_keywords : list[str]):
+        
+        len_keywords = len(target_keywords) if len(target_keywords) < 3 else 3
+
+        hashtaged_target_keywords = [f"#{tk}" for tk in target_keywords[len_keywords]]
+        
+        mod_dict = {
+            "thumb-hashtag-primary" : ' '.join(hashtaged_target_keywords),
+            "thumb-brand-wordmark" : brand_name,
+            "thumb-subheadline-selling_point" : f"{brand_name} · {normalize_location(region)}",
+            "thumb-headline-hook_claim-aspirational" : brand_concept,
+            "thumb-badge-category" : category_definition,
+        }
+        return mod_dict

app/video/api/routers/v1/video.py

@@ -230,12 +230,21 @@ async def generate_video(
                 )
             project_id = project.id
             store_address = project.detail_region_info
-            # customer_name = project.store_name
+            brand_name = project.store_name
+            region = project.region
 
             marketing_result = await session.execute(
                 select(MarketingIntel).where(MarketingIntel.id == project.marketing_intelligence)
             )
-            marketing_intelligence = marketing_result.scalar_one_or_none()
+            marketing_intelligence : MarketingIntel = marketing_result.scalar_one_or_none()
+
+            category_definition= marketing_intelligence.intel_result["market_positioning"]["category_definition"]
+            target_keywords=marketing_intelligence.intel_result["target_keywords"]
+            
+            brand_concept = ""
+            for sp in marketing_intelligence.intel_result["selling_points"]:
+                if "concept" in sp["english_category"].lower():
+                    brand_concept = sp["description"]
 
             # ===== 결과 처리: Lyric =====
             lyric = lyric_result.scalar_one_or_none()
@@ -368,6 +377,17 @@ async def generate_video(
         subtitle_modifications = marketing_intelligence.subtitle
 
         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)
+        
+        # modifications.update(thumbnail_modifications)
+
         # 6-3. elements 수정
         new_elements = creatomate_service.modify_element(
             template["source"]["elements"],