Spaces:
Runtime error
Runtime error
| """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(), | |
| ) | |