from __future__ import annotations import base64 import hashlib import hmac import html import io import json import os import re from dataclasses import asdict, dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional import httpx from dotenv import load_dotenv from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, JSONResponse from fastapi.staticfiles import StaticFiles from PIL import Image, ImageDraw, ImageFont from pydantic import BaseModel, Field try: from duckduckgo_search import DDGS except Exception: # pragma: no cover - optional dependency DDGS = None try: from openai import OpenAI except Exception: # pragma: no cover - optional dependency OpenAI = None try: from supabase import create_client except Exception: # pragma: no cover - optional dependency create_client = None load_dotenv() APP_NAME = "Fair Dinkum Publishing Studio" BASE_DIR = Path(__file__).resolve().parent DATA_DIR = BASE_DIR / "data" EXPORTS_DIR = BASE_DIR / "exports" LIBRARY_PATH = DATA_DIR / "library.json" MEMBERS_PATH = DATA_DIR / "members.json" DATA_DIR.mkdir(parents=True, exist_ok=True) EXPORTS_DIR.mkdir(parents=True, exist_ok=True) app = FastAPI(title=APP_NAME) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) app.mount("/exports", StaticFiles(directory=str(EXPORTS_DIR)), name="exports") @dataclass(frozen=True) class SeedBook: slug: str title: str subtitle: str description: str category: str audience: str length: str tags: tuple[str, ...] accent: tuple[str, str] featured: bool = False SEED_BOOKS: List[SeedBook] = [ SeedBook( slug="ebook-market-research-blueprint", title="Ebook Market Research Blueprint", subtitle="Find demand before you write", description="A practical research-first workbook for spotting profitable topics, reading public signals, and turning a vague idea into a marketable plan.", category="Research", audience="Independent publishers", length="58 pages", tags=("research", "demand", "strategy"), accent=("#0F766E", "#042F2E"), featured=True, ), SeedBook( slug="seo-keyword-stack", title="SEO Keyword Stack", subtitle="Build ideas people search for", description="A keyword-led system for creating ebook ideas, subtitles, and product page copy that matches reader intent.", category="SEO", audience="Creators and marketers", length="44 pages", tags=("keywords", "intent", "copy"), accent=("#B45309", "#451A03"), ), SeedBook( slug="ebook-build-playbook", title="Ebook Build Playbook", subtitle="Outline, draft, package, publish", description="An end-to-end writing workflow that turns research into a publishable ebook without over-engineering the process.", category="Writing", audience="Solo authors", length="72 pages", tags=("outline", "drafting", "publishing"), accent=("#1D4ED8", "#172554"), featured=True, ), SeedBook( slug="thumbnail-design-system", title="Thumbnail Design System", subtitle="Covers that hold up in a grid", description="A clean visual system for ebook covers, series art, and thumbnail previews that stay readable at small sizes.", category="Design", audience="Self publishers", length="39 pages", tags=("covers", "thumbnails", "brand"), accent=("#14532D", "#052E16"), ), SeedBook( slug="membership-offer-playbook", title="Membership Offer Playbook", subtitle="Subscription packaging for readers", description="A guide to turning a book catalog into recurring revenue with clear tiers, strong onboarding, and useful member benefits.", category="Membership", audience="Subscription businesses", length="51 pages", tags=("stripe", "membership", "pricing"), accent=("#7C2D12", "#431407"), ), SeedBook( slug="catalog-copy-that-converts", title="Catalog Copy That Converts", subtitle="Write blurbs people click", description="A copywriting system for ebook descriptions, benefit-led blurbs, and bookstore metadata that improves click-through.", category="Copywriting", audience="Publishing teams", length="46 pages", tags=("catalog", "blurb", "metadata"), accent=("#312E81", "#1E1B4B"), ), ] PALETTES = [ ("#0F766E", "#042F2E"), ("#B45309", "#451A03"), ("#1D4ED8", "#172554"), ("#14532D", "#052E16"), ("#7C2D12", "#431407"), ("#7C3AED", "#312E81"), ] OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-4o-mini") OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "").strip() SUPABASE_URL = os.getenv("SUPABASE_URL", "").strip() SUPABASE_SERVICE_KEY = ( os.getenv("SUPABASE_SECRET_KEY", "").strip() or os.getenv("SUPABASE_SERVICE_ROLE_KEY", "").strip() ) SUPABASE_PUBLISHABLE_KEY = os.getenv("SUPABASE_PUBLISHABLE_KEY", "").strip() SUPABASE_JWKS_URL = os.getenv("SUPABASE_JWKS_URL", "").strip() STRIPE_SECRET_KEY = os.getenv("STRIPE_SECRET_KEY", "").strip() STRIPE_PRICE_ID_MONTHLY = os.getenv("STRIPE_PRICE_ID_MONTHLY", "").strip() STRIPE_PRICE_ID_ANNUAL = os.getenv("STRIPE_PRICE_ID_ANNUAL", "").strip() STRIPE_WEBHOOK_SECRET = os.getenv("STRIPE_WEBHOOK_SECRET", "").strip() PUBLIC_BASE_URL = os.getenv("PUBLIC_BASE_URL", "").strip() class ResearchRequest(BaseModel): topic: str = Field(..., min_length=2, max_length=120) market: str = Field(default="United States", max_length=80) audience: str = Field(default="independent readers", max_length=120) tone: str = Field(default="practical", max_length=40) source_limit: int = Field(default=6, ge=3, le=12) class BuildRequest(BaseModel): topic: str = Field(..., min_length=2, max_length=120) market: str = Field(default="United States", max_length=80) audience: str = Field(default="independent readers", max_length=120) tone: str = Field(default="practical", max_length=40) chapter_count: int = Field(default=6, ge=4, le=8) idea_index: int = Field(default=0, ge=0, le=20) research: Optional[Dict[str, Any]] = None class CheckoutRequest(BaseModel): email: str = Field(..., min_length=5, max_length=180) name: Optional[str] = Field(default=None, max_length=120) plan: str = Field(default="monthly", max_length=20) public_base_url: Optional[str] = Field(default=None, max_length=200) def _now_iso() -> str: return datetime.now(timezone.utc).isoformat() def _load_json(path: Path, default: Any) -> Any: if not path.exists(): return default try: return json.loads(path.read_text(encoding="utf-8")) except Exception: return default def _save_json(path: Path, payload: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8") def _slugify(text: str) -> str: slug = re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-") return slug or "ebook" def _esc(value: Any) -> str: return html.escape("" if value is None else str(value)) def _font(size: int, bold: bool = False) -> ImageFont.ImageFont: candidates = [ "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf" if bold else "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", "/usr/share/fonts/truetype/liberation2/LiberationSans-Bold.ttf" if bold else "/usr/share/fonts/truetype/liberation2/LiberationSans-Regular.ttf", ] for candidate in candidates: if os.path.exists(candidate): return ImageFont.truetype(candidate, size=size) return ImageFont.load_default() def _wrap_lines(draw: ImageDraw.ImageDraw, text: str, font: ImageFont.ImageFont, max_width: int) -> List[str]: words = text.split() lines: List[str] = [] current = "" for word in words: trial = f"{current} {word}".strip() if not current or draw.textbbox((0, 0), trial, font=font)[2] <= max_width: current = trial else: lines.append(current) current = word if current: lines.append(current) return lines def _palette_for(text: str) -> tuple[str, str]: idx = abs(hash(text)) % len(PALETTES) return PALETTES[idx] def _base_url(request: Request) -> str: if PUBLIC_BASE_URL: return PUBLIC_BASE_URL.rstrip("/") return str(request.base_url).rstrip("/") def _status_snapshot() -> Dict[str, Any]: return { "openai_ready": bool(OPENAI_API_KEY and OpenAI is not None), "supabase_ready": bool(SUPABASE_URL and (SUPABASE_SERVICE_KEY or SUPABASE_PUBLISHABLE_KEY) and create_client is not None), "stripe_ready": bool(STRIPE_SECRET_KEY and STRIPE_PRICE_ID_MONTHLY), "library_count": len(_all_library_records()), "member_count": len(_load_json(MEMBERS_PATH, [])), "jwks_configured": bool(SUPABASE_JWKS_URL), } def _openai_client() -> Optional[Any]: if not OPENAI_API_KEY or OpenAI is None: return None return OpenAI(api_key=OPENAI_API_KEY) def _call_openai_json(system_prompt: str, user_prompt: str, *, temperature: float = 0.2) -> Optional[Dict[str, Any]]: client = _openai_client() if client is None: return None try: completion = client.chat.completions.create( model=OPENAI_MODEL, temperature=temperature, response_format={"type": "json_object"}, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], ) content = completion.choices[0].message.content or "{}" return json.loads(content) except Exception as exc: print(f"OpenAI JSON generation failed: {exc}") return None def _search_region_for_market(market: str) -> str: market_lower = market.lower() if "australia" in market_lower: return "au-en" if "united states" in market_lower or "usa" in market_lower or "us " in market_lower: return "us-en" if "united kingdom" in market_lower or "uk" in market_lower: return "uk-en" return "wt-wt" def _build_queries(topic: str, market: str) -> List[str]: return [ f"{topic} ebook market trends {market}", f"{topic} reader pain points {market}", f"best {topic} ebooks {market}", f"{topic} seo keywords", f"{topic} buying intent phrases", f"{topic} content gaps", ] def _collect_sources(topic: str, market: str, limit: int = 6) -> List[Dict[str, str]]: queries = _build_queries(topic, market) region = _search_region_for_market(market) collected: List[Dict[str, str]] = [] seen_urls: set[str] = set() if DDGS is not None: per_query = max(1, min(3, limit // max(1, len(queries)))) try: with DDGS() as ddgs: for query in queries: for result in ddgs.text( query, region=region, safesearch="moderate", max_results=per_query, ): item = { "query": query, "title": (result.get("title") or "").strip(), "url": result.get("href") or result.get("url") or "", "snippet": (result.get("body") or result.get("snippet") or "").strip(), } url = item["url"] if not url or url in seen_urls: continue seen_urls.add(url) collected.append(item) if len(collected) >= limit: return collected except Exception as exc: print(f"DuckDuckGo search failed: {exc}") if not collected: for index, query in enumerate(queries[:limit], start=1): collected.append( { "query": query, "title": f"{topic} market signal {index}", "url": "", "snippet": f"Fallback search signal for {topic} in {market}.", } ) return collected[:limit] def _fallback_keyword_clusters(topic: str) -> List[Dict[str, Any]]: core = topic.lower() return [ { "cluster": f"{topic} for beginners", "keywords": [f"{core} for beginners", f"how to start {core}", f"{core} guide"], "intent": "entry-level", "angle": "Start here and show the first win quickly.", }, { "cluster": f"best {topic} ideas", "keywords": [f"best {core}", f"{core} ideas", f"{core} examples"], "intent": "comparison", "angle": "Frame the book as the practical shortlist readers want.", }, { "cluster": f"{topic} checklist", "keywords": [f"{core} checklist", f"{core} template", f"{core} workbook"], "intent": "action", "angle": "Turn advice into a repeatable process.", }, { "cluster": f"advanced {topic}", "keywords": [f"advanced {core}", f"{core} strategy", f"{core} system"], "intent": "advanced", "angle": "Give the reader a second-level framework once the basics are clear.", }, ] def _fallback_book_ideas(topic: str, market: str, audience: str, research: Dict[str, Any]) -> List[Dict[str, Any]]: keywords = research.get("seo_clusters", []) keyword_pool: List[str] = [] for cluster in keywords: keyword_pool.extend(cluster.get("keywords", [])) if not keyword_pool: keyword_pool = [f"{topic} guide", f"{topic} for beginners", f"best {topic}"] chapter_plans = [ [ "Why the market wants this book", "How to position the promise", "Keyword research and reader intent", "Outline, chapters, and flow", "Writing the manuscript", "Publishing, pricing, and launch", ], [ "Reader pain points and demand", "SEO angles and long-tail queries", "Book structure and chapter map", "Drafting with clarity and pace", "Cover, product page, and metadata", "Membership, upsells, and retention", ], [ "Research signals that matter", "Choosing a title that sells", "Creating the chapter skeleton", "Writing practical content", "Packaging for the marketplace", "Launching the offer", ], ] titles = [ f"{topic}: The Practical Playbook", f"The {topic} Blueprint", f"{topic} Made Simple", f"The {topic} Growth System", f"{topic} SEO and Sales Guide", ] subtitles = [ f"SEO-ready ideas for {audience} in {market}", f"A clear path from research to launch in {market}", f"How to turn search intent into a publishable ebook", f"Build a product readers can find and buy", f"Create a book that fits a catalog and a subscription", ] ideas: List[Dict[str, Any]] = [] for index in range(5): ideas.append( { "title": titles[index], "subtitle": subtitles[index], "description": ( f"This angle helps {audience.lower()} turn {topic.lower()} into a focused, searchable, " f"and commercially useful ebook for {market}." ), "seo_keywords": [ keyword_pool[index % len(keyword_pool)], keyword_pool[(index + 1) % len(keyword_pool)], f"{topic} guide", f"{topic} ebook", f"best {topic.lower()}", ], "positioning": f"Useful for readers who want a practical {topic.lower()} outcome, not theory.", "why_it_works": "It matches search intent, gives the reader a clear promise, and converts naturally into product copy.", "chapter_plan": chapter_plans[index % len(chapter_plans)], "cover_direction": "High-contrast editorial cover with bold title, clear subtitle, and a premium band of color.", "price_point": "$14.99", } ) return ideas def _normalize_research(report: Dict[str, Any], *, topic: str, market: str, audience: str, sources: List[Dict[str, str]]) -> Dict[str, Any]: normalized = { "topic": topic, "market": market, "audience": audience, "executive_summary": report.get("executive_summary") or f"{topic} shows enough public search interest to justify a focused ebook for {audience} in {market}.", "demand_signals": report.get("demand_signals") or [ f"Readers ask for practical {topic.lower()} guidance.", f"Search intent favors step-by-step answers and examples.", f"Comparison and checklist queries are common entry points.", ], "reader_pain_points": report.get("reader_pain_points") or [ f"Readers are unsure how to start with {topic.lower()}.", f"They want a clearer path from idea to implementation.", f"They need concise, actionable steps instead of broad advice.", ], "seo_clusters": report.get("seo_clusters") or _fallback_keyword_clusters(topic), "book_ideas": report.get("book_ideas") or _fallback_book_ideas(topic, market, audience, report), "pricing_guidance": report.get("pricing_guidance") or "Position the ebook as a practical premium guide in the $12 to $19 range, then use membership for recurring value.", "launch_notes": report.get("launch_notes") or [ "Lead with a keyword-led title and subtitle.", "Use a cover that reads clearly in thumbnail size.", "Add a members-only upsell around the catalog.", ], "risks": report.get("risks") or [ "Overly broad positioning will blur the search intent.", "Weak cover readability will reduce click-through.", "Thin product pages will underperform even if the content is good.", ], "sources": sources, "generated_at": _now_iso(), } if len(normalized["book_ideas"]) < 5: normalized["book_ideas"] = (normalized["book_ideas"] + _fallback_book_ideas(topic, market, audience, normalized))[:5] return normalized def _research_from_openai(topic: str, market: str, audience: str, tone: str, sources: List[Dict[str, str]]) -> Optional[Dict[str, Any]]: user_prompt = { "topic": topic, "market": market, "audience": audience, "tone": tone, "sources": sources, } system_prompt = ( "You are a senior ebook market researcher and publishing strategist. " "Return only valid JSON with actionable findings, SEO keyword clusters, and five book ideas." ) prompt = ( "Return a strict JSON object with this structure:\n" "{\n" ' "executive_summary": "string",\n' ' "demand_signals": ["string"],\n' ' "reader_pain_points": ["string"],\n' ' "seo_clusters": [{"cluster":"string","keywords":["string"],"intent":"string","angle":"string"}],\n' ' "book_ideas": [{"title":"string","subtitle":"string","description":"string","seo_keywords":["string"],"positioning":"string","why_it_works":"string","chapter_plan":["string"],"cover_direction":"string","price_point":"string"}],\n' ' "pricing_guidance": "string",\n' ' "launch_notes": ["string"],\n' ' "risks": ["string"]\n' "}\n\n" f"Research brief:\n{json.dumps(user_prompt, indent=2)}" ) return _call_openai_json(system_prompt, prompt, temperature=0.2) def build_research_report(*, topic: str, market: str, audience: str, tone: str, source_limit: int) -> Dict[str, Any]: sources = _collect_sources(topic, market, limit=source_limit) report = None if OPENAI_API_KEY and OpenAI is not None: report = _research_from_openai(topic, market, audience, tone, sources) if report is None: report = { "executive_summary": ( f"{topic} has a workable search footprint in {market}. The strongest opportunity is to package the topic " f"as a clear, practical ebook for {audience}." ), "demand_signals": [ f"Searchers want direct answers about {topic.lower()}.", f"How-to and checklist queries are recurring signals.", f"Readers respond to narrower, outcome-led positioning.", ], "reader_pain_points": [ f"People struggle to turn {topic.lower()} into a first purchase.", f"They want a roadmap that removes setup friction.", f"They need better titles, covers, and metadata to choose from.", ], "seo_clusters": _fallback_keyword_clusters(topic), "book_ideas": _fallback_book_ideas(topic, market, audience, {}), "pricing_guidance": "Price the standalone ebook in the middle of the impulse-buy range, then use membership for ongoing access.", "launch_notes": [ "Make the promise specific.", "Keep the thumbnail readable.", "Bundle a membership upsell with the catalog.", ], "risks": [ "Generic positioning makes the book invisible.", "Thin metadata weakens discovery.", "A weak product page reduces conversion.", ], } normalized = _normalize_research(report, topic=topic, market=market, audience=audience, sources=sources) normalized["status"] = "complete" return normalized def _selected_idea_from_research(research: Dict[str, Any], idea_index: int) -> Dict[str, Any]: ideas = research.get("book_ideas") or _fallback_book_ideas( research["topic"], research["market"], research["audience"], research ) if not ideas: ideas = _fallback_book_ideas(research["topic"], research["market"], research["audience"], research) index = max(0, min(idea_index, len(ideas) - 1)) selected = dict(ideas[index]) selected["index"] = index selected["keyword_focus"] = selected.get("seo_keywords", [])[:3] return selected def _chapter_blueprint(selected_idea: Dict[str, Any], research: Dict[str, Any], chapter_count: int) -> List[Dict[str, Any]]: chapter_plan = list(selected_idea.get("chapter_plan") or []) if not chapter_plan: chapter_plan = [ f"Why {selected_idea['title']} matches the market", "Keyword positioning and reader intent", "Outline and manuscript structure", "Drafting the chapters", "Cover, metadata, and pricing", "Launch and membership growth", ] if len(chapter_plan) < chapter_count: chapter_plan.extend( f"Extended chapter {number}" for number in range(len(chapter_plan) + 1, chapter_count + 1) ) keyword_pool = selected_idea.get("seo_keywords") or ["ebook", "guide", "strategy"] demand = research.get("demand_signals") or [] pain_points = research.get("reader_pain_points") or [] chapters: List[Dict[str, Any]] = [] for number, title in enumerate(chapter_plan[:chapter_count], start=1): primary_keyword = keyword_pool[(number - 1) % len(keyword_pool)] signal = demand[(number - 1) % len(demand)] if demand else primary_keyword pain = pain_points[(number - 1) % len(pain_points)] if pain_points else selected_idea["subtitle"] sections = [ { "heading": "What the reader is trying to solve", "paragraphs": [ f"{title} should start by showing why {primary_keyword} matters to readers in {research['market']}.", f"For {research['audience']}, the book should be written around one visible outcome instead of broad theory.", ], }, { "heading": "How to make the chapter useful", "paragraphs": [ f"Turn the search signal '{signal}' into a concrete workflow and keep the steps short enough to act on immediately.", f"The reader pain point '{pain}' is a cue to add examples, checklists, and a simple decision path.", ], }, { "heading": "Where the SEO lives", "paragraphs": [ f"Use {', '.join(keyword_pool[:3])} naturally in the chapter heading, subheads, and product copy.", f"Keep the language plain and specific so the ebook reads well on a catalog page and in search results.", ], }, ] chapters.append( { "number": number, "title": title, "summary": f"Build a practical chapter around {primary_keyword} and the reader's main question.", "sections": sections, } ) return chapters def _chapter_to_markdown(chapter: Dict[str, Any]) -> str: lines = [f"# Chapter {chapter['number']}: {chapter['title']}", "", chapter["summary"], ""] for section in chapter["sections"]: lines.extend([f"## {section['heading']}", ""]) for paragraph in section["paragraphs"]: lines.extend([paragraph, ""]) return "\n".join(lines).strip() + "\n" def _render_paragraphs(paragraphs: List[str]) -> str: return "".join(f"

