"""Gradio UI for GuestPostSuggester. Search is free/local (TF-IDF + a per-search availability check). Analyze crawls a site's blog and asks an LLM for content-gap suggestions, billed to the HF token entered in the UI. Gradio Blocks can't create components dynamically at runtime, so results are rendered into a fixed, pre-allocated pool of MAX_RESULT_SLOTS rows (each: info + Analyze button + an Accordion for that slot's analysis output), revealed 10 at a time. "More" reveals the next 10 without touching already-visible slots (append, not replace). `results_state` (the full ordered, live-filtered, OPR-ranked candidate list for the current search) is the single source of truth both for rendering and for resolving which domain occupies a given slot at Analyze-click time — no per-slot State needed. """ from __future__ import annotations import gradio as gr from pipeline import availability, catalog, config, crawler, openpagerank, search, suggest INTRO = """ # 🔎 GuestPostSuggester Find guest-post-accepting websites relevant to your topic, ranked by domain authority. Search is **free** (local text matching, no AI call) and checks every result for live reachability before showing it. Click **Analyze** on any result to crawl its blog and get ~5 AI-suggested guest-post topics that fit its niche but aren't already covered — that step calls an LLM billed to **your own** Hugging Face token. """ MAX_SLOTS = config.MAX_RESULT_SLOTS # --- startup: load catalog, fit search index, attach OpenPageRank scores (once) --- _SITES = catalog.load_catalog() _BY_DOMAIN = {s.domain: s for s in _SITES} _INDEX = search.CatalogSearchIndex(_SITES) _OPR_SCORES = openpagerank.load_cached_scores() for _s in _SITES: _s.page_rank = _OPR_SCORES.get(_s.domain, 0.0) def _result_dict(s) -> dict: return { "domain": s.domain, "title": s.title, "niche": s.niche, "page_rank": s.page_rank, "guest_posts_url": s.guest_posts_url, } def do_search(topic: str, progress=gr.Progress()): if not topic.strip(): raise gr.Error("Please enter a topic.") progress(0.1, desc="Searching catalog (TF-IDF)…") shortlist = _INDEX.shortlist(topic, k=config.SHORTLIST_SIZE) if not shortlist: raise gr.Error("No matching sites found for that topic. Try different wording.") progress(0.35, desc=f"Checking {len(shortlist)} domains for availability…") alive_map = availability.check_availability([s.domain for s in shortlist]) live = [s for s in shortlist if alive_map.get(s.domain)] if not live: raise gr.Error("Found matching sites, but none are currently reachable. Try a different topic.") progress(0.8, desc=f"Ranking {len(live)} live results by authority…") live.sort(key=lambda s: s.page_rank, reverse=True) results = [_result_dict(s) for s in live] revealed = min(config.RESULTS_PAGE_SIZE, len(results), MAX_SLOTS) progress(1.0, desc=f"Found {len(results)} live, relevant sites.") return (results, revealed, *_render(results, revealed, prior_revealed=None)) def do_more(results: list, revealed: int): new_revealed = min(revealed + config.RESULTS_PAGE_SIZE, len(results), MAX_SLOTS) return (new_revealed, *_render(results, new_revealed, prior_revealed=revealed)) def _render(results: list, revealed: int, prior_revealed: int | None): """Build gr.update(...) values for every slot component, in the same order they're declared in ALL_OUTPUTS. prior_revealed=None means a fresh search (reset every slot); an int means an append (leave slots below prior_revealed untouched).""" start_touch = 0 if prior_revealed is None else prior_revealed row_upd, dom_upd, btn_upd, acc_upd, acc_md_upd = [], [], [], [], [] for i in range(MAX_SLOTS): if i < start_touch: # Already rendered by a previous call (search or an earlier More click) — leave # untouched so in-progress/expanded Accordion state isn't clobbered. row_upd.append(gr.update()) dom_upd.append(gr.update()) btn_upd.append(gr.update()) acc_upd.append(gr.update()) acc_md_upd.append(gr.update()) elif i < revealed: r = results[i] niche_part = f" · {r['niche']}" if r["niche"] else "" body = ( f"**{r['domain']}**{niche_part}\n\n" f"{r['title']}\n\n" f"OpenPageRank: `{r['page_rank']:.2f}` · [Guest post page]({r['guest_posts_url']})" ) row_upd.append(gr.update(visible=True)) dom_upd.append(gr.update(value=body)) btn_upd.append(gr.update(visible=True, interactive=True)) acc_upd.append(gr.update(visible=True, open=False, label="Analysis")) acc_md_upd.append(gr.update(value="")) else: # Beyond what's revealed: force-hidden/cleared on a fresh search reset; on an # append these were already hidden and this is a harmless no-op re-assertion. row_upd.append(gr.update(visible=False)) dom_upd.append(gr.update(value="")) btn_upd.append(gr.update(visible=False)) acc_upd.append(gr.update(visible=False, open=False, label="Analysis")) acc_md_upd.append(gr.update(value="")) more_visible = revealed < len(results) return (*row_upd, *dom_upd, *btn_upd, *acc_upd, *acc_md_upd, gr.update(visible=more_visible)) def make_analyze_handler(slot_index: int): def handler(results: list, revealed: int, hf_token: str, topic: str, progress=gr.Progress()): # Gate on `revealed`, not just len(results): a real user can never click a slot's # Analyze button before _render() has revealed it (Gradio enforces visible/ # interactive client-side), but Gradio's event backend doesn't independently # re-validate that server-side, so this guard is what actually enforces it. if slot_index >= revealed or slot_index >= len(results): raise gr.Error("This slot is not currently populated.") r = results[slot_index] if not hf_token.strip(): raise gr.Error("Enter your Hugging Face token to run analysis (billed to you).") site = _BY_DOMAIN.get(r["domain"]) if site is None: raise gr.Error("Internal error: site not found in catalog.") try: progress(0.2, desc=f"Crawling {r['domain']} for existing post titles…") lib = crawler.get_title_library(site, hf_token) if not lib.titles: return ( gr.update(visible=True, open=True, label=f"Analysis — {r['domain']}"), gr.update( value="Could not discover this site's post titles via RSS, WordPress " "REST API, sitemap, keyword-based blog discovery, or AI-assisted page " "inspection. No suggestions available." ), ) progress(0.6, desc=f"Found {len(lib.titles)} existing titles (via {lib.tier_used}). " f"Generating content-gap suggestions…") suggestions = suggest.suggest_topics(site, topic, lib.titles, hf_token) except Exception as e: # noqa: BLE001 - surface a clean message, not a raw traceback return ( gr.update(visible=True, open=True, label=f"Analysis — {r['domain']} (error)"), gr.update(value=f"Analysis failed: {e}"), ) body = ( f"Found **{len(lib.titles)}** existing post titles (via `{lib.tier_used}`).\n\n" f"**Suggested guest-post topics:**\n\n" + "\n".join(f"- {s}" for s in suggestions) ) return ( gr.update(visible=True, open=True, label=f"Analysis — {r['domain']} ✓"), gr.update(value=body), ) return handler def build_ui() -> gr.Blocks: with gr.Blocks(title="GuestPostSuggester") as demo: gr.Markdown(INTRO) with gr.Row(): topic_box = gr.Textbox(label="Topic", placeholder="e.g. sustainable travel", scale=4) search_btn = gr.Button("Search", variant="primary", scale=1) hf_token_box = gr.Textbox( label="Hugging Face token (only needed for Analyze — billed to you)", type="password", placeholder="hf_...", ) results_state = gr.State([]) revealed_state = gr.State(0) slot_rows, slot_domain_md, slot_analyze_btn = [], [], [] slot_accordion, slot_accordion_md = [], [] with gr.Column(): for _ in range(MAX_SLOTS): with gr.Row(visible=False) as row: with gr.Column(scale=4): dom_md = gr.Markdown() with gr.Column(scale=1, min_width=100): analyze_btn = gr.Button("Analyze", size="sm") with gr.Accordion("Analysis", open=False, visible=False) as acc: acc_md = gr.Markdown() slot_rows.append(row) slot_domain_md.append(dom_md) slot_analyze_btn.append(analyze_btn) slot_accordion.append(acc) slot_accordion_md.append(acc_md) more_btn = gr.Button("More", visible=False) render_outputs = [ *slot_rows, *slot_domain_md, *slot_analyze_btn, *slot_accordion, *slot_accordion_md, more_btn, ] search_btn.click( fn=do_search, inputs=[topic_box], outputs=[results_state, revealed_state, *render_outputs], ) more_btn.click( fn=do_more, inputs=[results_state, revealed_state], outputs=[revealed_state, *render_outputs], ) for i in range(MAX_SLOTS): slot_analyze_btn[i].click( fn=make_analyze_handler(i), inputs=[results_state, revealed_state, hf_token_box, topic_box], outputs=[slot_accordion[i], slot_accordion_md[i]], ) gr.Markdown(f"Suggestion model: `{config.MODEL_SUGGEST}`.") return demo if __name__ == "__main__": # show_api was deprecated in gradio 6.13 and removed entirely by 6.19 — don't pass it. # theme moved from the Blocks constructor to launch() as of gradio 6.0. build_ui().queue().launch( server_name="0.0.0.0", server_port=7860, theme=gr.themes.Soft(), )