import gradio as gr
import google.generativeai as genai
import json
import os
import requests
from datetime import datetime
# Configure Gemini
GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY", "")
AIRTABLE_API_KEY = os.environ.get("AIRTABLE_API_KEY", "")
AIRTABLE_BASE_ID = os.environ.get("AIRTABLE_BASE_ID", "app7WFijpDJ46Cib3")
AIRTABLE_TABLE = "Pipeline Queue"
if GEMINI_API_KEY:
genai.configure(api_key=GEMINI_API_KEY)
NICHES = [
"passive income", "dropshipping Australia", "lawn care business",
"pet care business", "AI tools for small business", "freelancing",
"print on demand", "affiliate marketing", "ecommerce Australia",
"side hustle ideas", "digital products", "self publishing",
"social media marketing", "home based business", "trading for beginners",
"lawn mowing business Australia", "WarmPaws pet products"
]
# ── Global state to hold last results ──────────────────────────────────────────
last_ideas = []
def research_ebook_ideas(niche: str, count: int, api_key: str):
global last_ideas
key = api_key.strip() or GEMINI_API_KEY
if not key:
return "❌ Please enter a Gemini API key.", "{}", "[]", gr.update(visible=False)
try:
genai.configure(api_key=key)
model = genai.GenerativeModel(
model_name="gemini-2.0-flash",
system_instruction=(
"You are an expert SEO strategist and digital publishing consultant "
"for Brett Apps, an Australian self-publishing business. "
"Return ONLY valid JSON — no markdown, no backticks, no explanation."
)
)
# Step 1: SEO Keyword Research
kw_prompt = f"""Generate SEO keyword research for niche: "{niche}" targeting Australian audiences.
Return ONLY valid JSON:
{{
"niche": "{niche}",
"primary_keywords": ["kw1","kw2","kw3","kw4","kw5"],
"long_tail_keywords": ["lt1","lt2","lt3","lt4","lt5"],
"buyer_intent_keywords": ["bi1","bi2","bi3"],
"australian_keywords": ["au1","au2","au3"],
"trending_topics": ["t1","t2","t3"],
"search_volume_estimate": "high",
"competition_level": "medium",
"monetisation_potential": "high"
}}"""
kw_response = model.generate_content(kw_prompt)
kw_data = json.loads(kw_response.text.strip())
# Step 2: Generate eBook Ideas
all_keywords = (
kw_data.get("primary_keywords", []) +
kw_data.get("long_tail_keywords", []) +
kw_data.get("australian_keywords", [])
)
ideas_prompt = f"""Create {count} high-converting eBook ideas for niche "{niche}".
Use these SEO keywords naturally in titles/subtitles: {", ".join(all_keywords[:10])}
Trending topics to reference: {", ".join(kw_data.get("trending_topics", []))}
Rules:
- Titles must include 1-2 primary keywords
- Subtitles should include long-tail or buyer intent keywords
- Target Australian audiences
- Price range $7-$27 AUD
- Word count 8,000-15,000 words
Return ONLY valid JSON:
{{
"ebook_ideas": [
{{
"rank": 1,
"title": "SEO-optimised title with keyword",
"subtitle": "Long-tail keyword subtitle for Australian audience",
"primary_keyword": "main keyword",
"secondary_keywords": ["k1","k2","k3"],
"target_audience": "specific audience",
"pain_point": "core problem solved",
"unique_angle": "differentiator from competitors",
"chapter_count": 8,
"estimated_words": 10000,
"recommended_price_aud": 17,
"competition_score": "low",
"demand_score": "high",
"seo_score": "high",
"estimated_monthly_searches": "500-1000",
"pipeline_ready_topic": "exact topic for pipeline"
}}
]
}}"""
ideas_response = model.generate_content(ideas_prompt)
ideas_data = json.loads(ideas_response.text.strip())
ideas = ideas_data.get("ebook_ideas", [])
last_ideas = ideas
# Format output
output_lines = []
output_lines.append(f"## 📚 eBook Market Research — {niche}\n")
output_lines.append(
f"**Search Volume:** {kw_data.get('search_volume_estimate','?').upper()} | "
f"**Competition:** {kw_data.get('competition_level','?').upper()} | "
f"**Monetisation:** {kw_data.get('monetisation_potential','?').upper()}\n"
)
output_lines.append(f"**🔑 Top Keywords:** {' • '.join(kw_data.get('primary_keywords',[])[:5])}\n")
output_lines.append(f"**📈 Trending:** {' • '.join(kw_data.get('trending_topics',[]))}\n")
output_lines.append("---\n")
for idea in ideas:
se = {"high": "🟢", "medium": "🟡", "low": "🔴"}
output_lines.append(f"### #{idea.get('rank','?')} {idea.get('title','')}")
output_lines.append(f"*{idea.get('subtitle','')}*\n")
output_lines.append(f"**🔑 Primary Keyword:** `{idea.get('primary_keyword','')}`")
output_lines.append(f"**🔍 Monthly Searches:** ~{idea.get('estimated_monthly_searches','?')}")
output_lines.append(
f"**💰 Price:** ${idea.get('recommended_price_aud','?')} AUD | "
f"**📄 Words:** ~{idea.get('estimated_words',0):,} | "
f"**Chapters:** {idea.get('chapter_count','?')}"
)
output_lines.append(
f"**Demand:** {se.get(idea.get('demand_score',''),'❓')} {idea.get('demand_score','?').upper()} | "
f"**SEO:** {se.get(idea.get('seo_score',''),'❓')} {idea.get('seo_score','?').upper()} | "
f"**Competition:** {se.get(idea.get('competition_score',''),'❓')} {idea.get('competition_score','?').upper()}"
)
output_lines.append(f"**🎯 Unique Angle:** {idea.get('unique_angle','')}")
output_lines.append(f"**👥 Audience:** {idea.get('target_audience','')}")
output_lines.append(f"**😩 Pain Point:** {idea.get('pain_point','')}")
output_lines.append(f"**🚀 Pipeline Topic:** `{idea.get('pipeline_ready_topic','')}`")
output_lines.append("\n---\n")
return (
"\n".join(output_lines),
json.dumps(kw_data, indent=2),
json.dumps(ideas, indent=2),
gr.update(visible=True)
)
except json.JSONDecodeError as e:
return f"❌ JSON parse error: {e}\n\nTry again — Gemini occasionally returns malformed JSON.", "{}", "[]", gr.update(visible=False)
except Exception as e:
msg = str(e)
if "429" in msg:
return "⚠️ Gemini quota exceeded. Wait a minute and try again, or use a different API key.", "{}", "[]", gr.update(visible=False)
return f"❌ Error: {msg}", "{}", "[]", gr.update(visible=False)
def send_to_pipeline(selected_rank: int, at_key: str):
"""Send the selected eBook idea to the Airtable Pipeline Queue."""
global last_ideas
key = at_key.strip() or AIRTABLE_API_KEY
if not key:
return "❌ Please enter your Airtable API key."
if not last_ideas:
return "❌ No ideas found. Generate ideas first."
# Find the selected idea (rank is 1-based)
idx = max(0, selected_rank - 1)
if idx >= len(last_ideas):
idx = 0
idea = last_ideas[idx]
topic = idea.get("pipeline_ready_topic") or f"{idea.get('title','')} — {idea.get('subtitle','')}"
payload = {
"fields": {
"Topic": topic,
"Title": idea.get("title", ""),
"Subtitle": idea.get("subtitle", ""),
"Primary Keyword": idea.get("primary_keyword", ""),
"Target Audience": idea.get("target_audience", ""),
"Pain Point": idea.get("pain_point", ""),
"Unique Angle": idea.get("unique_angle", ""),
"Recommended Price (AUD)": idea.get("recommended_price_aud", 17),
"Estimated Words": idea.get("estimated_words", 10000),
"Chapter Count": idea.get("chapter_count", 8),
"Demand Score": idea.get("demand_score", ""),
"SEO Score": idea.get("seo_score", ""),
"Competition Score": idea.get("competition_score", ""),
"Monthly Searches": idea.get("estimated_monthly_searches", ""),
"Status": "Queued",
"Queued At": datetime.utcnow().isoformat() + "Z",
"Source": "EbookAgent HF Space",
}
}
try:
url = f"https://api.airtable.com/v0/{AIRTABLE_BASE_ID}/{requests.utils.quote(AIRTABLE_TABLE)}"
response = requests.post(
url,
headers={
"Authorization": f"Bearer {key}",
"Content-Type": "application/json",
},
json=payload,
timeout=15,
)
if response.status_code in (200, 201):
data = response.json()
record_id = data.get("id", "unknown")
return (
f"✅ **Sent to Pipeline Queue!**\n\n"
f"**Topic:** {topic}\n"
f"**Airtable Record:** `{record_id}`\n"
f"**Status:** Queued\n\n"
f"Your eBook pipeline will pick this up on the next run. 🚀"
)
else:
err = response.json()
# If table doesn't exist, provide setup instructions
if "NOT_FOUND" in str(err) or response.status_code == 404:
return (
f"⚠️ **Pipeline Queue table not found in Airtable.**\n\n"
f"Create a table called **'Pipeline Queue'** in base `{AIRTABLE_BASE_ID}` with these fields:\n"
f"- Topic (Single line text)\n"
f"- Title, Subtitle, Primary Keyword, Target Audience (Single line text)\n"
f"- Pain Point, Unique Angle (Long text)\n"
f"- Recommended Price (AUD), Estimated Words, Chapter Count (Number)\n"
f"- Demand Score, SEO Score, Competition Score, Monthly Searches (Single line text)\n"
f"- Status (Single select: Queued, In Progress, Done)\n"
f"- Queued At (Date), Source (Single line text)\n\n"
f"Then click **Send to Pipeline** again."
)
return f"❌ Airtable error ({response.status_code}): {json.dumps(err, indent=2)}"
except Exception as e:
return f"❌ Request failed: {str(e)}"
# ── Gradio UI ──────────────────────────────────────────────────────────────────
with gr.Blocks(
title="Brett Apps — eBook Market Research Agent",
theme=gr.themes.Base(
primary_hue="purple",
secondary_hue="indigo",
neutral_hue="slate",
font=gr.themes.GoogleFont("Inter"),
),
css="""
.header { text-align: center; padding: 2rem 0 1rem; }
.header h1 { font-size: 2rem; font-weight: 800; color: #a855f7; }
.header p { color: #94a3b8; font-size: 0.95rem; }
.send-box { border: 1px solid #7c3aed; border-radius: 12px; padding: 1rem; background: #1e1b2e; }
footer { display: none !important; }
"""
) as demo:
with gr.Column(elem_classes="header"):
gr.HTML("""
📚 eBook Market Research Agent
Brett Apps · SEO Keyword-Injected eBook Idea Generator · brett@brettapps.com
""")
with gr.Row():
# ── Left panel ─────────────────────────────────────────────────────────
with gr.Column(scale=1):
api_key_input = gr.Textbox(
label="🔑 Gemini API Key",
placeholder="AIza... (or set GEMINI_API_KEY secret)",
type="password",
info="Get your key at aistudio.google.com"
)
niche_dropdown = gr.Dropdown(
choices=NICHES,
value="passive income",
label="📂 Select Niche",
allow_custom_value=True,
info="Choose a niche or type your own"
)
count_slider = gr.Slider(
minimum=1, maximum=10, value=5, step=1,
label="💡 Number of eBook Ideas"
)
run_btn = gr.Button("🔍 Generate eBook Ideas", variant="primary", size="lg")
gr.Markdown("---")
# ── Send to Pipeline panel ─────────────────────────────────────────
with gr.Column(visible=False, elem_classes="send-box") as pipeline_panel:
gr.Markdown("### 🚀 Send to Pipeline")
at_key_input = gr.Textbox(
label="Airtable API Key",
placeholder="patXXX... (or set AIRTABLE_API_KEY secret)",
type="password"
)
idea_rank = gr.Slider(
minimum=1, maximum=10, value=1, step=1,
label="Which idea to send? (by rank #)"
)
send_btn = gr.Button("📤 Send to Airtable Pipeline Queue", variant="secondary")
pipeline_status = gr.Markdown("")
gr.Markdown("""
---
**Brett Apps eBook Pipeline**
`Market Research → Outline → Writer → Design → Publish → Sales`
""")
# ── Right panel ────────────────────────────────────────────────────────
with gr.Column(scale=2):
output_md = gr.Markdown(
value="*Results will appear here after generation...*"
)
with gr.Accordion("📦 Raw JSON Output", open=False):
with gr.Row():
kw_json = gr.Code(label="Keyword Data", language="json", lines=15)
ideas_json = gr.Code(label="eBook Ideas", language="json", lines=15)
# ── Event handlers ─────────────────────────────────────────────────────────
run_btn.click(
fn=research_ebook_ideas,
inputs=[niche_dropdown, count_slider, api_key_input],
outputs=[output_md, kw_json, ideas_json, pipeline_panel],
show_progress=True
)
send_btn.click(
fn=send_to_pipeline,
inputs=[idea_rank, at_key_input],
outputs=[pipeline_status]
)
gr.Examples(
examples=[
["passive income", 5, ""],
["lawn care business", 3, ""],
["dropshipping Australia", 5, ""],
["AI tools for small business", 5, ""],
],
inputs=[niche_dropdown, count_slider, api_key_input],
label="Quick Examples"
)
if __name__ == "__main__":
demo.launch()