File size: 6,878 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 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 |
"""Database endpoints: stats, list ads, get/delete ad, edit copy."""
from typing import Optional
from fastapi import APIRouter, HTTPException, Depends
from api.schemas import DbStatsResponse, EditAdCopyRequest
from services.database import db_service
from services.auth_dependency import get_current_user
router = APIRouter(tags=["database"])
@router.get("/db/stats", response_model=DbStatsResponse)
async def get_database_stats(username: str = Depends(get_current_user)):
"""Get statistics about stored ad creatives for the current user."""
return await db_service.get_stats(username=username)
@router.get("/db/ads")
async def list_stored_ads(
niche: Optional[str] = None,
generation_method: Optional[str] = None,
limit: int = 50,
offset: int = 0,
username: str = Depends(get_current_user),
):
"""List ad creatives for the current user with optional filters and pagination."""
ads, total = await db_service.list_ad_creatives(
username=username,
niche=niche,
generation_method=generation_method,
limit=limit,
offset=offset,
)
return {
"total": total,
"limit": limit,
"offset": offset,
"ads": [
{
"id": str(ad.get("id", "")),
"niche": ad.get("niche", ""),
"title": ad.get("title"),
"headline": ad.get("headline", ""),
"primary_text": ad.get("primary_text"),
"description": ad.get("description"),
"body_story": ad.get("body_story"),
"cta": ad.get("cta", ""),
"psychological_angle": ad.get("psychological_angle", ""),
"image_url": ad.get("image_url"),
"r2_url": ad.get("r2_url"),
"image_filename": ad.get("image_filename"),
"image_model": ad.get("image_model"),
"angle_key": ad.get("angle_key"),
"concept_key": ad.get("concept_key"),
"generation_method": ad.get("generation_method", "standard"),
"created_at": ad.get("created_at"),
}
for ad in ads
],
}
@router.get("/db/ad/{ad_id}")
async def get_stored_ad(ad_id: str):
"""Get a specific ad creative by ID."""
ad = await db_service.get_ad_creative(ad_id)
if not ad:
raise HTTPException(status_code=404, detail=f"Ad '{ad_id}' not found")
return {
"id": str(ad.get("id", "")),
"niche": ad.get("niche", ""),
"title": ad.get("title"),
"headline": ad.get("headline", ""),
"primary_text": ad.get("primary_text"),
"description": ad.get("description"),
"body_story": ad.get("body_story"),
"cta": ad.get("cta", ""),
"psychological_angle": ad.get("psychological_angle", ""),
"why_it_works": ad.get("why_it_works"),
"image_url": ad.get("image_url"),
"image_filename": ad.get("image_filename"),
"image_model": ad.get("image_model"),
"image_seed": ad.get("image_seed"),
"r2_url": ad.get("r2_url"),
"angle_key": ad.get("angle_key"),
"angle_name": ad.get("angle_name"),
"angle_trigger": ad.get("angle_trigger"),
"angle_category": ad.get("angle_category"),
"concept_key": ad.get("concept_key"),
"concept_name": ad.get("concept_name"),
"concept_structure": ad.get("concept_structure"),
"concept_visual": ad.get("concept_visual"),
"concept_category": ad.get("concept_category"),
"generation_method": ad.get("generation_method", "standard"),
"metadata": ad.get("metadata"),
"created_at": ad.get("created_at"),
"updated_at": ad.get("updated_at"),
}
@router.delete("/db/ad/{ad_id}")
async def delete_stored_ad(ad_id: str, username: str = Depends(get_current_user)):
"""Delete an ad creative. Users can only delete their own ads."""
success = await db_service.delete_ad_creative(ad_id, username=username)
if not success:
raise HTTPException(status_code=404, detail=f"Ad '{ad_id}' not found or could not be deleted")
return {"success": True, "deleted_id": ad_id}
@router.post("/db/ad/edit")
async def edit_ad_copy(
request: EditAdCopyRequest,
username: str = Depends(get_current_user),
):
"""
Edit ad copy fields. Modes: manual (direct update) or ai (AI-improved version).
"""
from services.llm import LLMService
ad = await db_service.get_ad_creative(request.ad_id)
if not ad:
raise HTTPException(status_code=404, detail=f"Ad '{request.ad_id}' not found")
if ad.get("username") != username:
raise HTTPException(status_code=403, detail="You can only edit your own ads")
if request.mode == "manual":
success = await db_service.update_ad_creative(
ad_id=request.ad_id,
username=username,
**{request.field: request.value},
)
if not success:
raise HTTPException(status_code=500, detail="Failed to update ad")
return {"edited_value": request.value, "success": True}
llm_service = LLMService()
field_labels = {
"title": "title",
"headline": "headline",
"primary_text": "primary text",
"description": "description",
"body_story": "body story",
"cta": "call to action",
}
field_label = field_labels.get(request.field, request.field)
current_value = request.value
niche = ad.get("niche", "general")
system_prompt = f"""You are an expert copywriter specializing in high-converting ad copy for {niche.replace('_', ' ')}.
Your task is to improve the {field_label} while maintaining its core message and emotional impact.
Keep the same tone and style, but make it more compelling, clear, and effective."""
user_prompt = f"""Current {field_label}:\n{current_value}\n\n"""
if request.user_suggestion:
user_prompt += f"User's suggestion: {request.user_suggestion}\n\n"
user_prompt += f"""Please provide an improved version of this {field_label} that:
1. Maintains the core message and emotional impact
2. Is more compelling and engaging
3. Follows best practices for {field_label} in ad copy
4. {"Incorporates the user's suggestion" if request.user_suggestion else "Is optimized for conversion"}
Return ONLY the improved {field_label} text, without any explanations or additional text."""
try:
edited_value = await llm_service.generate(
prompt=user_prompt,
system_prompt=system_prompt,
temperature=0.7,
)
edited_value = edited_value.strip().strip('"').strip("'")
return {"edited_value": edited_value, "success": True}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to generate AI edit: {str(e)}")
|