""" Servicio de subida y gestión de imágenes. Sube imágenes locales y devuelve URLs públicas para las APIs de Meta. """ import os import uuid import shutil from pathlib import Path from typing import Optional from datetime import datetime from app.core.config import settings class ImageUploadService: """Servicio para subir imágenes y obtener URLs públicas.""" def __init__(self): self.upload_dir = Path("uploads/images") self.upload_dir.mkdir(parents=True, exist_ok=True) self._base_url: Optional[str] = None @property def base_url(self) -> str: """URL base para servir imágenes.""" if self._base_url: return self._base_url # En producción, usar dominio público if settings.APP_ENV == "production": return f"{settings.BUSINESS_WEBSITE}/uploads/images" # En desarrollo, usar localhost return "http://localhost:8000/uploads/images" def set_base_url(self, url: str): """Establecer URL base manualmente.""" self._base_url = url.rstrip("/") def _generate_filename(self, original_name: str) -> str: """Generar nombre único para archivo.""" ext = Path(original_name).suffix.lower() or ".png" timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") unique_id = uuid.uuid4().hex[:8] return f"{timestamp}_{unique_id}{ext}" async def upload_from_path(self, file_path: str) -> str: """ Subir imagen desde ruta local. Args: file_path: Ruta al archivo de imagen local Returns: URL pública de la imagen """ source = Path(file_path) if not source.exists(): raise FileNotFoundError(f"Archivo no encontrado: {file_path}") # Generar nombre único new_name = self._generate_filename(source.name) dest = self.upload_dir / new_name # Copiar archivo shutil.copy2(source, dest) return f"{self.base_url}/{new_name}" async def upload_from_bytes( self, data: bytes, filename: str = "image.png" ) -> str: """ Subir imagen desde bytes. Args: data: Contenido del archivo en bytes filename: Nombre original para extensión Returns: URL pública de la imagen """ new_name = self._generate_filename(filename) dest = self.upload_dir / new_name dest.write_bytes(data) return f"{self.base_url}/{new_name}" async def upload_generated_image(self, image_path: str) -> str: """ Subir imagen generada por ImageGenerator. Alias de upload_from_path para claridad. """ return await self.upload_from_path(image_path) def get_local_path(self, url: str) -> Optional[Path]: """Obtener ruta local desde URL.""" if self.base_url not in url: return None filename = url.split("/")[-1] path = self.upload_dir / filename return path if path.exists() else None def delete(self, url: str) -> bool: """Eliminar imagen por URL.""" path = self.get_local_path(url) if path and path.exists(): path.unlink() return True return False def cleanup_old(self, days: int = 30) -> int: """ Limpiar imágenes antiguas. Args: days: Eliminar imágenes más antiguas que estos días Returns: Cantidad de archivos eliminados """ import time cutoff = time.time() - (days * 86400) deleted = 0 for file in self.upload_dir.iterdir(): if file.is_file() and file.stat().st_mtime < cutoff: file.unlink() deleted += 1 return deleted def list_images(self, limit: int = 100) -> list: """Listar imágenes subidas.""" images = [] for file in sorted( self.upload_dir.iterdir(), key=lambda f: f.stat().st_mtime, reverse=True )[:limit]: if file.is_file(): images.append({ "filename": file.name, "url": f"{self.base_url}/{file.name}", "size": file.stat().st_size, "modified": datetime.fromtimestamp( file.stat().st_mtime ).isoformat() }) return images # Instancia global image_upload = ImageUploadService()