""" Gradio Blocks UI, mounted onto the FastAPI app in main.py. Design choice: the UI talks to the *same* FastAPI routes (auth, generate, brand-voice, history, billing) via an in-process ASGI HTTP client, rather than calling the underlying Python functions directly. This means every rule enforced in the routes (rate limits, tier gating, quota checks, validation) applies identically whether a request comes from the UI or from a future external API consumer - no duplicated / drifting business logic. """ import httpx import gradio as gr from app.config import DEMO_MODE, TIERS, Tier def _client(app) -> httpx.AsyncClient: transport = httpx.ASGITransport(app=app) return httpx.AsyncClient(transport=transport, base_url="http://internal") def build_ui(app) -> gr.Blocks: with gr.Blocks(title="Etsy Listing Optimizer") as demo: token_state = gr.State(value=None) email_state = gr.State(value=None) gr.Markdown("# ๐Ÿท๏ธ Etsy Listing Optimizer") if DEMO_MODE: gr.Markdown( "โš ๏ธ **Demo mode**: no real Supabase/Groq/Gemini/Stripe credentials are configured, " "so this is running on in-memory storage and a mock AI provider that returns " "already-valid sample output. Add real API keys (see `.env.example` / `DEPLOY.md`) " "to go live. Signup/login works with a local in-memory demo auth store โ€” data is " "lost on restart." ) with gr.Tab("Account"): gr.Markdown("### Sign up or log in") with gr.Row(): email_box = gr.Textbox(label="Email") password_box = gr.Textbox(label="Password", type="password") with gr.Row(): signup_btn = gr.Button("Sign up") login_btn = gr.Button("Log in") logout_btn = gr.Button("Log out") auth_status = gr.Markdown("Not logged in.") async def do_signup(email, password): async with _client(app) as client: resp = await client.post("/auth/signup", json={"email": email, "password": password}) if resp.status_code != 200: return None, None, f"โŒ {resp.json().get('detail', 'Sign up failed.')}" data = resp.json() return data["access_token"], data["email"], f"โœ… Signed up and logged in as {data['email']}." async def do_login(email, password): async with _client(app) as client: resp = await client.post("/auth/login", json={"email": email, "password": password}) if resp.status_code != 200: return None, None, f"โŒ {resp.json().get('detail', 'Login failed.')}" data = resp.json() return data["access_token"], data["email"], f"โœ… Logged in as {data['email']}." def do_logout(): return None, None, "Logged out." signup_btn.click(do_signup, [email_box, password_box], [token_state, email_state, auth_status]) login_btn.click(do_login, [email_box, password_box], [token_state, email_state, auth_status]) logout_btn.click(do_logout, None, [token_state, email_state, auth_status]) gr.Markdown("### Plans") plan_lines = [] for tier in [Tier.FREE, Tier.STARTER, Tier.PRO, Tier.BUSINESS]: c = TIERS[tier] variants = f"{c.title_variants} title variant(s)" gens = "unlimited generations" if c.generations_per_month == -1 else f"{c.generations_per_month} generations/mo" plan_lines.append(f"- **{c.display_name}** ({c.price_czk} CZK/mo): {gens}, {variants}") gr.Markdown("\n".join(plan_lines)) with gr.Row(): upgrade_tier = gr.Dropdown( choices=[Tier.STARTER.value, Tier.PRO.value, Tier.BUSINESS.value], label="Upgrade to", value=Tier.STARTER.value, ) upgrade_btn = gr.Button("Get checkout link") checkout_output = gr.Markdown() async def do_checkout(tier, token): if not token: return "Log in first." async with _client(app) as client: resp = await client.post( f"/billing/checkout/{tier}", headers={"Authorization": f"Bearer {token}"} ) if resp.status_code != 200: return f"โŒ {resp.json().get('detail', 'Checkout unavailable.')}" return f"[Click here to complete checkout]({resp.json()['checkout_url']})" upgrade_btn.click(do_checkout, [upgrade_tier, token_state], checkout_output) gr.Markdown( "### Your data\n" "Use the buttons below for GDPR data export / account deletion." ) with gr.Row(): export_btn = gr.Button("Export my data (JSON)") delete_btn = gr.Button("๐Ÿ—‘๏ธ Delete my account", variant="stop") account_status = gr.Markdown() async def do_export(token): if not token: return "Log in first." async with _client(app) as client: resp = await client.get("/account/export", headers={"Authorization": f"Bearer {token}"}) return f"```json\n{resp.text}\n```" async def do_delete(token): if not token: return None, None, "Log in first." async with _client(app) as client: resp = await client.delete("/account/delete", headers={"Authorization": f"Bearer {token}"}) if resp.status_code != 200: return token, None, f"โŒ {resp.json().get('detail', 'Delete failed.')}" return None, None, "โœ… Account and all data deleted." export_btn.click(do_export, token_state, account_status) delete_btn.click(do_delete, token_state, [token_state, email_state, account_status]) with gr.Tab("Generate listing"): gr.Markdown( "Describe your product in plain language. We'll generate SEO-optimized " "title(s), exactly 13 tags, and a description โ€” then validate everything " "programmatically (length limits, tag count, no duplicate words) before " "showing it to you." ) product_description = gr.Textbox( label="Product description", placeholder="Hand-poured soy candle in a reused wine bottle, lavender and vanilla scent...", lines=4, ) target_keywords = gr.Textbox(label="Target keywords (optional)", placeholder="e.g. eco friendly gift") generate_btn = gr.Button("Generate", variant="primary") quota_md = gr.Markdown() titles_out = gr.Textbox(label="Title variant(s)", lines=3, interactive=False) tags_out = gr.Textbox(label="Tags (13)", lines=3, interactive=False) description_out = gr.Textbox(label="Description", lines=6, interactive=False) category_out = gr.Textbox(label="Category/attribute hints (Pro/Business)", interactive=False) async def do_generate(description, keywords, token): if not token: return "Log in first (Account tab).", "", "", "", "" async with _client(app) as client: resp = await client.post( "/generate", json={"product_description": description, "target_keywords": keywords or None}, headers={"Authorization": f"Bearer {token}"}, ) if resp.status_code != 200: detail = resp.json().get("detail", "Generation failed.") return f"โŒ {detail}", "", "", "", "" data = resp.json() quota = ( f"Used {data['generations_used_this_month']} / " f"{data['generations_limit_this_month']} generations this month." ) return ( quota, "\n".join(f"{i+1}. {t}" for i, t in enumerate(data["titles"])), ", ".join(data["tags"]), data["description"], ", ".join(data["category_hints"]) if data["category_hints"] else "(none for your plan)", ) generate_btn.click( do_generate, [product_description, target_keywords, token_state], [quota_md, titles_out, tags_out, description_out, category_out], ) with gr.Tab("Brand voice"): gr.Markdown("Set a brand voice profile to automatically flavor every generation.") bv_name = gr.Textbox(label="Profile name", value="Default") bv_tone = gr.Textbox(label="Tone (e.g. playful, elegant, minimalist, or free text)") bv_audience = gr.Textbox(label="Target audience") bv_words = gr.Textbox(label="Favorite words/phrases to reuse") bv_save_btn = gr.Button("Save profile") bv_status = gr.Markdown() bv_list = gr.JSON(label="Your brand voice profiles") async def save_brand_voice(name, tone, audience, words, token): if not token: return "Log in first.", [] async with _client(app) as client: resp = await client.post( "/brand-voice", json={"name": name, "tone": tone, "target_audience": audience, "favorite_words": words}, headers={"Authorization": f"Bearer {token}"}, ) if resp.status_code != 200: return f"โŒ {resp.json().get('detail', 'Save failed.')}", [] listing = await client.get("/brand-voice", headers={"Authorization": f"Bearer {token}"}) return "โœ… Saved.", listing.json() bv_save_btn.click( save_brand_voice, [bv_name, bv_tone, bv_audience, bv_words, token_state], [bv_status, bv_list] ) with gr.Tab("History"): gr.Markdown("Past generations (visibility depends on your plan).") history_refresh_btn = gr.Button("Refresh history") history_json = gr.JSON() async def refresh_history(token): if not token: return {"error": "Log in first."} async with _client(app) as client: resp = await client.get("/history", headers={"Authorization": f"Bearer {token}"}) if resp.status_code != 200: return {"error": resp.json().get("detail", "History unavailable.")} return resp.json() history_refresh_btn.click(refresh_history, token_state, history_json) gr.Markdown( "[Privacy Policy](/legal/privacy) ยท [Terms of Service](/legal/terms) ยท " "[Refund Policy](/legal/refunds)" ) return demo