import requests
import csv
import time
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
API_KEY = "YOUR_API_KEY"
BASE = "https://api.sorsa.io/v3"
HEADERS = {"ApiKey": API_KEY}
JSON_HEADERS = {**HEADERS, "Content-Type": "application/json"}
BRAND = "your_handle"
COMPETITORS = ["competitor1", "competitor2", "competitor3"]
SNAPSHOT_FILE = "snapshots.csv"
analyzer = SentimentIntensityAnalyzer()
def get_profiles(usernames):
resp = requests.get(
f"{BASE}/info-batch",
headers=HEADERS,
params=[("usernames", u) for u in usernames],
timeout=30,
)
resp.raise_for_status()
return resp.json().get("users", [])
def log_snapshot(profiles, output_file=SNAPSHOT_FILE):
file_exists = Path(output_file).exists()
today = date.today().isoformat()
with open(output_file, "a", newline="") as f:
writer = csv.writer(f)
if not file_exists:
writer.writerow(["date", "username", "followers", "tweets", "following"])
for p in profiles:
writer.writerow([
today, p["username"], p["followers_count"],
p["tweets_count"], p["followings_count"],
])
def compute_growth(csv_file, username, days=7):
with open(csv_file, encoding="utf-8") as f:
rows = [r for r in csv.DictReader(f) if r["username"].lower() == username.lower()]
if len(rows) < 2:
return None
rows.sort(key=lambda row: row["date"])
latest_row = rows[-1]
cutoff = date.fromisoformat(latest_row["date"]) - timedelta(days=days)
earlier_rows = [r for r in rows if date.fromisoformat(r["date"]) <= cutoff]
if not earlier_rows:
return None
latest = int(latest_row["followers"])
earlier = int(earlier_rows[-1]["followers"])
return ((latest - earlier) / earlier) * 100 if earlier else None
def fetch_user_tweets(username, max_pages=5):
all_tweets = []
cursor = None
for _ in range(max_pages):
body = {"username": username}
if cursor:
body["next_cursor"] = cursor
resp = requests.post(f"{BASE}/user-tweets", headers=JSON_HEADERS, json=body, timeout=30)
resp.raise_for_status()
data = resp.json()
all_tweets.extend(data.get("tweets", []))
cursor = data.get("next_cursor")
if not cursor:
break
time.sleep(0.1)
return all_tweets
def analyze_content(tweets, username):
if not tweets:
return None
total = len(tweets)
likes = [t.get("likes_count", 0) for t in tweets]
original = sum(1 for t in tweets if not t.get("is_reply") and not t.get("retweeted_status"))
with_media = sum(1 for t in tweets if t.get("entities"))
return {
"username": username,
"sample_size": total,
"avg_likes": sum(likes) / total,
"original_pct": original / total * 100,
"media_pct": with_media / total * 100,
}
def fetch_verified_followers(username, max_pages=3):
all_users = []
cursor = None
for _ in range(max_pages):
params = {"username": username}
if cursor:
params["next_cursor"] = cursor
resp = requests.get(f"{BASE}/verified-followers", headers=HEADERS, params=params, timeout=30)
resp.raise_for_status()
data = resp.json()
all_users.extend(data.get("users", []))
cursor = data.get("next_cursor")
if not cursor:
break
time.sleep(0.1)
return all_users
def fetch_mentions(handle, min_likes=10, since_date=None, until_date=None, max_pages=5):
all_mentions = []
cursor = None
for _ in range(max_pages):
body = {"query": handle, "order": "popular", "min_likes": min_likes}
if since_date:
body["since_date"] = since_date
if until_date:
body["until_date"] = until_date
if cursor:
body["next_cursor"] = cursor
resp = requests.post(f"{BASE}/mentions", headers=JSON_HEADERS, json=body, timeout=30)
resp.raise_for_status()
data = resp.json()
all_mentions.extend(data.get("tweets", []))
cursor = data.get("next_cursor")
if not cursor:
break
time.sleep(0.1)
return all_mentions
def classify_sentiment(mentions):
results = {"positive": [], "negative": [], "neutral": []}
for m in mentions:
score = analyzer.polarity_scores(m["full_text"])["compound"]
if score >= 0.05:
results["positive"].append((score, m))
elif score <= -0.05:
results["negative"].append((score, m))
else:
results["neutral"].append((score, m))
return results
def count_mentions(handle, days=7, min_likes=0):
until = datetime.now(timezone.utc).date().isoformat()
since = (datetime.now(timezone.utc) - timedelta(days=days)).date().isoformat()
return len(fetch_mentions(handle, min_likes=min_likes, since_date=since, until_date=until, max_pages=20))
def header(text):
line = "=" * 64
print(f"\n{line}\n{text}\n{line}")
def run_weekly_report():
header("PHASE 1: PROFILE BENCHMARKS")
profiles = get_profiles(COMPETITORS + [BRAND])
print(f"{'Handle':<18} {'Followers':>12} {'Tweets':>10} {'Verified':>10}")
for p in profiles:
print(f"@{p['username']:<17} {p['followers_count']:>12,} "
f"{p['tweets_count']:>10,} {str(p.get('verified', False)):>10}")
log_snapshot(profiles)
for h in COMPETITORS + [BRAND]:
g = compute_growth(SNAPSHOT_FILE, h, days=7)
if g is not None:
print(f" @{h}: {g:+.2f}% weekly follower growth")
header("PHASE 2: CONTENT STRATEGY")
for handle in COMPETITORS:
tweets = fetch_user_tweets(handle, max_pages=5)
result = analyze_content(tweets, handle)
if result:
print(f"@{result['username']}: avg {result['avg_likes']:.0f} likes/tweet, "
f"{result['original_pct']:.0f}% original, "
f"{result['media_pct']:.0f}% with media")
header("PHASE 3: VERIFIED FOLLOWERS")
for handle in COMPETITORS:
verified = fetch_verified_followers(handle, max_pages=3)
print(f"@{handle}: {len(verified)} verified followers in top pages")
header("PHASE 4: SENTIMENT")
for handle in COMPETITORS:
mentions = fetch_mentions(handle, min_likes=10, max_pages=3)
s = classify_sentiment(mentions)
print(f"@{handle}: {len(s['positive'])} pos / {len(s['negative'])} neg "
f"/ {len(s['neutral'])} neutral (n={len(mentions)})")
header("PHASE 5: SHARE OF VOICE (7d)")
your_n = count_mentions(BRAND, days=7, min_likes=5)
comp_n = {h: count_mentions(h, days=7, min_likes=5) for h in COMPETITORS}
total = your_n + sum(comp_n.values())
if total:
print(f" @{BRAND}: {your_n} ({your_n/total*100:.1f}%)")
for h, n in sorted(comp_n.items(), key=lambda x: -x[1]):
print(f" @{h}: {n} ({(n/total*100 if total else 0):.1f}%)")
if __name__ == "__main__":
run_weekly_report()
