757 lines
32 KiB
Python
757 lines
32 KiB
Python
"""
|
|
TheNicheQuiz.com - AI-Powered Facebook Ads Niche Finder & Campaign Generator
|
|
Production Flask application with auth, landing page, and full campaign builder.
|
|
"""
|
|
|
|
import os
|
|
import csv
|
|
import io
|
|
import json
|
|
import uuid
|
|
import subprocess
|
|
import time
|
|
import secrets
|
|
from datetime import datetime
|
|
from functools import wraps
|
|
|
|
import bcrypt
|
|
import psycopg2
|
|
import psycopg2.extras
|
|
from flask import Flask, render_template, request, jsonify, send_file, send_from_directory, redirect, url_for, flash, session
|
|
from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required, current_user
|
|
|
|
import google.generativeai as genai
|
|
|
|
# ── App Setup ─────────────────────────────────────────────────────────────────
|
|
|
|
app = Flask(__name__)
|
|
app.config['SECRET_KEY'] = secrets.token_hex(32)
|
|
app.config['MAX_CONTENT_LENGTH'] = 50 * 1024 * 1024 # 50MB max
|
|
|
|
GENERATED_DIR = os.path.join(os.path.dirname(__file__), 'static', 'generated')
|
|
os.makedirs(GENERATED_DIR, exist_ok=True)
|
|
|
|
GEMINI_API_KEY = 'AIzaSyClMlVU3Z1jh1UBxTRn25yesH8RU1q_umY'
|
|
genai.configure(api_key=GEMINI_API_KEY)
|
|
gemini_model = genai.GenerativeModel('gemini-2.0-flash')
|
|
|
|
# ── Database ──────────────────────────────────────────────────────────────────
|
|
|
|
def get_db():
|
|
conn = psycopg2.connect(dbname='nichequiz', host='localhost')
|
|
conn.autocommit = True
|
|
return conn
|
|
|
|
# ── Flask-Login Setup ─────────────────────────────────────────────────────────
|
|
|
|
login_manager = LoginManager()
|
|
login_manager.init_app(app)
|
|
login_manager.login_view = 'login'
|
|
login_manager.login_message = ''
|
|
login_manager.login_message_category = 'info'
|
|
|
|
|
|
class User(UserMixin):
|
|
def __init__(self, id, email, name, created_at=None, last_login=None):
|
|
self.id = id
|
|
self.email = email
|
|
self.name = name
|
|
self.created_at = created_at
|
|
self.last_login = last_login
|
|
|
|
|
|
@login_manager.user_loader
|
|
def load_user(user_id):
|
|
try:
|
|
conn = get_db()
|
|
cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
|
|
cur.execute("SELECT id, email, name, created_at, last_login FROM users WHERE id = %s", (int(user_id),))
|
|
row = cur.fetchone()
|
|
cur.close()
|
|
conn.close()
|
|
if row:
|
|
return User(id=row['id'], email=row['email'], name=row['name'],
|
|
created_at=row['created_at'], last_login=row['last_login'])
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
# ── Field Definitions ─────────────────────────────────────────────────────────
|
|
|
|
FIELD_DEFINITIONS = {
|
|
"campaign": {
|
|
"label": "Campaign",
|
|
"fields": [
|
|
{"key": "campaign_name", "label": "Campaign Name", "type": "text", "required": True, "placeholder": "e.g. Summer Sale 2025"},
|
|
{"key": "campaign_objective", "label": "Campaign Objective", "type": "select", "required": True,
|
|
"options": ["OUTCOME_TRAFFIC", "OUTCOME_SALES", "OUTCOME_LEADS", "OUTCOME_ENGAGEMENT", "OUTCOME_APP_PROMOTION", "OUTCOME_AWARENESS"]},
|
|
{"key": "buying_type", "label": "Buying Type", "type": "select", "required": False,
|
|
"options": ["AUCTION", "RESERVATION"], "default": "AUCTION"},
|
|
{"key": "campaign_status", "label": "Campaign Status", "type": "select", "required": True,
|
|
"options": ["ACTIVE", "PAUSED"], "default": "PAUSED"},
|
|
{"key": "campaign_budget_optimization", "label": "Campaign Budget Optimization (CBO)", "type": "select", "required": False,
|
|
"options": ["TRUE", "FALSE"], "default": "FALSE"},
|
|
{"key": "campaign_budget", "label": "Campaign Budget", "type": "number", "required": False, "placeholder": "e.g. 50.00"},
|
|
{"key": "campaign_bid_strategy", "label": "Campaign Bid Strategy", "type": "select", "required": False,
|
|
"options": ["", "LOWEST_COST_WITHOUT_CAP", "LOWEST_COST_WITH_BID_CAP", "COST_CAP", "LOWEST_COST_WITH_MIN_ROAS"]},
|
|
{"key": "campaign_spend_limit", "label": "Campaign Spend Limit", "type": "number", "required": False, "placeholder": "e.g. 1000.00"},
|
|
]
|
|
},
|
|
"adset": {
|
|
"label": "Ad Set",
|
|
"fields": [
|
|
{"key": "adset_name", "label": "Ad Set Name", "type": "text", "required": True, "placeholder": "e.g. US 25-45 Interest Targeting"},
|
|
{"key": "optimization_goal", "label": "Optimization Goal", "type": "select", "required": False,
|
|
"options": ["", "LINK_CLICKS", "LANDING_PAGE_VIEWS", "IMPRESSIONS", "REACH", "LEAD_GENERATION", "OFFSITE_CONVERSIONS", "VALUE", "APP_INSTALLS", "CONVERSATIONS"]},
|
|
{"key": "billing_event", "label": "Billing Event", "type": "select", "required": False,
|
|
"options": ["", "IMPRESSIONS", "LINK_CLICKS", "APP_INSTALLS", "THRUPLAY"], "default": "IMPRESSIONS"},
|
|
{"key": "bid_strategy", "label": "Bid Strategy", "type": "select", "required": False,
|
|
"options": ["", "LOWEST_COST_WITHOUT_CAP", "LOWEST_COST_WITH_BID_CAP", "COST_CAP"]},
|
|
{"key": "bid_amount", "label": "Bid Amount", "type": "number", "required": False, "placeholder": "e.g. 5.00"},
|
|
{"key": "daily_budget", "label": "Daily Budget", "type": "number", "required": False, "placeholder": "e.g. 20.00"},
|
|
{"key": "lifetime_budget", "label": "Lifetime Budget", "type": "number", "required": False, "placeholder": "e.g. 500.00"},
|
|
{"key": "start_date", "label": "Start Date", "type": "datetime-local", "required": False},
|
|
{"key": "end_date", "label": "End Date", "type": "datetime-local", "required": False},
|
|
{"key": "adset_status", "label": "Ad Set Status", "type": "select", "required": True,
|
|
"options": ["ACTIVE", "PAUSED"], "default": "PAUSED"},
|
|
{"key": "targeting_age_min", "label": "Targeting Age Min", "type": "number", "required": False},
|
|
{"key": "targeting_age_max", "label": "Targeting Age Max", "type": "number", "required": False},
|
|
{"key": "targeting_genders", "label": "Targeting Gender", "type": "select", "required": False,
|
|
"options": ["All", "Male", "Female"], "default": "All"},
|
|
{"key": "targeting_geo_locations", "label": "Targeting Locations", "type": "text", "required": False},
|
|
{"key": "targeting_interests", "label": "Targeting Interests", "type": "text", "required": False},
|
|
{"key": "targeting_behaviors", "label": "Targeting Behaviors", "type": "text", "required": False},
|
|
]
|
|
},
|
|
"ad": {
|
|
"label": "Ad Creative",
|
|
"fields": [
|
|
{"key": "ad_name", "label": "Ad Name", "type": "text", "required": True},
|
|
{"key": "ad_status", "label": "Ad Status", "type": "select", "required": True,
|
|
"options": ["ACTIVE", "PAUSED"], "default": "PAUSED"},
|
|
{"key": "body", "label": "Ad Body Text", "type": "textarea", "required": False, "maxlength": 2200},
|
|
{"key": "title", "label": "Headline", "type": "text", "required": False, "maxlength": 255},
|
|
{"key": "caption", "label": "Description", "type": "text", "required": False, "maxlength": 255},
|
|
{"key": "link", "label": "Link", "type": "url", "required": False},
|
|
{"key": "call_to_action", "label": "Call to Action", "type": "select", "required": False,
|
|
"options": ["", "LEARN_MORE", "SHOP_NOW", "SIGN_UP", "DOWNLOAD", "GET_QUOTE", "CONTACT_US", "BOOK_NOW", "APPLY_NOW", "NO_BUTTON"]},
|
|
{"key": "image_hash", "label": "Image Hash", "type": "text", "required": False},
|
|
{"key": "image_file", "label": "Image File", "type": "text", "required": False},
|
|
]
|
|
}
|
|
}
|
|
|
|
# CSV column mapping
|
|
CSV_COLUMNS = [
|
|
"Campaign Name", "Campaign Objective", "Buying Type", "Campaign Status",
|
|
"Campaign Budget Optimization", "Campaign Budget", "Campaign Bid Strategy", "Campaign Spend Limit",
|
|
"Ad Set Name", "Optimization Goal", "Billing Event", "Bid Strategy", "Bid Amount",
|
|
"Daily Budget", "Lifetime Budget", "Start Date", "End Date", "Ad Set Status",
|
|
"Targeting Age Min", "Targeting Age Max", "Targeting Genders", "Targeting Geo Locations",
|
|
"Targeting Custom Audiences", "Targeting Excluded Custom Audiences",
|
|
"Targeting Interests", "Targeting Behaviors", "Targeting Locales",
|
|
"Placements", "Platform Positions", "Optimization Window", "Attribution Setting",
|
|
"Conversion Window", "Promoted Object Pixel ID", "Promoted Object Custom Event Type", "Destination Type",
|
|
"Ad Name", "Ad Status", "Body", "Title", "Caption", "Link", "Call To Action",
|
|
"Image Hash", "Image File", "Video ID", "Video File", "Website URL", "Display Link",
|
|
"UTM Source", "UTM Medium", "UTM Campaign", "UTM Term", "UTM Content",
|
|
"Dynamic Creative", "Facebook Page ID"
|
|
]
|
|
|
|
FIELD_TO_CSV = {
|
|
"campaign_name": "Campaign Name", "campaign_objective": "Campaign Objective",
|
|
"buying_type": "Buying Type", "campaign_status": "Campaign Status",
|
|
"campaign_budget_optimization": "Campaign Budget Optimization",
|
|
"campaign_budget": "Campaign Budget", "campaign_bid_strategy": "Campaign Bid Strategy",
|
|
"campaign_spend_limit": "Campaign Spend Limit", "adset_name": "Ad Set Name",
|
|
"optimization_goal": "Optimization Goal", "billing_event": "Billing Event",
|
|
"bid_strategy": "Bid Strategy", "bid_amount": "Bid Amount",
|
|
"daily_budget": "Daily Budget", "lifetime_budget": "Lifetime Budget",
|
|
"start_date": "Start Date", "end_date": "End Date", "adset_status": "Ad Set Status",
|
|
"targeting_age_min": "Targeting Age Min", "targeting_age_max": "Targeting Age Max",
|
|
"targeting_genders": "Targeting Genders", "targeting_geo_locations": "Targeting Geo Locations",
|
|
"targeting_custom_audiences": "Targeting Custom Audiences",
|
|
"targeting_excluded_custom_audiences": "Targeting Excluded Custom Audiences",
|
|
"targeting_interests": "Targeting Interests", "targeting_behaviors": "Targeting Behaviors",
|
|
"targeting_locales": "Targeting Locales", "placements": "Placements",
|
|
"platform_positions": "Platform Positions", "optimization_window": "Optimization Window",
|
|
"attribution_setting": "Attribution Setting", "conversion_window": "Conversion Window",
|
|
"promoted_object_pixel_id": "Promoted Object Pixel ID",
|
|
"promoted_object_custom_event_type": "Promoted Object Custom Event Type",
|
|
"destination_type": "Destination Type", "ad_name": "Ad Name", "ad_status": "Ad Status",
|
|
"body": "Body", "title": "Title", "caption": "Caption", "link": "Link",
|
|
"call_to_action": "Call To Action", "image_hash": "Image Hash", "image_file": "Image File",
|
|
"video_id": "Video ID", "video_file": "Video File", "website_url": "Website URL",
|
|
"display_link": "Display Link", "utm_source": "UTM Source", "utm_medium": "UTM Medium",
|
|
"utm_campaign": "UTM Campaign", "utm_term": "UTM Term", "utm_content": "UTM Content",
|
|
"dynamic_creative": "Dynamic Creative", "facebook_page_id": "Facebook Page ID",
|
|
}
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# PUBLIC ROUTES (No Auth Required)
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@app.route('/')
|
|
def landing():
|
|
if current_user.is_authenticated:
|
|
return redirect(url_for('app_dashboard'))
|
|
return render_template('landing.html')
|
|
|
|
|
|
@app.route('/signup', methods=['GET', 'POST'])
|
|
def signup():
|
|
if current_user.is_authenticated:
|
|
return redirect(url_for('app_dashboard'))
|
|
|
|
if request.method == 'POST':
|
|
name = request.form.get('name', '').strip()
|
|
email = request.form.get('email', '').strip().lower()
|
|
password = request.form.get('password', '').strip()
|
|
|
|
if not name or not email or not password:
|
|
return render_template('signup.html', error='All fields are required.')
|
|
|
|
if len(password) < 6:
|
|
return render_template('signup.html', error='Password must be at least 6 characters.', name=name, email=email)
|
|
|
|
try:
|
|
conn = get_db()
|
|
cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
|
|
|
|
# Check if email exists
|
|
cur.execute("SELECT id FROM users WHERE email = %s", (email,))
|
|
if cur.fetchone():
|
|
cur.close()
|
|
conn.close()
|
|
return render_template('signup.html', error='An account with that email already exists.', name=name)
|
|
|
|
# Hash password
|
|
password_hash = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
|
|
|
|
# Insert user
|
|
cur.execute(
|
|
"INSERT INTO users (name, email, password_hash, created_at) VALUES (%s, %s, %s, NOW()) RETURNING id",
|
|
(name, email, password_hash)
|
|
)
|
|
user_id = cur.fetchone()['id']
|
|
cur.close()
|
|
conn.close()
|
|
|
|
# Log the user in
|
|
user = User(id=user_id, email=email, name=name)
|
|
login_user(user)
|
|
|
|
return redirect(url_for('app_generate'))
|
|
|
|
except Exception as e:
|
|
return render_template('signup.html', error=f'Something went wrong. Please try again.', name=name, email=email)
|
|
|
|
return render_template('signup.html')
|
|
|
|
|
|
@app.route('/login', methods=['GET', 'POST'])
|
|
def login():
|
|
if current_user.is_authenticated:
|
|
return redirect(url_for('app_dashboard'))
|
|
|
|
if request.method == 'POST':
|
|
email = request.form.get('email', '').strip().lower()
|
|
password = request.form.get('password', '').strip()
|
|
|
|
if not email or not password:
|
|
return render_template('login.html', error='Email and password are required.')
|
|
|
|
try:
|
|
conn = get_db()
|
|
cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
|
|
cur.execute("SELECT id, email, name, password_hash FROM users WHERE email = %s", (email,))
|
|
row = cur.fetchone()
|
|
|
|
if row and bcrypt.checkpw(password.encode('utf-8'), row['password_hash'].encode('utf-8')):
|
|
# Update last login
|
|
cur.execute("UPDATE users SET last_login = NOW() WHERE id = %s", (row['id'],))
|
|
cur.close()
|
|
conn.close()
|
|
|
|
user = User(id=row['id'], email=row['email'], name=row['name'])
|
|
login_user(user, remember=True)
|
|
|
|
next_page = request.args.get('next')
|
|
return redirect(next_page or url_for('app_dashboard'))
|
|
else:
|
|
cur.close()
|
|
conn.close()
|
|
return render_template('login.html', error='Invalid email or password.', email=email)
|
|
|
|
except Exception as e:
|
|
return render_template('login.html', error='Something went wrong. Please try again.', email=email)
|
|
|
|
return render_template('login.html')
|
|
|
|
|
|
@app.route('/api/signup', methods=['POST'])
|
|
def api_signup():
|
|
data = request.json or {}
|
|
name = data.get('name', '').strip()
|
|
email = data.get('email', '').strip().lower()
|
|
password = data.get('password', '').strip()
|
|
|
|
if not name or not email or not password:
|
|
return jsonify({"success": False, "error": "All fields are required."}), 400
|
|
if len(password) < 6:
|
|
return jsonify({"success": False, "error": "Password must be at least 6 characters."}), 400
|
|
|
|
try:
|
|
conn = get_db()
|
|
cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
|
|
cur.execute("SELECT id FROM users WHERE email = %s", (email,))
|
|
if cur.fetchone():
|
|
cur.close()
|
|
conn.close()
|
|
return jsonify({"success": False, "error": "An account with that email already exists."}), 409
|
|
|
|
password_hash = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
|
|
cur.execute(
|
|
"INSERT INTO users (name, email, password_hash, created_at) VALUES (%s, %s, %s, NOW()) RETURNING id",
|
|
(name, email, password_hash)
|
|
)
|
|
user_id = cur.fetchone()['id']
|
|
cur.close()
|
|
conn.close()
|
|
|
|
user = User(id=user_id, email=email, name=name)
|
|
login_user(user)
|
|
return jsonify({"success": True, "redirect": "/app/generate"})
|
|
except Exception as e:
|
|
return jsonify({"success": False, "error": "Something went wrong. Please try again."}), 500
|
|
|
|
|
|
@app.route('/api/login', methods=['POST'])
|
|
def api_login():
|
|
data = request.json or {}
|
|
email = data.get('email', '').strip().lower()
|
|
password = data.get('password', '').strip()
|
|
|
|
if not email or not password:
|
|
return jsonify({"success": False, "error": "Email and password are required."}), 400
|
|
|
|
try:
|
|
conn = get_db()
|
|
cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
|
|
cur.execute("SELECT id, email, name, password_hash FROM users WHERE email = %s", (email,))
|
|
row = cur.fetchone()
|
|
|
|
if row and bcrypt.checkpw(password.encode('utf-8'), row['password_hash'].encode('utf-8')):
|
|
cur.execute("UPDATE users SET last_login = NOW() WHERE id = %s", (row['id'],))
|
|
cur.close()
|
|
conn.close()
|
|
user = User(id=row['id'], email=row['email'], name=row['name'])
|
|
login_user(user, remember=True)
|
|
next_page = request.args.get('next', '/app')
|
|
return jsonify({"success": True, "redirect": next_page})
|
|
else:
|
|
cur.close()
|
|
conn.close()
|
|
return jsonify({"success": False, "error": "Invalid email or password."}), 401
|
|
except Exception as e:
|
|
return jsonify({"success": False, "error": "Something went wrong. Please try again."}), 500
|
|
|
|
|
|
@app.route('/logout')
|
|
@login_required
|
|
def logout():
|
|
logout_user()
|
|
return redirect(url_for('landing'))
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# APP ROUTES (Auth Required)
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@app.route('/app')
|
|
@login_required
|
|
def app_dashboard():
|
|
try:
|
|
conn = get_db()
|
|
cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
|
|
|
|
# Get campaign count
|
|
cur.execute("SELECT COUNT(*) as count FROM campaigns WHERE user_id = %s", (current_user.id,))
|
|
campaign_count = cur.fetchone()['count']
|
|
|
|
# Get image count
|
|
cur.execute("SELECT COUNT(*) as count FROM generated_images WHERE user_id = %s", (current_user.id,))
|
|
image_count = cur.fetchone()['count']
|
|
|
|
# Get export count
|
|
cur.execute("SELECT COUNT(*) as count FROM csv_exports WHERE user_id = %s", (current_user.id,))
|
|
export_count = cur.fetchone()['count']
|
|
|
|
# Get recent campaigns
|
|
cur.execute(
|
|
"SELECT id, name, industry, sub_niche, micro_niche, created_at FROM campaigns WHERE user_id = %s ORDER BY created_at DESC LIMIT 5",
|
|
(current_user.id,)
|
|
)
|
|
recent_campaigns = cur.fetchall()
|
|
|
|
cur.close()
|
|
conn.close()
|
|
|
|
return render_template('dashboard.html',
|
|
campaign_count=campaign_count,
|
|
image_count=image_count,
|
|
export_count=export_count,
|
|
recent_campaigns=recent_campaigns)
|
|
except Exception as e:
|
|
return render_template('dashboard.html',
|
|
campaign_count=0, image_count=0, export_count=0, recent_campaigns=[])
|
|
|
|
|
|
@app.route('/app/generate')
|
|
@login_required
|
|
def app_generate():
|
|
return render_template('generate.html')
|
|
|
|
|
|
@app.route('/app/campaigns')
|
|
@login_required
|
|
def app_campaigns():
|
|
try:
|
|
conn = get_db()
|
|
cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
|
|
cur.execute(
|
|
"SELECT id, name, industry, sub_niche, micro_niche, campaign_data, created_at FROM campaigns WHERE user_id = %s ORDER BY created_at DESC",
|
|
(current_user.id,)
|
|
)
|
|
campaigns = cur.fetchall()
|
|
cur.close()
|
|
conn.close()
|
|
return render_template('campaigns.html', campaigns=campaigns)
|
|
except Exception:
|
|
return render_template('campaigns.html', campaigns=[])
|
|
|
|
|
|
@app.route('/app/campaigns/<int:campaign_id>')
|
|
@login_required
|
|
def app_campaign_detail(campaign_id):
|
|
try:
|
|
conn = get_db()
|
|
cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
|
|
cur.execute(
|
|
"SELECT * FROM campaigns WHERE id = %s AND user_id = %s",
|
|
(campaign_id, current_user.id)
|
|
)
|
|
campaign = cur.fetchone()
|
|
cur.close()
|
|
conn.close()
|
|
if not campaign:
|
|
return redirect(url_for('app_campaigns'))
|
|
return render_template('campaign_detail.html', campaign=campaign)
|
|
except Exception:
|
|
return redirect(url_for('app_campaigns'))
|
|
|
|
|
|
@app.route('/app/account')
|
|
@login_required
|
|
def app_account():
|
|
return render_template('account.html')
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# API ROUTES
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@app.route('/api/fields')
|
|
def get_fields():
|
|
return jsonify(FIELD_DEFINITIONS)
|
|
|
|
|
|
@app.route('/api/suggest-niches', methods=['POST'])
|
|
@login_required
|
|
def suggest_niches():
|
|
data = request.json
|
|
industry = data.get('industry', '').strip()
|
|
sub_niche = data.get('sub_niche', '').strip()
|
|
|
|
if not industry:
|
|
return jsonify({"error": "Industry is required"}), 400
|
|
|
|
try:
|
|
if sub_niche:
|
|
prompt = (
|
|
f"List 10 ultra-specific micro-niches within '{sub_niche}' (which is a sub-niche of '{industry}'). "
|
|
f"These should be extremely specific, the kind of niche that would crush on Facebook ads. "
|
|
f"Think: specific demographic + specific problem + specific solution. "
|
|
f"Examples of the specificity level: 'Leaky Gut Protocol for Busy Moms Over 40', "
|
|
f"'Keto Meal Prep for Type 2 Diabetic Men', 'Anxiety Relief Through Breathwork for College Students'. "
|
|
f"Just list them, one per line, no numbers or bullets or extra explanation."
|
|
)
|
|
else:
|
|
prompt = (
|
|
f"List 10 specific sub-niches within the '{industry}' industry. "
|
|
f"These should be specific enough to target but broad enough to have an audience. "
|
|
f"Just list them, one per line, no numbers or bullets or extra explanation."
|
|
)
|
|
|
|
response = gemini_model.generate_content(prompt)
|
|
text = response.text.strip()
|
|
niches = [line.strip().strip('-').strip('•').strip('*').strip() for line in text.split('\n') if line.strip()]
|
|
niches = [n for n in niches if len(n) > 2][:12]
|
|
|
|
return jsonify({"success": True, "niches": niches})
|
|
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
@app.route('/api/generate-parallel-campaigns', methods=['POST'])
|
|
@login_required
|
|
def generate_parallel_campaigns():
|
|
data = request.json
|
|
industry = data.get('industry', '').strip()
|
|
sub_niche = data.get('sub_niche', '').strip()
|
|
micro_niche = data.get('micro_niche', '').strip()
|
|
|
|
if not industry or not sub_niche or not micro_niche:
|
|
return jsonify({"error": "Industry, sub-niche, and micro-niche are all required"}), 400
|
|
|
|
try:
|
|
prompt = f"""You are a Facebook Ads strategist. Given this niche path:
|
|
Industry: {industry}
|
|
Sub-niche: {sub_niche}
|
|
Chosen micro-niche: {micro_niche}
|
|
|
|
Generate 10 OTHER parallel micro-niches that are siblings of "{micro_niche}" — they should be within "{sub_niche}" but target different specific angles/demographics/problems.
|
|
|
|
For EACH of the 10 parallel micro-niches, provide a complete Facebook ad campaign in this EXACT JSON format. Return a JSON array of 10 objects:
|
|
|
|
[
|
|
{{
|
|
"micro_niche": "Ultra-specific micro-niche name",
|
|
"campaign_name": "Campaign name (short, descriptive)",
|
|
"adset_name": "Ad set name with targeting description",
|
|
"targeting": {{
|
|
"age_min": 25,
|
|
"age_max": 55,
|
|
"gender": "All",
|
|
"interests": "interest1, interest2, interest3",
|
|
"behaviors": "behavior1, behavior2"
|
|
}},
|
|
"ad_copy": {{
|
|
"body": "The primary ad text (2-3 sentences, compelling, with emoji)",
|
|
"headline": "Short punchy headline",
|
|
"description": "Description line under headline"
|
|
}},
|
|
"image_prompt": "Detailed prompt for AI image generation - describe a compelling ad image",
|
|
"cta": "LEARN_MORE",
|
|
"landing_page_angle": "Brief description of what the landing page should focus on"
|
|
}}
|
|
]
|
|
|
|
Make the ad copy compelling and direct-response focused. Use emotional triggers. Each campaign should feel unique.
|
|
Valid CTAs: LEARN_MORE, SHOP_NOW, SIGN_UP, DOWNLOAD, GET_QUOTE, CONTACT_US, BOOK_NOW, APPLY_NOW
|
|
Return ONLY the JSON array, no markdown formatting, no code blocks."""
|
|
|
|
response = gemini_model.generate_content(prompt)
|
|
text = response.text.strip()
|
|
|
|
if text.startswith('```'):
|
|
text = text.split('\n', 1)[1] if '\n' in text else text[3:]
|
|
if text.endswith('```'):
|
|
text = text[:-3]
|
|
if text.startswith('json'):
|
|
text = text[4:]
|
|
text = text.strip()
|
|
|
|
campaigns = json.loads(text)
|
|
|
|
valid_ctas = ['LEARN_MORE', 'SHOP_NOW', 'SIGN_UP', 'DOWNLOAD', 'GET_QUOTE', 'CONTACT_US', 'BOOK_NOW', 'APPLY_NOW']
|
|
for c in campaigns:
|
|
if c.get('cta') not in valid_ctas:
|
|
c['cta'] = 'LEARN_MORE'
|
|
c['selected'] = True
|
|
c['image_url'] = None
|
|
|
|
return jsonify({"success": True, "campaigns": campaigns[:10]})
|
|
|
|
except json.JSONDecodeError as e:
|
|
return jsonify({"error": f"Failed to parse AI response as JSON: {str(e)}"}), 500
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
@app.route('/api/generate-image', methods=['POST'])
|
|
@login_required
|
|
def generate_image():
|
|
data = request.json
|
|
prompt = data.get('prompt', '').strip()
|
|
if not prompt:
|
|
return jsonify({"error": "Prompt is required"}), 400
|
|
|
|
filename = f"ad_{uuid.uuid4().hex[:12]}.png"
|
|
output_path = os.path.join(GENERATED_DIR, filename)
|
|
|
|
cmd = [
|
|
"uv", "run",
|
|
"/opt/homebrew/lib/node_modules/clawdbot/skills/nano-banana-pro/scripts/generate_image.py",
|
|
"--prompt", prompt,
|
|
"--filename", output_path,
|
|
"--resolution", "1K"
|
|
]
|
|
|
|
try:
|
|
env = os.environ.copy()
|
|
env['GEMINI_API_KEY'] = GEMINI_API_KEY
|
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=120, env=env)
|
|
if result.returncode != 0:
|
|
return jsonify({"error": f"Image generation failed: {result.stderr or result.stdout}"}), 500
|
|
|
|
for _ in range(10):
|
|
if os.path.exists(output_path) and os.path.getsize(output_path) > 0:
|
|
break
|
|
time.sleep(0.5)
|
|
|
|
if not os.path.exists(output_path):
|
|
return jsonify({"error": "Image file was not created"}), 500
|
|
|
|
# Track in database
|
|
try:
|
|
conn = get_db()
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
"INSERT INTO generated_images (user_id, filename, prompt) VALUES (%s, %s, %s)",
|
|
(current_user.id, filename, prompt)
|
|
)
|
|
cur.close()
|
|
conn.close()
|
|
except Exception:
|
|
pass
|
|
|
|
return jsonify({
|
|
"success": True,
|
|
"filename": filename,
|
|
"url": f"/static/generated/{filename}",
|
|
"prompt": prompt
|
|
})
|
|
|
|
except subprocess.TimeoutExpired:
|
|
return jsonify({"error": "Image generation timed out (120s)"}), 504
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
@app.route('/api/generate-csv', methods=['POST'])
|
|
@login_required
|
|
def generate_csv():
|
|
data = request.json
|
|
ads = data.get('ads', [])
|
|
|
|
if not ads:
|
|
return jsonify({"error": "No ad data provided"}), 400
|
|
|
|
output = io.StringIO()
|
|
writer = csv.DictWriter(output, fieldnames=CSV_COLUMNS, extrasaction='ignore')
|
|
writer.writeheader()
|
|
|
|
rows = []
|
|
for ad_data in ads:
|
|
row = {}
|
|
for field_key, csv_col in FIELD_TO_CSV.items():
|
|
value = ad_data.get(field_key, '')
|
|
if value is None:
|
|
value = ''
|
|
row[csv_col] = str(value).strip()
|
|
writer.writerow(row)
|
|
rows.append(row)
|
|
|
|
csv_content = output.getvalue()
|
|
output.close()
|
|
|
|
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
|
csv_filename = f"fb_ads_{timestamp}.csv"
|
|
csv_path = os.path.join(GENERATED_DIR, csv_filename)
|
|
with open(csv_path, 'w', newline='') as f:
|
|
f.write(csv_content)
|
|
|
|
# Track in database
|
|
try:
|
|
conn = get_db()
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
"INSERT INTO csv_exports (user_id, filename, row_count) VALUES (%s, %s, %s)",
|
|
(current_user.id, csv_filename, len(rows))
|
|
)
|
|
cur.close()
|
|
conn.close()
|
|
except Exception:
|
|
pass
|
|
|
|
return jsonify({
|
|
"success": True,
|
|
"filename": csv_filename,
|
|
"download_url": f"/static/generated/{csv_filename}",
|
|
"csv_content": csv_content,
|
|
"rows": rows,
|
|
"columns": CSV_COLUMNS
|
|
})
|
|
|
|
|
|
@app.route('/api/save-campaign', methods=['POST'])
|
|
@login_required
|
|
def save_campaign():
|
|
data = request.json
|
|
name = data.get('name', '').strip()
|
|
industry = data.get('industry', '').strip()
|
|
sub_niche = data.get('sub_niche', '').strip()
|
|
micro_niche = data.get('micro_niche', '').strip()
|
|
campaign_data = data.get('campaign_data', {})
|
|
|
|
if not name:
|
|
return jsonify({"error": "Campaign name is required"}), 400
|
|
|
|
try:
|
|
conn = get_db()
|
|
cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
|
|
cur.execute(
|
|
"INSERT INTO campaigns (user_id, name, industry, sub_niche, micro_niche, campaign_data) VALUES (%s, %s, %s, %s, %s, %s) RETURNING id",
|
|
(current_user.id, name, industry, sub_niche, micro_niche, json.dumps(campaign_data))
|
|
)
|
|
campaign_id = cur.fetchone()['id']
|
|
cur.close()
|
|
conn.close()
|
|
|
|
return jsonify({"success": True, "campaign_id": campaign_id})
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
@app.route('/api/delete-campaign/<int:campaign_id>', methods=['DELETE'])
|
|
@login_required
|
|
def delete_campaign(campaign_id):
|
|
try:
|
|
conn = get_db()
|
|
cur = conn.cursor()
|
|
cur.execute("DELETE FROM generated_images WHERE campaign_id = %s AND user_id = %s", (campaign_id, current_user.id))
|
|
cur.execute("DELETE FROM campaigns WHERE id = %s AND user_id = %s", (campaign_id, current_user.id))
|
|
cur.close()
|
|
conn.close()
|
|
return jsonify({"success": True})
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
@app.route('/api/download-csv/<filename>')
|
|
@login_required
|
|
def download_csv(filename):
|
|
return send_from_directory(GENERATED_DIR, filename, as_attachment=True)
|
|
|
|
|
|
# ── Main ──────────────────────────────────────────────────────────────────────
|
|
|
|
if __name__ == '__main__':
|
|
print("\n✦ TheNicheQuiz.com")
|
|
print(f" → http://localhost:8877\n")
|
|
app.run(host='0.0.0.0', port=8877, debug=False)
|