{_esc(paragraph)}

" for paragraph in paragraphs) def _render_bullets(items: List[str]) -> str: return "" def _render_cover_png(title: str, subtitle: str, category: str, accent: tuple[str, str]) -> str: width, height = 1200, 1600 img = Image.new("RGB", (width, height), accent[1]) draw = ImageDraw.Draw(img) draw.rectangle((0, 0, width, int(height * 0.42)), fill=accent[0]) draw.rounded_rectangle((56, 56, width - 56, height - 56), radius=48, outline=(255, 255, 255), width=4) draw.rounded_rectangle((90, 90, width - 90, height - 90), radius=30, outline=(255, 255, 255, 170), width=2) label_font = _font(40, bold=True) title_font = _font(84, bold=True) subtitle_font = _font(46, bold=False) meta_font = _font(34, bold=False) draw.text((112, 126), "FAIR DINKUM PUBLISHING", font=label_font, fill=(255, 255, 255)) draw.line((112, 218, width - 112, 218), fill=(255, 255, 255), width=2) title_lines = _wrap_lines(draw, title, title_font, width - 224) y = 440 for line in title_lines[:4]: draw.text((112, y), line, font=title_font, fill=(255, 255, 255)) y += 88 subtitle_lines = _wrap_lines(draw, subtitle, subtitle_font, width - 224) y = 860 for line in subtitle_lines[:4]: draw.text((112, y), line, font=subtitle_font, fill=(246, 248, 250)) y += 56 draw.text((112, 1390), category.upper(), font=meta_font, fill=(255, 255, 255)) buffer = io.BytesIO() img.save(buffer, format="PNG", optimize=True) return base64.b64encode(buffer.getvalue()).decode("ascii") def _render_ebook_html( *, title: str, subtitle: str, description: str, research: Dict[str, Any], selected_idea: Dict[str, Any], chapters: List[Dict[str, Any]], cover_b64: str, ) -> str: chapter_nav = "".join( f'
  • Chapter {chapter["number"]}: {_esc(chapter["title"])}
    • ' for chapter in chapters ) chapter_sections = [] for chapter in chapters: section_html = [] for section in chapter["sections"]: section_html.append( f"""

    {_esc(section['heading'])}

    {_render_paragraphs(section['paragraphs'])}
    """ ) chapter_sections.append( f"""
    Chapter {chapter['number']}

    {_esc(chapter['title'])}

    {_esc(chapter['summary'])}

    {''.join(section_html)}
    """ ) source_list = "".join( f"
  • {_esc(source['title'])}
    {_esc(source['snippet'])}
    • " for source in research.get("sources", []) ) keyword_tags = "".join(f"{_esc(keyword)}" for keyword in selected_idea.get("seo_keywords", [])) return f""" {_esc(title)}
    {_esc(title)} cover

    {_esc(title)}

    {_esc(subtitle)}

    {_esc(selected_idea.get('price_point', '$14.99'))} {_esc(selected_idea.get('positioning', 'Practical ebook'))} {_esc(research.get('market', ''))}
    {keyword_tags}

    Research summary

    {_esc(research.get('executive_summary', ''))}

    Demand signals

      {''.join(f'
    • {_esc(item)}
      • ' for item in research.get('demand_signals', []))}

    Reader pain points

      {''.join(f'
    • {_esc(item)}
      • ' for item in research.get('reader_pain_points', []))}
    {''.join(chapter_sections)}

    Sources

      {source_list}

    Next step

    {_esc(description)}

    Generated by Fair Dinkum Publishing Studio
    """ def _render_ebook_markdown( *, title: str, subtitle: str, research: Dict[str, Any], selected_idea: Dict[str, Any], chapters: List[Dict[str, Any]], ) -> str: parts = [f"# {title}", "", subtitle, "", "## Research summary", "", research.get("executive_summary", ""), ""] parts.append("### Demand signals") parts.extend(f"- {item}" for item in research.get("demand_signals", [])) parts.append("") parts.append("### Reader pain points") parts.extend(f"- {item}" for item in research.get("reader_pain_points", [])) parts.append("") for chapter in chapters: parts.append(f"## Chapter {chapter['number']}: {chapter['title']}") parts.append("") parts.append(chapter["summary"]) parts.append("") for section in chapter["sections"]: parts.append(f"### {section['heading']}") parts.append("") for paragraph in section["paragraphs"]: parts.append(paragraph) parts.append("") parts.append("## Sources") parts.extend(f"- {source['title']}: {source.get('snippet', '')}" for source in research.get("sources", [])) parts.append("") parts.append(f"## Positioning") parts.append(selected_idea.get("positioning", "")) return "\n".join(parts).strip() + "\n" def _supabase_client() -> Optional[Any]: if create_client is None: return None key = SUPABASE_SERVICE_KEY or SUPABASE_PUBLISHABLE_KEY if not SUPABASE_URL or not key: return None try: return create_client(SUPABASE_URL, key) except Exception as exc: print(f"Supabase client init failed: {exc}") return None def _supabase_insert(table: str, payload: Dict[str, Any]) -> None: client = _supabase_client() if client is None: return try: client.table(table).upsert(payload).execute() except Exception as exc: print(f"Supabase insert into {table} failed: {exc}") def _merge_records(records: List[Dict[str, Any]]) -> List[Dict[str, Any]]: merged: Dict[str, Dict[str, Any]] = {} for record in records: slug = record["slug"] merged[slug] = record return sorted(merged.values(), key=lambda item: item.get("created_at", ""), reverse=True) def _all_library_records() -> List[Dict[str, Any]]: records = [] for seed in SEED_BOOKS: records.append( { "slug": seed.slug, "title": seed.title, "subtitle": seed.subtitle, "description": seed.description, "category": seed.category, "audience": seed.audience, "length": seed.length, "tags": list(seed.tags), "featured": seed.featured, "kind": "seed", "cover_url": _seed_cover_url(seed), "html_url": "", "markdown_url": "", "report_url": "", "created_at": "", } ) local_records = _load_json(LIBRARY_PATH, []) if isinstance(local_records, list): records.extend(local_records) return _merge_records(records) def _seed_cover_url(seed: SeedBook) -> str: return f"data:image/png;base64,{_render_cover_png(seed.title, seed.subtitle, seed.category, seed.accent)}" def _render_library_cards(records: List[Dict[str, Any]]) -> str: cards = [] for record in records: badge_html = [] if record.get("featured"): badge_html.append('Featured') if record.get("kind") == "seed": badge_html.append('Showcase') else: badge_html.append('Generated') badge_html.append('Members') links = [] if record.get("html_url"): links.append(f'Read') if record.get("markdown_url"): links.append(f'Markdown') if record.get("report_url"): links.append(f'Report') cover = record.get("cover_url") or "" cards.append( f"""
    {_esc(record['title'])} cover
    {_esc(record.get('category', 'Ebook'))} {''.join(badge_html)}

    {_esc(record['title'])}

    {_esc(record['subtitle'])}

    {_esc(record['description'])}

    {_esc(record.get('audience', ''))} · {_esc(record.get('length', ''))}
    {''.join(links)}
    """ ) return "".join(cards) def _persist_library_record(record: Dict[str, Any]) -> None: existing = _load_json(LIBRARY_PATH, []) if not isinstance(existing, list): existing = [] existing = [item for item in existing if item.get("slug") != record["slug"]] existing.append(record) _save_json(LIBRARY_PATH, _merge_records(existing)) _supabase_insert("ebook_library", record) def _persist_member_record(record: Dict[str, Any]) -> None: existing = _load_json(MEMBERS_PATH, []) if not isinstance(existing, list): existing = [] existing = [item for item in existing if item.get("email") != record["email"]] existing.append(record) _save_json(MEMBERS_PATH, existing) _supabase_insert("ebook_members", record) def _build_book_package( *, topic: str, market: str, audience: str, tone: str, chapter_count: int, idea_index: int, research: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: if research is None: research = build_research_report( topic=topic, market=market, audience=audience, tone=tone, source_limit=6, ) selected_idea = _selected_idea_from_research(research, idea_index) chapters = _chapter_blueprint(selected_idea, research, chapter_count) slug = _slugify(selected_idea["title"]) cover_accent = _palette_for(selected_idea["title"]) cover_b64 = _render_cover_png( selected_idea["title"], selected_idea["subtitle"], selected_idea.get("price_point", "Ebook"), cover_accent, ) html_doc = _render_ebook_html( title=selected_idea["title"], subtitle=selected_idea["subtitle"], description=selected_idea["description"], research=research, selected_idea=selected_idea, chapters=chapters, cover_b64=cover_b64, ) markdown_doc = _render_ebook_markdown( title=selected_idea["title"], subtitle=selected_idea["subtitle"], research=research, selected_idea=selected_idea, chapters=chapters, ) export_dir = EXPORTS_DIR / slug export_dir.mkdir(parents=True, exist_ok=True) cover_path = export_dir / "cover.png" html_path = export_dir / "book.html" markdown_path = export_dir / "book.md" report_path = export_dir / "report.json" cover_path.write_bytes(base64.b64decode(cover_b64)) html_path.write_text(html_doc, encoding="utf-8") markdown_path.write_text(markdown_doc, encoding="utf-8") package = { "slug": slug, "title": selected_idea["title"], "subtitle": selected_idea["subtitle"], "description": selected_idea["description"], "market": market, "audience": audience, "tone": tone, "topic": topic, "cover_url": f"/exports/{slug}/cover.png", "html_url": f"/exports/{slug}/book.html", "markdown_url": f"/exports/{slug}/book.md", "report_url": f"/exports/{slug}/report.json", "created_at": _now_iso(), "featured": True, "kind": "generated", "length": f"{chapter_count} chapters", "tags": list(selected_idea.get("seo_keywords", [])[:5]), "category": "Generated Ebook", "source_idea_index": selected_idea["index"], "selected_idea": selected_idea, "research": research, "chapters": chapters, "book_html": html_doc, "book_markdown": markdown_doc, "cover_base64": cover_b64, } report_path.write_text(json.dumps({"research": research, "package": package}, indent=2, ensure_ascii=False), encoding="utf-8") record = { "slug": slug, "title": selected_idea["title"], "subtitle": selected_idea["subtitle"], "description": selected_idea["description"], "category": "Generated Ebook", "audience": audience, "length": f"{chapter_count} chapters", "tags": list(selected_idea.get("seo_keywords", [])[:5]), "featured": True, "kind": "generated", "cover_url": f"/exports/{slug}/cover.png", "html_url": f"/exports/{slug}/book.html", "markdown_url": f"/exports/{slug}/book.md", "report_url": f"/exports/{slug}/report.json", "created_at": package["created_at"], } _persist_library_record(record) package["library_record"] = record return package def _stripe_plan_price_id(plan: str) -> Optional[str]: if plan == "annual": return STRIPE_PRICE_ID_ANNUAL or None return STRIPE_PRICE_ID_MONTHLY or None def _stripe_configured() -> bool: return bool(STRIPE_SECRET_KEY and STRIPE_PRICE_ID_MONTHLY) def _build_checkout_session(request: Request, payload: CheckoutRequest) -> Dict[str, Any]: if not STRIPE_SECRET_KEY: raise HTTPException(status_code=501, detail="STRIPE_SECRET_KEY is not configured.") if "@" not in payload.email or payload.email.count("@") != 1: raise HTTPException(status_code=400, detail="A valid email address is required.") price_id = _stripe_plan_price_id(payload.plan) if not price_id: raise HTTPException(status_code=501, detail=f"Missing Stripe price ID for plan '{payload.plan}'.") origin = (payload.public_base_url or "").strip() or _base_url(request) success_url = f"{origin}/?success=1&session_id={{CHECKOUT_SESSION_ID}}" cancel_url = f"{origin}/?canceled=1" form = { "mode": "subscription", "success_url": success_url, "cancel_url": cancel_url, "customer_email": payload.email, "line_items[0][price]": price_id, "line_items[0][quantity]": "1", "metadata[plan]": payload.plan, "metadata[email]": payload.email, "metadata[name]": payload.name or "", } response = httpx.post( "https://api.stripe.com/v1/checkout/sessions", data=form, auth=(STRIPE_SECRET_KEY, ""), timeout=20.0, ) if response.status_code >= 400: raise HTTPException(status_code=response.status_code, detail=response.text) session = response.json() return { "checkout_url": session.get("url"), "session_id": session.get("id"), "message": "Checkout session created.", } def _stripe_session_status(session_id: str) -> Dict[str, Any]: if not STRIPE_SECRET_KEY: raise HTTPException(status_code=501, detail="STRIPE_SECRET_KEY is not configured.") response = httpx.get( f"https://api.stripe.com/v1/checkout/sessions/{session_id}", auth=(STRIPE_SECRET_KEY, ""), timeout=20.0, ) if response.status_code >= 400: raise HTTPException(status_code=response.status_code, detail=response.text) session = response.json() return { "session_id": session.get("id"), "status": session.get("status"), "payment_status": session.get("payment_status"), "customer_email": session.get("customer_email"), "mode": session.get("mode"), "metadata": session.get("metadata", {}), "subscription": session.get("subscription"), } def _parse_stripe_signature(header: str) -> tuple[str, List[str]]: parts: Dict[str, List[str]] = {} for chunk in header.split(","): if "=" in chunk: key, value = chunk.split("=", 1) parts.setdefault(key.strip(), []).append(value.strip()) if "t" not in parts or "v1" not in parts: raise ValueError("Invalid Stripe signature header.") return parts["t"][0], parts["v1"] def _verify_stripe_signature(secret: str, body: bytes, timestamp: str, signatures: List[str]) -> None: expected = hmac.new( secret.encode("utf-8"), msg=f"{timestamp}.".encode("utf-8") + body, digestmod=hashlib.sha256, ).hexdigest() if not any(hmac.compare_digest(expected, signature) for signature in signatures): raise ValueError("Invalid Stripe signature.") def _upsert_member_from_session(session: Dict[str, Any]) -> None: email = session.get("customer_email") or "" if not email: return record = { "email": email, "name": session.get("metadata", {}).get("name") or "", "plan": session.get("metadata", {}).get("plan") or "monthly", "customer_id": session.get("customer") or "", "subscription_id": session.get("subscription") or "", "session_id": session.get("id") or "", "status": session.get("payment_status") or session.get("status") or "active", "updated_at": _now_iso(), } existing = _load_json(MEMBERS_PATH, []) if not isinstance(existing, list): existing = [] existing = [item for item in existing if item.get("email") != email] existing.append(record) _save_json(MEMBERS_PATH, existing) _supabase_insert("ebook_members", record) @app.get("/", response_class=HTMLResponse) def home(request: Request) -> HTMLResponse: status = _status_snapshot() library_records = _all_library_records() library_html = _render_library_cards(library_records) status_json = json.dumps(status) return HTMLResponse( f""" {_esc(APP_NAME)}
    {_esc(APP_NAME)}

    Ebook Research, Build, and Subscription Studio

    Run market research, generate SEO-led ebook ideas, build a polished ebook package, and publish a showcase library with Stripe membership signup.

    -OpenAI
    -Supabase
    -Stripe

    Workspace

    {len(library_records)} books {status["member_count"]} members
    Idle
    Preview cover
    Preview Ready

    {_esc(SEED_BOOKS[0].title)}

    {_esc(SEED_BOOKS[0].subtitle)}

    {_esc(SEED_BOOKS[0].description)}

    HTML Markdown Report

    Chapters

    No ebook built yet.

    Research

    Summary

    Run a research pass to fill this section.

    Demand

      Pain points

        Keyword clusters

        Sources

          Ideas

          Select one idea, then build the ebook.

          Showcase

          Search the library and open the built ebook files.
          {library_html}

          Membership

          Stripe subscription checkout for readers and members.
          Checkout ready when Stripe keys are configured.
          OpenAI: optional Supabase: optional persistence Stripe: subscriptions
          """ ) @app.get("/api/health") def api_health() -> Dict[str, Any]: return {"status": "ok", **_status_snapshot()} @app.get("/api/library") def api_library() -> Dict[str, Any]: return {"books": _all_library_records()} @app.post("/api/research") def api_research(payload: ResearchRequest) -> Dict[str, Any]: report = build_research_report( topic=payload.topic, market=payload.market, audience=payload.audience, tone=payload.tone, source_limit=payload.source_limit, ) return report @app.post("/api/build") def api_build(payload: BuildRequest) -> Dict[str, Any]: package = _build_book_package( topic=payload.topic, market=payload.market, audience=payload.audience, tone=payload.tone, chapter_count=payload.chapter_count, idea_index=payload.idea_index, research=payload.research, ) return { "slug": package["slug"], "title": package["title"], "subtitle": package["subtitle"], "description": package["description"], "cover_url": package["cover_url"], "html_url": package["html_url"], "markdown_url": package["markdown_url"], "report_url": package["report_url"], "created_at": package["created_at"], "selected_idea": package["selected_idea"], "research": package["research"], "chapters": package["chapters"], "tags": package["tags"], } @app.get("/api/books/{slug}") def api_book(slug: str) -> Dict[str, Any]: records = _all_library_records() for record in records: if record["slug"] == slug: return record raise HTTPException(status_code=404, detail="Book not found.") @app.post("/api/checkout/session") def api_checkout_session(request: Request, payload: CheckoutRequest) -> Dict[str, Any]: return _build_checkout_session(request, payload) @app.get("/api/checkout/session-status/{session_id}") def api_checkout_session_status(session_id: str) -> Dict[str, Any]: return _stripe_session_status(session_id) @app.post("/api/webhooks/stripe") async def api_stripe_webhook(request: Request) -> JSONResponse: raw_body = await request.body() if STRIPE_WEBHOOK_SECRET: signature = request.headers.get("Stripe-Signature", "") try: timestamp, signatures = _parse_stripe_signature(signature) _verify_stripe_signature(STRIPE_WEBHOOK_SECRET, raw_body, timestamp, signatures) except Exception as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc event = await request.json() event_type = event.get("type") if event_type == "checkout.session.completed": session = event.get("data", {}).get("object", {}) _upsert_member_from_session(session) return JSONResponse({"received": True}) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=int(os.getenv("PORT", "7860")))