"""Garment analysis pipeline: image save → LLM analysis → DB persist. Supports two modes via auto_save flag: - auto_save=True (default): analyze + save to DB in one step (analyze_and_save) - auto_save=False: analyze only (analyze_only), then client can selectively save (save_garments) Ghost mannequin generation is a separate step via generate_ghost() (requires DB record) or generate_ghost_preview() (stateless, before save). """ import asyncio import logging import uuid from pathlib import Path from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.config import GLOBAL_USER_ID, settings from app.models.garment import Garment from app.services.llm.base import ( BaseLLMProvider, GarmentAnalysis, StyleAnalysis, StyleAnalysisTemp, ) from app.services.segmentation_service import SegmentationService from app.utils.image import ( resize_image_if_needed, save_ghost_mannequin_image, save_original_image, ) logger = logging.getLogger(__name__) class GarmentService: def __init__( self, llm_provider: BaseLLMProvider, segmentation_service: SegmentationService, ): self.llm = llm_provider self.segmentation = segmentation_service async def _strip_ghost_background(self, ghost_bytes: bytes) -> bytes: """Strip the plain white backdrop from an LLM ghost mannequin output. The ghost prompt asks the model for a solid white background because no image model can be relied on to emit a true alpha channel (gpt-image-2 rejects transparent backgrounds outright). rembg runs here to produce the transparent PNG that VTON and the wardrobe UI expect. rembg is synchronous, so it is offloaded to a worker thread; on failure the raw LLM bytes are returned so the caller still gets a usable image. """ try: return await asyncio.to_thread( self.segmentation.remove_background, ghost_bytes ) except Exception: logger.warning( "Ghost mannequin background removal failed; using raw LLM output", exc_info=True, ) return ghost_bytes async def analyze_only( self, image_bytes: bytes, user_id: uuid.UUID | None, ) -> tuple[list[GarmentAnalysis], str]: """Analyze image without saving to DB. 1. Resolve user_id (None → global) 2. Resize & save original image to disk 3. LLM garment analysis Returns (analysis_items, source_image_path). """ resolved_user_id = user_id or GLOBAL_USER_ID resized = resize_image_if_needed(image_bytes) source_path = save_original_image(image_bytes, resolved_user_id) items = await self.llm.analyze_garment(resized) if not items: logger.warning("LLM returned no garment items") return [], source_path logger.info("LLM detected %d garment(s)", len(items)) return items, source_path async def save_garments( self, db: AsyncSession, items: list[GarmentAnalysis], source_image_path: str, user_id: uuid.UUID | None, ghost_image_paths: list[str | None] | None = None, ) -> list[Garment]: """Save analyzed garment items to DB. Args: items: LLM analysis results to persist. source_image_path: Path to the already-saved source image. user_id: Owner (None → global user). ghost_image_paths: Optional per-item ghost image paths. Returns list of saved Garment ORM objects. """ resolved_user_id = user_id or GLOBAL_USER_ID garments: list[Garment] = [] for i, item in enumerate(items): ghost_path = None if ghost_image_paths and i < len(ghost_image_paths): ghost_path = ghost_image_paths[i] garment = Garment( user_id=resolved_user_id, category_main=item.category_main, category_sub=item.category_sub, description=item.description, tags=item.tags, source_image_path=source_image_path, ghost_image_path=ghost_path, ) db.add(garment) garments.append(garment) await db.commit() for g in garments: await db.refresh(g) return garments async def analyze_and_save( self, db: AsyncSession, image_bytes: bytes, user_id: uuid.UUID | None, ) -> list[Garment]: """Analysis pipeline (fast path — no ghost mannequin). 1. Resolve user_id (None → global) 2. Resize & save original image 3. LLM garment analysis 4. Persist garments to DB Returns list of saved Garment ORM objects. """ items, source_path = await self.analyze_only(image_bytes, user_id) if not items: return [] return await self.save_garments(db, items, source_path, user_id) async def generate_ghost_preview( self, source_image_path: str, item: GarmentAnalysis, user_id: uuid.UUID | None, image_model: str | None = None, ) -> str: """Generate ghost mannequin without a DB record (stateless preview). Reads source image from disk, generates ghost via LLM, saves to disk. Returns the ghost image path (relative). ``image_model`` selects the image-generation model (``None`` → provider default). """ resolved_user_id = user_id or GLOBAL_USER_ID source_full = Path(settings.UPLOAD_DIR) / source_image_path if not source_full.exists(): raise ValueError(f"Source image not found: {source_image_path}") image_bytes = source_full.read_bytes() resized = resize_image_if_needed(image_bytes, max_size=1024) try: ghost_bytes = await self.llm.generate_ghost_mannequin( resized, item, "edit", image_model ) except Exception as e: raise RuntimeError(f"LLM 고스트 마네킹 생성 실패: {e}") from e if ghost_bytes is None: raise RuntimeError("LLM이 고스트 마네킹 이미지를 반환하지 않았습니다") ghost_bytes = await self._strip_ghost_background(ghost_bytes) ghost_path = save_ghost_mannequin_image(ghost_bytes, resolved_user_id) return ghost_path async def generate_ghost( self, db: AsyncSession, garment_id: uuid.UUID, image_model: str | None = None, ) -> Garment: """Generate ghost mannequin for an existing garment. Reads the source image, calls LLM image generation, saves result, and updates the garment record. ``image_model`` selects the image-generation model (``None`` → provider default). """ result = await db.execute(select(Garment).where(Garment.id == garment_id)) garment = result.scalar_one_or_none() if garment is None: raise ValueError(f"Garment {garment_id} not found") if garment.ghost_image_path is not None: return garment # already generated # Read source image from disk source_full = Path(settings.UPLOAD_DIR) / garment.source_image_path if not source_full.exists(): raise ValueError(f"Source image not found: {garment.source_image_path}") image_bytes = source_full.read_bytes() resized = resize_image_if_needed(image_bytes, max_size=1024) # Reconstruct GarmentAnalysis for the prompt item = GarmentAnalysis( category_main=garment.category_main, category_sub=garment.category_sub, description=garment.description, tags=garment.tags, ) try: ghost_bytes = await self.llm.generate_ghost_mannequin( resized, item, "edit", image_model ) except Exception as e: raise RuntimeError(f"LLM 고스트 마네킹 생성 실패: {e}") from e if ghost_bytes is None: raise RuntimeError("LLM이 고스트 마네킹 이미지를 반환하지 않았습니다") ghost_bytes = await self._strip_ghost_background(ghost_bytes) ghost_path = save_ghost_mannequin_image(ghost_bytes, garment.user_id) garment.ghost_image_path = ghost_path await db.commit() await db.refresh(garment) return garment async def analyze_style(self, image_bytes: bytes) -> StyleAnalysis: """Analyze overall outfit style from image. 1. Resize image for LLM input 2. Call LLM style analysis Returns StyleAnalysis with scores and comments. """ resized = resize_image_if_needed(image_bytes) return await self.llm.analyze_style(resized) async def analyze_style_temp(self, image_bytes: bytes) -> StyleAnalysisTemp: """Context-aware outfit style analysis (time/occasion/weather).""" resized = resize_image_if_needed(image_bytes) return await self.llm.analyze_style_temp(resized)