# -*- coding: utf-8 -*- import gradio as gr import json import requests # Added for API calls import os # Added for environment variables import urllib.parse # Added for URL encoding (though requests handles params well) # Assuming these custom modules exist in your project directory or Python path 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 # Import the function from your utils file from gradio_utils import get_url_user_token # Assuming gradio_utils.py is in the same directory from Bubble_API_Calls import fetch_linkedin_token_from_bubble # Shared state for token received via POST or Bubble token_received = {"status": False, "token": None, "client_id": None} # --- Handlers for token reception (POST) 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: print(f"Error decoding accessToken (POST): {e}") token_received["status"] = False token_received["token"] = None token_received["client_id"] = client_id return "❌ Invalid token format (POST)", "", client_id token_received["status"] = True token_received["token"] = token_dict # This should be the dict like {"access_token": "value"} token_received["client_id"] = client_id print(f"Token (from POST) received successfully. Client ID: {client_id}") # Update status box, token display, client display directly return check_status(), show_token(), show_client() def check_status(): return "✅ Token available" if token_received["status"] else "❌ Waiting for token…" def show_token(): # Shows access_token if available if token_received["status"] and token_received["token"] and isinstance(token_received["token"], dict): return token_received["token"].get("access_token", "Access token key missing in dict") elif token_received["status"] and token_received["token"]: # If token is a raw string (should not happen with new logic) return str(token_received["token"]) # Fallback, but ideally token_received["token"] is always a dict if status is True return "" def show_client(): return token_received["client_id"] if token_received["status"] and token_received["client_id"] else "" # --- Guarded fetch functions (using token from POST or Bubble) --- # These functions expect token_received["token"] to be a dictionary # like {"access_token": "actual_token_value", ...} def guarded_fetch_dashboard(): if not token_received["status"]: return "
❌ Access denied. No token available. Please send token first or ensure URL token is valid.
" html = fetch_and_render_dashboard( token_received["client_id"], token_received["token"] ) return html def guarded_fetch_analytics(): if not token_received["status"]: return ( "❌ Access denied. No token available.
", None, None, None, None, None, None, None ) count_md, plot, growth_plot, avg_post_eng_rate, interaction_metrics, eb_metrics, mentions_vol_metrics, mentions_sentiment_metrics = fetch_and_render_analytics( token_received["client_id"], token_received["token"] ) return count_md, plot, growth_plot, avg_post_eng_rate, interaction_metrics, eb_metrics, mentions_vol_metrics, mentions_sentiment_metrics def run_mentions_and_load(): if not token_received["status"]: return ("❌ Access denied. No token available.
", None) 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("Token can be supplied via URL parameter (for Bubble.io lookup) or hidden POST. Then explore dashboard and analytics.") # Hidden elements: simulate POST endpoint for OAuth token hidden_token_input = gr.Textbox(visible=False, elem_id="hidden_token") hidden_client_input = gr.Textbox(visible=False, elem_id="hidden_client_id") hidden_btn = gr.Button(visible=False, elem_id="hidden_btn") # --- Display elements --- url_user_token_display = gr.Textbox( label="User Token (from URL - Hidden)", interactive=False, placeholder="Attempting to load from URL...", visible=False ) parsed_token_dict = gr.Textbox(label="Bubble API Call Status", interactive=False, placeholder="Waiting for URL token...") status_box = gr.Textbox(label="Overall Token Status", interactive=False) token_display = gr.Textbox(label="Access Token (Active)", interactive=False) client_display = gr.Textbox(label="Client ID (Active)", interactive=False) # --- Load URL parameter on app start & Link to Bubble Fetch --- app.load( fn=get_url_user_token, inputs=None, outputs=[url_user_token_display] ) url_user_token_display.change( fn=fetch_linkedin_token_from_bubble, inputs=[url_user_token_display], outputs=[parsed_token_dict] ) hidden_btn.click( fn=receive_token, inputs=[hidden_token_input, hidden_client_input], outputs=[status_box, token_display, client_display] ) app.load(fn=check_status, outputs=status_box) app.load(fn=show_token, outputs=token_display) app.load(fn=show_client, outputs=client_display) timer = gr.Timer(2.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="Waiting for token...
") 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("Waiting for token...
") with gr.Row(): follower_plot = gr.Plot(visible=True) growth_rate_plot = gr.Plot(visible=True) with gr.Row(): post_eng_rate_plot = gr.Plot(visible=True) with gr.Row(): interaction_data = gr.Plot(visible=True) with gr.Row(): eb_data = gr.Plot(visible=True) with gr.Row(): mentions_vol_data = gr.Plot(visible=True) mentions_sentiment_data = gr.Plot(visible=True) fetch_analytics_btn.click( fn=guarded_fetch_analytics, inputs=[], outputs=[follower_count, follower_plot, growth_rate_plot, post_eng_rate_plot, interaction_data, eb_data, mentions_vol_data, mentions_sentiment_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(value="Waiting for token...
") mentions_plot = gr.Plot(visible=True) fetch_mentions_btn.click( fn=run_mentions_and_load, inputs=[], outputs=[mentions_html, mentions_plot] ) # Launch the app if __name__ == "__main__": # Ensure the 'Bubble_API' environment variable is set where this app is run. # For local testing, you can set it in your terminal before running: # export Bubble_API="YOUR_ACTUAL_BUBBLE_API_KEY" # python app.py app.launch(server_name="0.0.0.0", server_port=7860, share=True)