"""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)}")