Spaces:
Running
Running
# -*- coding: utf-8 -*- | |
import gradio as gr | |
import json | |
from Data_Fetching_and_Rendering import fetch_and_render_dashboard | |
from analytics_fetch_and_rendering import fetch_and_render_analytics | |
from mentions_dashboard import generate_mentions_dashboard | |
# Shared state | |
token_received = {"status": False, "token": None, "client_id": None} | |
# --- Handlers for token reception and status --- | |
def receive_token(accessToken: str, client_id: str): | |
""" | |
Called by a hidden POST mechanism to supply the OAuth code/token and client ID. | |
""" | |
try: | |
token_dict = json.loads(accessToken.replace("'", '"')) | |
except json.JSONDecodeError as e: | |
return { | |
"status": "β Invalid token format", | |
"token": "", | |
"client_id": client_id | |
} | |
token_received["status"] = True | |
token_received["token"] = token_dict | |
token_received["client_id"] = client_id | |
return { | |
"status": "β Token received", | |
"token": token_dict.get("access_token", ""), | |
"client_id": client_id | |
} | |
def check_status(): | |
return "β Token received" if token_received["status"] else "β Waiting for tokenβ¦" | |
def show_token(): | |
return token_received["token"].get("access_token", "") if token_received["status"] else "" | |
def show_client(): | |
return token_received["client_id"] or "" if token_received["status"] else "" | |
# --- Guarded fetch functions --- | |
def guarded_fetch_dashboard(): | |
if not token_received["status"]: | |
return "<p style='color:red; text-align:center;'>β Access denied. No token available. Please send token first.</p>" | |
# token_received["client_id"] and token_received["token"] required by fetch function | |
html = fetch_and_render_dashboard( | |
token_received["client_id"], | |
token_received["token"] | |
) | |
return html | |
def guarded_fetch_analytics(): | |
if not token_received["status"]: | |
return ( | |
"<p style='color:red; text-align:center;'>β Access denied. No token available.</p>", | |
None, | |
None | |
) | |
count_md, plot, growth_plot, avg_post_eng_rate, interaction_metrics = fetch_and_render_analytics( | |
token_received["client_id"], | |
token_received["token"] | |
) | |
return count_md, plot, growth_plot, avg_post_eng_rate, interaction_metrics | |
def run_mentions_and_load(): | |
html, fig = generate_mentions_dashboard( | |
token_received["client_id"], | |
token_received["token"] | |
) | |
return html, fig | |
# --- Build the Gradio UI --- | |
with gr.Blocks(theme=gr.themes.Soft(primary_hue="blue", secondary_hue="sky"), | |
title="LinkedIn Post Viewer & Analytics") as app: | |
gr.Markdown("# π LinkedIn Organization Post Viewer & Analytics") | |
gr.Markdown("Send your OAuth token via API call, then explore dashboard and analytics.") | |
# Hidden elements: simulate POST endpoint | |
hidden_token = gr.Textbox(visible=False, elem_id="hidden_token") | |
hidden_client = gr.Textbox(visible=False, elem_id="hidden_client_id") | |
hidden_btn = gr.Button(visible=False, elem_id="hidden_btn") | |
status_box = gr.Textbox(value=check_status(), label="Status", interactive=False) | |
token_display = gr.Textbox(value=show_token(), label="Access Token", interactive=False) | |
client_display = gr.Textbox(value=show_client(), label="Client ID", interactive=False) | |
# Wire hidden POST handler | |
hidden_btn.click( | |
fn=receive_token, | |
inputs=[hidden_token, hidden_client], | |
outputs=[status_box, token_display, client_display] | |
) | |
# Polling timer to update status and displays | |
timer = gr.Timer(1.0) | |
timer.tick(fn=check_status, outputs=status_box) | |
timer.tick(fn=show_token, outputs=token_display) | |
timer.tick(fn=show_client, outputs=client_display) | |
# Tabs for functionality | |
with gr.Tabs(): | |
with gr.TabItem("1οΈβ£ Dashboard"): | |
gr.Markdown("View your organization's recent posts and their engagement statistics.") | |
fetch_dashboard_btn = gr.Button("π Fetch Posts & Stats", variant="primary") | |
dashboard_html = gr.HTML(value="<p style='text-align: center; color: #555;'>Waiting for token...</p>") | |
fetch_dashboard_btn.click( | |
fn=guarded_fetch_dashboard, | |
inputs=[], | |
outputs=[dashboard_html] | |
) | |
with gr.TabItem("2οΈβ£ Analytics"): | |
gr.Markdown("View follower count and monthly gains for your organization.") | |
fetch_analytics_btn = gr.Button("π Fetch Follower Analytics", variant="primary") | |
follower_count = gr.Markdown("<p style='text-align: center; color: #555;'>Waiting for token...</p>") | |
with gr.Row(): # Use Row to align the two plots side-by-side | |
follower_plot = gr.Plot(visible=False) | |
growth_rate_plot = gr.Plot(visible=False) | |
with gr.Row(): | |
post_eng_rate_plot = gr.Plot(visible=False) | |
with gr.Row(): | |
interaction_data = gr.Plot(visible=False) | |
fetch_analytics_btn.click( | |
fn=guarded_fetch_analytics, | |
inputs=[], | |
outputs=[follower_count, follower_plot, growth_rate_plot, post_eng_rate_plot, interaction_data] | |
) | |
with gr.TabItem("3οΈβ£ Mentions"): | |
gr.Markdown("Analyze sentiment of recent posts that mention your organization.") | |
fetch_mentions_btn = gr.Button("π§ Fetch Mentions & Sentiment", variant="primary") | |
mentions_html = gr.HTML() | |
mentions_plot = gr.Plot(visible=False) | |
fetch_mentions_btn.click( | |
fn=run_mentions_and_load, | |
inputs=[], | |
outputs=[mentions_html, mentions_plot] | |
) | |
# Launch the app | |
if __name__ == "__main__": | |
app.launch(server_name="0.0.0.0", server_port=7860, share=True) | |