Spaces:
Running
Running
File size: 3,569 Bytes
9e34c6c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 |
# mentions_dashboard.py
import json
import time
import re
import os
from datetime import datetime
from urllib.parse import quote
from requests_oauthlib import OAuth2Session
from textblob import TextBlob
import matplotlib.pyplot as plt
def extract_text_from_commentary(commentary):
return re.sub(r"{.*?}", "", commentary).strip()
def analyze_sentiment(text):
return TextBlob(text).sentiment.polarity
def generate_mentions_dashboard(comm_client_id, comm_token_dict):
linkedin = OAuth2Session(comm_client_id, token=comm_token_dict)
org_urn = "urn:li:organization:19010008"
encoded_urn = quote(org_urn, safe='')
linkedin.headers.update({
"LinkedIn-Version": "202502",
"X-Restli-Protocol-Version": "2.0.0"
})
base_url = (
"https://api.linkedin.com/rest/organizationalEntityNotifications"
"?q=criteria"
"&actions=List(COMMENT,SHARE_MENTION)"
f"&organizationalEntity={encoded_urn}"
"&count=50"
)
all_notifications = []
start = 0
while True:
url = f"{base_url}&start={start}"
resp = linkedin.get(url)
if resp.status_code != 200:
break
data = resp.json()
elements = data.get("elements", [])
all_notifications.extend(elements)
if len(elements) < data.get("paging", {}).get("count", 0):
break
start += len(elements)
time.sleep(0.5)
# Extract mentions and their share URNs
mention_shares = [e.get("generatedActivity") for e in all_notifications if e.get("action") == "SHARE_MENTION"]
mention_data = []
for share_urn in mention_shares:
if not share_urn:
continue
encoded_share_urn = quote(share_urn, safe='')
share_url = f"https://api.linkedin.com/rest/posts/{encoded_share_urn}"
response = linkedin.get(share_url)
if response.status_code != 200:
continue
post = response.json()
commentary_raw = post.get("commentary", "")
if not commentary_raw:
continue
commentary = extract_text_from_commentary(commentary_raw)
sentiment = analyze_sentiment(commentary)
timestamp = post.get("createdAt", 0)
dt = datetime.fromtimestamp(timestamp / 1000.0)
mention_data.append({
"date": dt,
"text": commentary,
"sentiment": sentiment
})
# Save HTML
html_parts = ["<h2 style='text-align:center;'>📣 Mentions Sentiment Dashboard</h2>"]
for mention in mention_data:
html_parts.append(f"""
<div style='border:1px solid #ccc; border-radius:10px; padding:15px; margin:10px;'>
<p><strong>Date:</strong> {mention["date"].strftime('%Y-%m-%d')}</p>
<p>{mention["text"]}</p>
<p><strong>Sentiment:</strong> {mention["sentiment"]:.2f}</p>
</div>
""")
html_path = "mentions_dashboard.html"
with open(html_path, "w", encoding="utf-8") as f:
f.write("\n".join(html_parts))
# Plot
if mention_data:
dates = [m["date"] for m in mention_data]
sentiments = [m["sentiment"] for m in mention_data]
plt.figure(figsize=(10, 5))
plt.plot(dates, sentiments, marker='o', linestyle='-', color='blue')
plt.title("Sentiment Over Time")
plt.xlabel("Date")
plt.ylabel("Sentiment")
plt.grid(True)
plt.tight_layout()
plt.savefig("mentions_sentiment_plot.png")
plt.close()
return html_path
|