File size: 6,790 Bytes
d4a4da7 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 |
"""Creative upload, analyze, and modify endpoints."""
import os
from datetime import datetime
from fastapi import APIRouter, HTTPException, Depends, File, UploadFile
from api.schemas import (
CreativeAnalyzeRequest,
CreativeAnalysisResponse,
CreativeModifyRequest,
CreativeModifyResponse,
FileUploadResponse,
)
from services.creative_modifier import creative_modifier_service
from services.image import image_service
from services.auth_dependency import get_current_user
from config import settings
router = APIRouter(tags=["creative"])
@router.post("/api/creative/upload", response_model=FileUploadResponse)
async def upload_creative(
file: UploadFile = File(...),
username: str = Depends(get_current_user),
):
"""
Upload a creative image for analysis and modification.
Accepts PNG, JPG, JPEG, WebP. Returns image URL for subsequent steps.
"""
allowed_types = ["image/png", "image/jpeg", "image/jpg", "image/webp"]
if file.content_type not in allowed_types:
raise HTTPException(status_code=400, detail=f"Invalid file type. Allowed: PNG, JPG, JPEG, WebP. Got: {file.content_type}")
contents = await file.read()
if len(contents) > 10 * 1024 * 1024:
raise HTTPException(status_code=400, detail="File too large. Maximum size is 10MB.")
try:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
unique_id = __import__("uuid").uuid4().hex[:8]
ext = file.filename.split(".")[-1] if file.filename else "png"
filename = f"upload_{username}_{timestamp}_{unique_id}.{ext}"
r2_url = None
try:
from services.r2_storage import get_r2_storage
r2_storage = get_r2_storage()
if r2_storage:
r2_url = r2_storage.upload_image(image_bytes=contents, filename=filename, niche="uploads")
except Exception:
pass
if not r2_url:
local_path = os.path.join(settings.output_dir, filename)
os.makedirs(os.path.dirname(local_path), exist_ok=True)
with open(local_path, "wb") as f:
f.write(contents)
r2_url = f"/images/{filename}"
return {"status": "success", "image_url": r2_url, "filename": filename}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.post("/api/creative/analyze", response_model=CreativeAnalysisResponse)
async def analyze_creative(
request: CreativeAnalyzeRequest,
username: str = Depends(get_current_user),
):
"""Analyze a creative image using AI vision (via URL)."""
if not request.image_url:
raise HTTPException(status_code=400, detail="image_url must be provided")
try:
image_bytes = await image_service.load_image(image_url=request.image_url)
except Exception as e:
raise HTTPException(status_code=400, detail=f"Failed to fetch image from URL: {e}")
if not image_bytes:
raise HTTPException(status_code=400, detail="Failed to load image")
try:
result = await creative_modifier_service.analyze_creative(image_bytes)
if result["status"] != "success":
return CreativeAnalysisResponse(status="error", error=result.get("error", "Analysis failed"))
return CreativeAnalysisResponse(
status="success",
analysis=result.get("analysis"),
suggested_angles=result.get("suggested_angles"),
suggested_concepts=result.get("suggested_concepts"),
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.post("/api/creative/analyze/upload", response_model=CreativeAnalysisResponse)
async def analyze_creative_upload(
file: UploadFile = File(...),
username: str = Depends(get_current_user),
):
"""Analyze a creative image using AI vision (via file upload)."""
allowed_types = ["image/png", "image/jpeg", "image/jpg", "image/webp"]
if file.content_type not in allowed_types:
raise HTTPException(status_code=400, detail=f"Invalid file type. Allowed: PNG, JPG, JPEG, WebP. Got: {file.content_type}")
image_bytes = await file.read()
if not image_bytes:
raise HTTPException(status_code=400, detail="Failed to load image")
try:
result = await creative_modifier_service.analyze_creative(image_bytes)
if result["status"] != "success":
return CreativeAnalysisResponse(status="error", error=result.get("error", "Analysis failed"))
return CreativeAnalysisResponse(
status="success",
analysis=result.get("analysis"),
suggested_angles=result.get("suggested_angles"),
suggested_concepts=result.get("suggested_concepts"),
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.post("/api/creative/modify", response_model=CreativeModifyResponse)
async def modify_creative(
request: CreativeModifyRequest,
username: str = Depends(get_current_user),
):
"""
Modify a creative with angle and/or concept.
Modes: 'modify' (image-to-image) or 'inspired' (new generation).
"""
if not request.angle and not request.concept:
raise HTTPException(status_code=400, detail="At least one of 'angle' or 'concept' must be provided")
analysis = request.analysis
if not analysis:
try:
image_bytes = await image_service.load_image(image_url=request.image_url)
if not image_bytes:
raise HTTPException(status_code=400, detail="Failed to load image from URL")
analysis_result = await creative_modifier_service.analyze_creative(image_bytes)
if analysis_result["status"] != "success":
raise HTTPException(status_code=500, detail=analysis_result.get("error", "Analysis failed"))
analysis = analysis_result.get("analysis", {})
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to analyze image: {e}")
try:
result = await creative_modifier_service.modify_creative(
image_url=request.image_url,
analysis=analysis,
user_angle=request.angle,
user_concept=request.concept,
mode=request.mode,
image_model=request.image_model,
user_prompt=request.user_prompt,
)
if result["status"] != "success":
return CreativeModifyResponse(status="error", error=result.get("error", "Modification failed"))
return CreativeModifyResponse(status="success", prompt=result.get("prompt"), image=result.get("image"))
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
|