import json import requests from datetime import datetime, timezone, timedelta import matplotlib.pyplot as plt import gradio as gr import traceback import html from sessions import create_session from error_handling import display_error from Data_Fetching_and_Rendering import fetch_posts_and_stats from mentions_dashboard import generate_mentions_dashboard import logging logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s') API_V2_BASE = 'https://api.linkedin.com/v2' API_REST_BASE = 'https://api.linkedin.com/rest' def extract_follower_gains(data): elements = data.get("elements", []) if not elements: return [] results = [] for item in elements: start_timestamp = item.get("timeRange", {}).get("start") if not start_timestamp: continue try: date_str = datetime.fromtimestamp(start_timestamp / 1000, tz=timezone.utc).strftime('%Y-%m') except Exception: continue gains = item.get("followerGains", {}) results.append({ "date": date_str, "organic": gains.get("organicFollowerGain", 0) or 0, "paid": gains.get("paidFollowerGain", 0) or 0 }) return sorted(results, key=lambda x: x['date']) def fetch_analytics_data(client_id, token): if not token: raise ValueError("comm_token is missing.") token_dict = token if isinstance(token, dict) else {'access_token': token, 'token_type': 'Bearer'} session = create_session(client_id, token=token_dict) try: org_urn, org_name = "urn:li:organization:19010008", "GRLS" count_url = f"{API_V2_BASE}/networkSizes/{org_urn}?edgeType=CompanyFollowedByMember" follower_count = session.get(count_url).json().get("firstDegreeSize", 0) start = datetime.now(timezone.utc) - timedelta(days=365) start = start.replace(day=1, hour=0, minute=0, second=0, microsecond=0) start_ms = int(start.timestamp() * 1000) gains_url = ( f"{API_REST_BASE}/organizationalEntityFollowerStatistics" f"?q=organizationalEntity&organizationalEntity={org_urn}" f"&timeIntervals.timeGranularityType=MONTH" f"&timeIntervals.timeRange.start={start_ms}" ) gains_data = session.get(gains_url).json() gains = extract_follower_gains(gains_data) return org_name, follower_count, gains except requests.exceptions.RequestException as e: status = getattr(e.response, 'status_code', 'N/A') msg = f"Failed to fetch LinkedIn analytics (Status: {status})." raise ValueError(msg) from e except Exception as e: raise ValueError("Unexpected error during LinkedIn analytics fetch.") from e def plot_follower_gains(data): plt.style.use('seaborn-v0_8-whitegrid') if not data: fig, ax = plt.subplots(figsize=(10, 5)) ax.text(0.5, 0.5, 'No follower gains data.', ha='center', va='center', transform=ax.transAxes) ax.set_title('Monthly Follower Gains') ax.set_xticks([]); ax.set_yticks([]) return fig dates = [d['date'] for d in data] organic = [d['organic'] for d in data] paid = [d['paid'] for d in data] fig, ax = plt.subplots(figsize=(12, 6)) ax.plot(dates, organic, label='Organic', marker='o', color='#0073b1') ax.plot(dates, paid, label='Paid', marker='x', linestyle='--', color='#d9534f') ax.set(title='Monthly Follower Gains', xlabel='Month', ylabel='New Followers') ax.tick_params(axis='x', rotation=45) ax.legend() plt.tight_layout() return fig def plot_growth_rate(data, total): if not data: fig, ax = plt.subplots(figsize=(10, 5)) ax.text(0.5, 0.5, 'No data for growth rate.', ha='center', va='center', transform=ax.transAxes) ax.set_title('Growth Rate (%)') ax.set_xticks([]); ax.set_yticks([]) return fig dates = [d['date'] for d in data] gains = [d['organic'] + d['paid'] for d in data] history = [] current = total for g in reversed(gains): history.insert(0, current) current -= g rates = [((history[i] - history[i-1]) / history[i-1] * 100 if history[i-1] else 0) for i in range(1, len(history))] fig, ax = plt.subplots(figsize=(12, 6)) ax.plot(dates[1:], rates, label='Growth Rate (%)', marker='o', color='green') ax.set(title='Monthly Growth Rate (%)', xlabel='Month', ylabel='Growth %') ax.tick_params(axis='x', rotation=45) ax.legend() plt.tight_layout() return fig def compute_monthly_avg_engagement_rate(posts): from collections import defaultdict import statistics if not posts: return [] monthly_data = defaultdict(lambda: {"engagement_sum": 0, "post_count": 0, "impression_total": 0}) for post in posts: try: month = post["when"][:7] # Format: YYYY-MM likes = post.get("likes", 0) comments = post.get("comments", 0) shares = post.get("shares", 0) clicks = post.get("clicks", 0) impressions = post.get("impressions", 0) engagement = likes + comments + shares + clicks monthly_data[month]["engagement_sum"] += engagement monthly_data[month]["post_count"] += 1 monthly_data[month]["impression_total"] += impressions except Exception: continue results = [] for month in sorted(monthly_data.keys()): data = monthly_data[month] if data["post_count"] == 0 or data["impression_total"] == 0: rate = 0 else: avg_impressions = data["impression_total"] / data["post_count"] rate = (data["engagement_sum"] / (data["post_count"] * avg_impressions)) * 100 results.append({"month": month, "engagement_rate": round(rate, 2)}) return results def plot_avg_engagement_rate(data): import matplotlib.pyplot as plt if not data: fig, ax = plt.subplots(figsize=(10, 5)) ax.text(0.5, 0.5, 'No engagement data.', ha='center', va='center', transform=ax.transAxes) ax.set_title('Average Post Engagement Rate (%)') ax.set_xticks([]); ax.set_yticks([]) return fig months = [d["month"] for d in data] rates = [d["engagement_rate"] for d in data] fig, ax = plt.subplots(figsize=(12, 6)) ax.plot(months, rates, label="Engagement Rate (%)", marker="s", color="#ff7f0e") ax.set(title="Average Post Engagement Rate (%)", xlabel="Month", ylabel="Engagement Rate %") ax.tick_params(axis='x', rotation=45) ax.legend() plt.tight_layout() return fig def compute_post_interaction_metrics(posts): from collections import defaultdict if not posts: return [] monthly_stats = defaultdict(lambda: { "comments": 0, "shares": 0, "clicks": 0, "likes": 0, "posts": 0 }) for post in posts: try: month = post["when"][:7] # YYYY-MM monthly_stats[month]["comments"] += post.get("comments", 0) monthly_stats[month]["shares"] += post.get("shares", 0) monthly_stats[month]["clicks"] += post.get("clicks", 0) monthly_stats[month]["likes"] += post.get("likes", 0) monthly_stats[month]["posts"] += 1 except Exception: continue results = [] for month in sorted(monthly_stats.keys()): stats = monthly_stats[month] total_engagement = stats["comments"] + stats["shares"] + stats["clicks"] + stats["likes"] posts_count = stats["posts"] or 1 # Avoid division by zero results.append({ "month": month, "comments_per_post": round(stats["comments"] / posts_count, 2), "shares_per_post": round(stats["shares"] / posts_count, 2), "clicks_per_post": round(stats["clicks"] / posts_count, 2), "comment_share_of_engagement": round((stats["comments"] / total_engagement) * 100 if total_engagement else 0, 2) }) logging.info(f"this are the inter 0 else 0 change = (((total - prev_total) / prev_total) * 100) if prev_total else None volume_data.append({"month": month, "count": total, "change": round(change, 2) if change is not None else None}) return volume_data, sentiment_data def plot_mention_volume_trend(volume_data): fig, ax = plt.subplots(figsize=(12, 6)) if not volume_data: ax.text(0.5, 0.5, 'No Mention Volume Data.', ha='center', va='center', transform=ax.transAxes) ax.set_title('Mention Volume Over Time') return fig months = [d["month"] for d in volume_data] counts = [d["count"] for d in volume_data] ax.plot(months, counts, marker='o', linestyle='-', color="#1f77b4") ax.set(title="Monthly Mention Volume", xlabel="Month", ylabel="Mentions") ax.tick_params(axis='x', rotation=45) plt.tight_layout() return fig def plot_mention_sentiment_score(sentiment_data): fig, ax = plt.subplots(figsize=(12, 6)) if not sentiment_data: ax.text(0.5, 0.5, 'No Sentiment Score Data.', ha='center', va='center', transform=ax.transAxes) ax.set_title('Mention Sentiment Score') return fig months = [d["month"] for d in sentiment_data] scores = [d["score"] for d in sentiment_data] ax.plot(months, scores, marker='o', linestyle='-', color="#ff7f0e") ax.set(title="Monthly Sentiment Score (% Positive - % Negative)", xlabel="Month", ylabel="Score") ax.axhline(0, color='gray', linestyle='--', linewidth=1) ax.tick_params(axis='x', rotation=45) plt.tight_layout() return fig def fetch_and_render_analytics(client_id, token): loading = gr.update(value="

