# ui.py import gradio as gr from PIL import Image import json import logging import traceback from orchestrator import ORCH import utils logger = logging.getLogger("newsorchestra.ui") title = "NewsOrchestra — Multi-agent AI Orchestrator for Explainable Fact-checking" description = """Paste article text or a URL and optionally upload an image. This demo orchestrates multiple models (zero-shot classifier, SerpApi evidence fetch, optional Gemini planner/synthesizer) and returns a verdict and human-friendly explanation.""" def build_md_from_merged(merged: dict) -> str: md_lines = [] md_lines.append(f"### Claim\n\n> {merged.get('claim')}\n") md_lines.append(f"**Final verdict:** **{merged.get('final_verdict', '—')}** ") conf = merged.get('confidence') md_lines.append(f"**Confidence:** {(conf*100):.1f}% " if conf is not None else "**Confidence:** — ") md_lines.append("\n---\n") ea = merged.get("evidence_agg", {}).get("evidence", []) if ea: md_lines.append("### Top evidence (click to open)\n") for e in ea[:6]: title = e.get('title') or e.get('snippet') or "Result" link_display = "🔗" if e.get('link') else "" link_url = e.get('link') or "#" md_lines.append(f"- **[{e.get('domain')}]** [{link_display}] [{title}]({link_url}) — trust **{e.get('trust')}** \n > {e.get('snippet')}\n") else: md_lines.append("- No top search results found.\n") md_lines.append("\n---\n") img_chk = merged.get("image_check") if img_chk: md_lines.append("### Attached media summary\n") if img_chk.get("caption"): md_lines.append(f"- Caption (auto): {img_chk.get('caption')} \n") if img_chk.get("ocr_text"): md_lines.append(f"- OCR text found: `{img_chk.get('ocr_text')}` \n") if img_chk.get("exif"): md_lines.append(f"- EXIF metadata keys: {', '.join(list(img_chk.get('exif').keys())[:6])} \n") df = img_chk.get("deepfake_check") if df and isinstance(df, list) and len(df) > 0: top = df[0] lab = top.get("label") sc = top.get("score") md_lines.append(f"- **Deepfake detector:** top label **{lab}** (score {sc:.2f}) \n") md_lines.append("\n---\n") if merged.get("explanation"): md_lines.append("### Why we reached this verdict\n") md_lines.append(merged.get("explanation") + "\n") if merged.get("planner"): md_lines.append("\n---\n**Orchestrator planner rationale:**\n") md_lines.append("```\n" + json.dumps(merged.get("planner"), indent=2) + "\n```\n") if merged.get("synthesizer"): md_lines.append("\n---\n**Synthesizer output (Gemini):**\n") md_lines.append("```\n" + json.dumps(merged.get("synthesizer"), indent=2) + "\n```\n") return "\n".join(md_lines) def create_ui(): with gr.Blocks(title=title) as demo: gr.Markdown(f"# {title}\n\n{description}") with gr.Row(): inp = gr.Textbox(lines=8, label="Article text or URL", placeholder="Paste full article text or a URL (http...)") with gr.Row(): img_upload = gr.Image(type="pil", label="Upload image (optional)") img_url = gr.Textbox(lines=1, label="Or image URL (optional)", placeholder="https://...") with gr.Row(): run_gemini_cb = gr.Checkbox(label="Run Gemini planner + synthesizer (if configured)", value=True) run_serp_cb = gr.Checkbox(label="Run SerpApi evidence fetch (recommended)", value=bool(False)) run_deepfake_cb = gr.Checkbox(label="Run deepfake detection (Hugging Face model)", value=True) with gr.Row(): analyze_btn = gr.Button("Analyze (run orchestration)") with gr.Row(): merged_out = gr.JSON(label="Merged verification report (JSON)") with gr.Row(): human_md = gr.Markdown(label="Friendly explanation & evidence") def on_click_analyze(text_or_url: str = "", image_upload=None, image_url: str = "", run_gemini: bool = True, run_serpapi: bool = True, run_deepfake: bool = True): try: raw_input = (text_or_url or "").strip() def is_url_only(s: str) -> bool: if not s: return False s = s.strip() return bool(__import__("re").match(r"^https?://\S+$", s)) mode_is_url = is_url_only(raw_input) article_text = None url = None claim = None if mode_is_url: url = raw_input # delegate fetch to orchestrator.fetch_article_text_from_url if available article_text = ORCH.fetch_article_text_from_url(url) if hasattr(ORCH, "fetch_article_text_from_url") else None claim = (article_text and utils.sanitize_text(article_text).split("\n",1)[0]) or "" else: article_text = raw_input or None if article_text: claim = utils.sanitize_text(article_text) else: # derive claim from image if possible img_obj = None if image_upload: if isinstance(image_upload, Image.Image): img_obj = image_upload else: img_obj = image_upload if img_obj is None and image_url: try: img_obj, b, err = ORCH.fetch_image_bytes(image_url) if hasattr(ORCH, "fetch_image_bytes") else (None, None, None) except Exception: img_obj = None if img_obj: ocr = None # Try a model-level OCR (pytesseract used in original) — omitted here for brevity cap = models.hf_image_caption(img_obj) claim = (ocr or cap or "No claim text provided") or "No claim text provided" else: claim = "No claim text provided" # pass image_upload as PIL if it is already PIL image_to_pass = image_upload if isinstance(image_upload, Image.Image) else None merged = ORCH.run( claim_text=claim, article_text=article_text, url=url, image_upload=image_to_pass, image_url=image_url, run_gemini=run_gemini, run_serpapi=run_serpapi, run_deepfake=run_deepfake ) md = build_md_from_merged(merged) return merged, md except Exception as e: tb = traceback.format_exc() logger.exception("Error in analyze handler: %s", e) return {"error": str(e), "trace": tb}, f"Error running analysis: {e}" analyze_btn.click(on_click_analyze, inputs=[inp, img_upload, img_url, run_gemini_cb, run_serp_cb, run_deepfake_cb], outputs=[merged_out, human_md]) gr.Markdown( "- Notes: set SERPAPI_KEY and GEMINI_API_KEY in environment for best results.\n" "- Set HF_DEEPFAKE_MODEL to choose a different deepfake detection model.\n" "- The demo is for research/educational purposes. Always human-review outputs before acting on them." ) return demo