"""
app.py — BERTopic Agentic Thematic Analysis — Gradio UI
THEMIS: Thematic Engine for Mining and Identifying Scholarly Topics
Implements Braun & Clarke (2006) 6-Phase Framework
"""
import gradio as gr
import json
import os
import pandas as pd
from agent import invoke_agent, reset_agent
CHECKPOINT_DIR = "checkpoints"
os.makedirs(CHECKPOINT_DIR, exist_ok=True)
# ── Phase Checkpoint Detection ─────────────────────────────────────────────────
def get_phase_status():
files = {
"summaries.json": False,
"labels.json": False,
"themes.json": False,
"taxonomy_map.json": False,
"comparison.csv": False,
"narrative.txt": False,
}
for fname in files:
files[fname] = os.path.exists(os.path.join(CHECKPOINT_DIR, fname))
return files
def render_phase_bar():
status = get_phase_status()
phases = [
("①", "Load", status["summaries.json"]),
("②", "Codes", status["labels.json"]),
("③", "Themes", status["themes.json"]),
("④⑤", "Review & Name", status["themes.json"]),
("⑤½", "PAJAIS Map", status["taxonomy_map.json"]),
("⑥", "Report", status["comparison.csv"] and status["narrative.txt"]),
]
items = ""
for num, name, done in phases:
icon = "✅" if done else "⬜"
cls = "phase-done" if done else "phase-pending"
items += f'{num}{icon} {name}
'
return f'{items}
'
# ── Review Table Loader ────────────────────────────────────────────────────────
def load_review_table():
"""Load highest-priority checkpoint for the review table."""
checkpoint_priority = [
("taxonomy_map.json", "themes"),
("themes.json", "themes"),
("labels.json", "topics"),
("summaries.json", "topics"),
]
for fname, key in checkpoint_priority:
fpath = os.path.join(CHECKPOINT_DIR, fname)
if os.path.exists(fpath):
with open(fpath) as f:
data = json.load(f)
items = data.get(key, [])
if not items:
continue
rows = []
for item in items:
is_taxonomy = fname == "taxonomy_map.json"
top_evidence = item.get("top_sentences", [""])[0] if item.get("top_sentences") else ""
rows.append({
"ID": item.get("topic_id", item.get("theme_name", "")),
"Label / Theme": item.get("label") or item.get("theme_name", ""),
"Category": item.get("category") or item.get("pajais_match", ""),
"Sentences": item.get("count") or item.get("sentence_count", 0),
"Confidence": round(float(item.get("confidence") or item.get("match_confidence", 0)), 2),
"Top Evidence": top_evidence[:120] + "..." if len(top_evidence) > 120 else top_evidence,
"Approve": item.get("approve", ""),
"Rename To": item.get("rename_to", ""),
"Reasoning": item.get("user_reasoning", ""),
})
return pd.DataFrame(rows)
return pd.DataFrame(columns=["ID", "Label / Theme", "Category", "Sentences",
"Confidence", "Top Evidence", "Approve", "Rename To", "Reasoning"])
# ── Charts Loader ──────────────────────────────────────────────────────────────
def get_chart_options():
options = []
for run_key in ["abstract", "title"]:
fpath = os.path.join(CHECKPOINT_DIR, f"{run_key}_charts.json")
if os.path.exists(fpath):
options.extend([
f"{run_key.title()} — Intertopic Map",
f"{run_key.title()} — Topic Sizes",
f"{run_key.title()} — Similarity Heatmap",
f"{run_key.title()} — Size Distribution",
])
return options if options else ["No charts yet — run analysis first"]
def load_chart(selection: str) -> str:
if not selection or "No charts" in selection:
return "Run analysis to generate charts
"
parts = selection.lower().split(" — ")
if len(parts) < 2:
return ""
run_key = parts[0].strip()
chart_key_map = {
"intertopic map": "intertopic",
"topic sizes": "bars",
"similarity heatmap": "heatmap",
"size distribution": "distribution",
}
chart_key = chart_key_map.get(parts[1].strip(), "intertopic")
fpath = os.path.join(CHECKPOINT_DIR, f"{run_key}_charts.json")
if not os.path.exists(fpath):
return f"No charts for {run_key} run yet.
"
with open(fpath) as f:
charts = json.load(f)
html = charts.get(chart_key, "Chart not found
")
return f'{html}
'
# ── Download Links ─────────────────────────────────────────────────────────────
def render_download_links():
downloads = [
("summaries.json", "① Load Summary", "phase-1"),
("labels.json", "② Topic Labels", "phase-2"),
("themes.json", "③ Consolidated Themes", "phase-3"),
("taxonomy_map.json", "⑤½ PAJAIS Taxonomy Map", "phase-5"),
("comparison.csv", "⑥ Abstract vs Title Comparison", "phase-6"),
("narrative.txt", "⑥ Section 7 Narrative", "phase-6"),
]
html = ''
for fname, label, phase_cls in downloads:
fpath = os.path.join(CHECKPOINT_DIR, fname)
if os.path.exists(fpath):
size = os.path.getsize(fpath)
size_str = f"{size/1024:.1f} KB" if size > 1024 else f"{size} B"
html += f'''
'''
else:
html += f'''
⏳
{label}
{fname} · not yet generated
Pending
'''
html += '
Required CSV columns:
· Authors
· Title
· Abstract
· Author Keywords
· Cited by
· Source title
· Year
""")
# ── Section ②: Agent Chat ──────────────────────────────────────────────────
with gr.Group(elem_classes=["section-block"]):
gr.HTML(section_header("②", "THEMIS Agent", "6-phase Braun & Clarke analysis pipeline"))
with gr.Column(elem_classes=["chat-wrap"]):
chatbot = gr.Chatbot(
label="",
height=420,
show_label=False,
elem_classes=["gradio-chatbot"],
)
with gr.Row():
msg_input = gr.Textbox(
placeholder="Type a message... (e.g. 'run abstract', 'run title', or ask a question)",
show_label=False,
scale=5,
elem_classes=["msg-input"],
container=False,
)
send_btn = gr.Button("Send ↗", scale=1, elem_classes=["btn-send"])
with gr.Row():
reset_btn = gr.Button("⟳ Reset Session", elem_classes=["btn-reset"], scale=1)
gr.HTML('')
# ── Section ③: Results ─────────────────────────────────────────────────────
with gr.Group(elem_classes=["section-block"]):
gr.HTML(section_header("③", "Results", "Review table · Charts · Downloads"))
with gr.Tabs():
# Tab A: Review Table
with gr.Tab("📋 Review Table"):
review_table = gr.Dataframe(
value=load_review_table(),
interactive=True,
wrap=False,
height=400,
elem_classes=["gradio-dataframe"],
column_widths=["60px","160px","120px","80px","80px","200px","80px","120px","160px"],
)
with gr.Row():
refresh_table_btn = gr.Button("↻ Refresh Table", elem_classes=["btn-reset"], scale=1)
submit_review_btn = gr.Button("✓ Submit Review →", elem_classes=["btn-review"], scale=2)
gr.HTML('')
# Tab B: Charts
with gr.Tab("📊 Charts"):
with gr.Row():
chart_dropdown = gr.Dropdown(
choices=get_chart_options(),
label="Select Chart",
value=None,
scale=2,
)
refresh_charts_btn = gr.Button("↻ Refresh", elem_classes=["btn-reset"], scale=1)
chart_output = gr.HTML(
''
'Run analysis to generate interactive charts
'
)
# Tab C: Downloads
with gr.Tab("⬇ Downloads"):
download_html = gr.HTML(render_download_links())
refresh_dl_btn = gr.Button("↻ Refresh Downloads", elem_classes=["btn-reset"])
# ── Event Handlers ──────────────────────────────────────────────────────────
# Send message
def handle_send(message: str, history: list):
if not message.strip():
return history, "", load_review_table(), render_phase_bar()
response = invoke_agent(message, history)
history = history + [[message, response]]
return history, "", load_review_table(), render_phase_bar()
send_btn.click(
fn=handle_send,
inputs=[msg_input, chatbot],
outputs=[chatbot, msg_input, review_table, phase_bar],
)
msg_input.submit(
fn=handle_send,
inputs=[msg_input, chatbot],
outputs=[chatbot, msg_input, review_table, phase_bar],
)
# CSV upload — auto-trigger agent
def handle_upload(file, history: list):
if file is None:
return history, load_review_table(), render_phase_bar()
response = invoke_agent(f"Analyze my Scopus CSV: {file.name}", history)
history = history + [[f"📂 Uploaded CSV: {os.path.basename(file.name)}", response]]
return history, load_review_table(), render_phase_bar()
file_input.change(
fn=handle_upload,
inputs=[file_input, chatbot],
outputs=[chatbot, review_table, phase_bar],
)
# Submit review
def handle_submit_review(table_data, history):
new_history, _ = submit_review(table_data, history)
return new_history, load_review_table(), render_phase_bar()
submit_review_btn.click(
fn=handle_submit_review,
inputs=[review_table, chatbot],
outputs=[chatbot, review_table, phase_bar],
)
# Refresh table
refresh_table_btn.click(
fn=lambda: (load_review_table(), render_phase_bar()),
outputs=[review_table, phase_bar],
)
# Chart dropdown
chart_dropdown.change(
fn=load_chart,
inputs=[chart_dropdown],
outputs=[chart_output],
)
# Refresh charts
def refresh_charts():
opts = get_chart_options()
return gr.update(choices=opts, value=opts[0] if opts else None)
refresh_charts_btn.click(fn=refresh_charts, outputs=[chart_dropdown])
# Downloads refresh
refresh_dl_btn.click(fn=render_download_links, outputs=[download_html])
# Reset
def handle_reset():
msg = reset_agent()
return [], render_phase_bar()
reset_btn.click(
fn=handle_reset,
outputs=[chatbot, phase_bar],
)
# ── Launch ──────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
demo.launch(
server_name="0.0.0.0",
server_port=7860,
show_error=True,
)