from pathlib import Path from fastapi import UploadFile, HTTPException, status from sqlmodel import Session from uuid import uuid4 from app.models.product_template import ProductTemplate from app.schemas.product_template import ProductTemplateRead from app.schemas.common import PaginatedResponse from app.repositories.product_template_repo import ProductTemplateRepository from app.utils.image import validate_base_file, save_upload_file_streamed, safe_delete from app.utils.psd_ratio import extract_psd_ratio from app.utils.slugify import normalize_unique_name from app.core.logging import setup_logger from app.core.config import get_settings from app.core.constants import ERR_NOT_FOUND, ERR_FILE_INVALID from app.utils.psd_masks import generate_psd_assets logger = setup_logger(__name__) settings = get_settings() PSD_FILES_DIRECTORY = Path(settings.PSD_FILES_DIRECTORY) PSD_FILES_DIRECTORY.mkdir(parents=True, exist_ok=True) class ProductTemplateService: @staticmethod async def upload_template( name: str, unique_name: str, psd_file: UploadFile, width: int, height: int, user_id: int, session: Session, ) -> ProductTemplate: normalized_unique_name = normalize_unique_name(unique_name) existing = ProductTemplateRepository.get_by_unique_name(session, normalized_unique_name) if existing: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"unique_name '{normalized_unique_name}' already exists." ) validate_base_file(psd_file) # Save PSD try: file_path = await save_upload_file_streamed(psd_file, PSD_FILES_DIRECTORY) except Exception: logger.exception("Failed to save PSD file") raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=ERR_FILE_INVALID) # Ratio ratio = extract_psd_ratio(file_path) # Generate base image + masks assets_dir = PSD_FILES_DIRECTORY / normalized_unique_name assets = generate_psd_assets( psd_path=file_path, output_dir=assets_dir, ) # DB Record tpl = ProductTemplate( name=name, unique_name=normalized_unique_name, uploaded_by=user_id, file_path=str(file_path), ratio=ratio, width=width, height=height, base_image_path=assets["base_image"], base_thumbnail_path=assets["base_thumbnail"], main_black_mask_path=assets["main_black_mask_path"], main_white_mask_path=assets["main_white_mask_path"], logo_black_mask_path=assets["logo_black_mask_path"], logo_white_mask_path=assets["logo_white_mask_path"], color_preview_path=assets["color_preview_path"], color_white_mask_path=assets["color_white_mask_path"], ) tpl = ProductTemplateRepository.create(session, tpl) return tpl @staticmethod async def update_template( template_id: int, name: str | None, new_psd: UploadFile | None, width: int, height: int, session: Session, ) -> ProductTemplate: tpl = ProductTemplateRepository.get_by_id(session, template_id) if not tpl or tpl.is_deleted: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERR_NOT_FOUND) if name: tpl.name = name if new_psd: validate_base_file(new_psd) try: new_path = await save_upload_file_streamed(new_psd, PSD_FILES_DIRECTORY) except Exception: raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=ERR_FILE_INVALID) # Ratio ratio = extract_psd_ratio(new_path) # Generate new base image + masks assets_dir = PSD_FILES_DIRECTORY / tpl.unique_name assets = generate_psd_assets( psd_path=new_path, output_dir=assets_dir, ) # Clean old files safe_delete(Path(tpl.file_path)) if tpl.base_image_path: safe_delete(Path(tpl.base_image_path)) if tpl.base_thumbnail_path: safe_delete(Path(tpl.base_thumbnail_path)) if tpl.main_black_mask_path: safe_delete(Path(tpl.main_black_mask_path)) if tpl.main_white_mask_path: safe_delete(Path(tpl.main_white_mask_path)) if tpl.logo_black_mask_path: safe_delete(Path(tpl.logo_black_mask_path)) if tpl.logo_white_mask_path: safe_delete(Path(tpl.logo_white_mask_path)) if tpl.color_preview_path: safe_delete(Path(tpl.color_preview_path)) if tpl.color_white_mask_path: safe_delete(Path(tpl.color_white_mask_path)) # Update paths tpl.file_path = str(new_path) tpl.ratio = ratio tpl.width = width tpl.height = height tpl.base_image_path = assets["base_image"] tpl.base_thumbnail_path=assets["base_thumbnail"] tpl.main_black_mask_path=assets["main_black_mask_path"] tpl.main_white_mask_path=assets["main_white_mask_path"] tpl.logo_black_mask_path=assets["logo_black_mask_path"] tpl.logo_white_mask_path=assets["logo_white_mask_path"] tpl.color_preview_path=assets["color_preview_path"] tpl.color_white_mask_path=assets["color_white_mask_path"] tpl = ProductTemplateRepository.update(session, tpl) return tpl @staticmethod def list_templates( session: Session, page: int, limit: int, search: str | None, sort_by: str | None, order: str | None, ): """Return all active (not deleted) templates.""" items, total = ProductTemplateRepository.list_advanced( session=session, page=page, limit=limit, search=search, sort_by=sort_by, order=order, ) return PaginatedResponse[ProductTemplateRead]( total=total, page=page, limit=limit, items=items, ) @staticmethod def get_template_by_id(template_id: int, session: Session) -> ProductTemplate: """ Retrieve a template ensuring it exists and is not deleted. """ tpl = ProductTemplateRepository.get_by_id(session, template_id) if not tpl or tpl.is_deleted: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERR_NOT_FOUND) return tpl @staticmethod def delete_template(template_id: int, session: Session): """Soft-delete template and remove files (non-critical).""" tpl = ProductTemplateRepository.get_by_id(session, template_id) if not tpl: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERR_NOT_FOUND) ProductTemplateRepository.soft_delete_template(session, tpl) # best-effort deletes safe_delete(Path(tpl.file_path)) if tpl.base_image_path: safe_delete(Path(tpl.base_image_path)) if tpl.base_thumbnail_path: safe_delete(Path(tpl.base_thumbnail_path)) if tpl.main_black_mask_path: safe_delete(Path(tpl.main_black_mask_path)) if tpl.main_white_mask_path: safe_delete(Path(tpl.main_white_mask_path)) if tpl.logo_black_mask_path: safe_delete(Path(tpl.logo_black_mask_path)) if tpl.logo_white_mask_path: safe_delete(Path(tpl.logo_white_mask_path)) if tpl.color_preview_path: safe_delete(Path(tpl.color_preview_path)) if tpl.color_white_mask_path: safe_delete(Path(tpl.color_white_mask_path))