Spaces:
Sleeping
Sleeping
File size: 2,778 Bytes
1061354 |
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 |
import json
from loguru import logger
from typing import Annotated
from fastapi import (
APIRouter,
HTTPException,
Depends
)
from fastapi.responses import StreamingResponse
from src.models import AdGenerationRequest, AdGenerationResponse
from src.service.ad_service import get_ad_service, AdService
from src.utils.helpers import generate_request_id
from src.config import settings
router = APIRouter(prefix=settings.API_V1_PREFIX, tags=["Advertisement Generation"])
@router.post(
"/generate",
response_model=AdGenerationResponse,
summary="Generate Advertisement",
description="Generate advertisement with optional image processing"
)
async def generate_ad(
request: AdGenerationRequest,
ad_service: Annotated[AdService, Depends(get_ad_service)],
):
"""
Generate advertisement content based on the provided request.
Parameters:
- request: AdGenerationRequest containing product details and ad settings.
Returns:
- AdGenerationResponse with generated ad content and metadata.
"""
try:
response = await ad_service.generate_ad(request)
return response
except HTTPException:
raise
except Exception as e:
raise HTTPException(
status_code=500,
detail={
"error": "generation_failed",
"message": str(e),
"request_id": generate_request_id()
}
)
@router.post(
"/generate-stream",
summary="Generate Advertisement (Streaming)",
description="Generate advertisement with streaming response"
)
async def generate_ad_streaming(
request: AdGenerationRequest,
ad_service: Annotated[AdService, Depends(get_ad_service)],
):
"""
Generate advertisement content with streaming response.
Parameters:
- request: AdGenerationRequest containing product details and ad settings.
Returns:
- StreamingResponse with chunks of generated ad content.
"""
async def stream_response():
try:
async for chunk in ad_service.generate_ad_streaming(request):
yield json.dumps(chunk, default=str) + "\n"
except Exception as e:
logger.info(e)
error_response = {
"status": "error",
"error_code": "generation_failed",
"message": str(e),
"request_id": generate_request_id()
}
yield json.dumps(error_response) + "\n"
return StreamingResponse(
stream_response(),
media_type="text/plain",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no" # Disable nginx buffering
}
) |