Loading follower count...

", visible=True) hidden = gr.update(value=None, visible=False) if not token: error = "

❌ Missing token. Please log in.

" return gr.update(value=error, visible=True), hidden, hidden try: name, count, gains = fetch_analytics_data(client_id, token) posts, org_name, sentiments = fetch_posts_and_stats(client_id, token, count=30) engagement_data = compute_monthly_avg_engagement_rate(posts) interaction_data = compute_post_interaction_metrics(posts) eb_data = compute_eb_content_ratio(posts) html_mentions, fig, mention_data = generate_mentions_dashboard(client_id, token) volume_data, sentiment_data = compute_mention_metrics(mention_data) count_html = f"""

Total Followers for

{html.escape(name)}

{count:,}

(As of latest data)

""" return gr.update(value=count_html, visible=True), gr.update(value=plot_follower_gains(gains), visible=True), gr.update(value=plot_growth_rate(gains, count), visible=True), gr.update(value=plot_avg_engagement_rate(engagement_data), visible=True), gr.update(value=plot_interaction_metrics(interaction_data), visible=True), gr.update(value=plot_eb_content_ratio(eb_data), visible=True), gr.update(value=plot_mention_volume_trend(volume_data), visible=True), gr.update(value=plot_mention_sentiment_score(sentiment_data), visible=True) except Exception as e: error = display_error("Analytics load failed.", e).get('value', "

Error loading data.

") return gr.update(value=error, visible=True), hidden, hidden