#!/usr/bin/env python3
"""FirstPour - monthly Instagram top-posts pull.
Uses the Instagram Graph API Business Discovery endpoint to fetch public
post metrics for a list of Business/Creator accounts, then builds:
posts.json - every post from the month, with engagement metrics + ranks
accounts.json - one row per account per month (history appended)
summary.json - outliers, format/day/hashtag trends for the written insights
Usage:
python3 ig_top_posts.py --token TOKEN --ig-user-id 17841420387812019 \
--accounts accounts.txt --month 2026-10 --out out/ [--history accounts.json]
"""
import argparse, json, re, statistics, sys, time
from collections import Counter, defaultdict
from datetime import datetime, timezone
import requests
API = "https://graph.facebook.com/v25.0"
MEDIA_FIELDS_RICH = ("id,caption,comments_count,like_count,view_count,media_type,"
"media_product_type,permalink,timestamp,media_url,thumbnail_url")
MEDIA_FIELDS_BASIC = "id,caption,comments_count,like_count,view_count,media_type,permalink,timestamp"
TOP_N = 30
def month_bounds(month):
start = datetime.strptime(month + "-01", "%Y-%m-%d").replace(tzinfo=timezone.utc)
end = start.replace(year=start.year + 1, month=1) if start.month == 12 else start.replace(month=start.month + 1)
return start, end
def bd_call(token, ig_user_id, username, media_fields, after=None):
media = "media"
if after:
media += f".after({after})"
media += f".limit(50){{{media_fields}}}"
fields = f"business_discovery.username({username}){{username,name,followers_count,media_count,{media}}}"
for attempt in range(3):
r = requests.get(f"{API}/{ig_user_id}", params={"fields": fields, "access_token": token}, timeout=60)
if r.status_code == 200:
return r.json().get("business_discovery", {}), None
err = r.json().get("error", {}) if r.headers.get("content-type", "").startswith("application/json") else {}
if err.get("code") in (4, 17, 32, 613): # rate limited
time.sleep(60 * (attempt + 1)); continue
return None, err or {"message": r.text[:300]}
return None, {"message": "rate limited repeatedly"}
def fetch_account(token, ig_user_id, username, start):
fields = MEDIA_FIELDS_RICH
posts, after, info = [], None, None
while True:
data, err = bd_call(token, ig_user_id, username, fields, after)
if err and fields == MEDIA_FIELDS_RICH and after is None:
fields = MEDIA_FIELDS_BASIC # some fields not allowed: retry plain
data, err = bd_call(token, ig_user_id, username, fields, after)
if err:
return info, posts, err
if info is None:
info = {k: data.get(k) for k in ("username", "name", "followers_count", "media_count")}
page = data.get("media", {})
items = page.get("data", [])
posts.extend(items)
oldest = min((i["timestamp"] for i in items), default=None)
after = page.get("paging", {}).get("cursors", {}).get("after")
if not items or not after or (oldest and datetime.fromisoformat(oldest.replace("+0000", "+00:00")) < start):
break
return info, posts, None
def fmt_type(p):
if p.get("media_product_type") == "REELS" or (p.get("media_type") == "VIDEO" and p.get("view_count") is not None):
return "Reel"
return {"CAROUSEL_ALBUM": "Carousel", "IMAGE": "Image", "VIDEO": "Video"}.get(p.get("media_type"), p.get("media_type"))
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--token", required=True)
ap.add_argument("--ig-user-id", required=True)
ap.add_argument("--accounts", required=True, help="text file, one username per line")
ap.add_argument("--month", required=True, help="YYYY-MM")
ap.add_argument("--out", required=True)
ap.add_argument("--history", help="previous accounts.json to append to")
a = ap.parse_args()
import os; os.makedirs(a.out, exist_ok=True)
start, end = month_bounds(a.month)
usernames = [u.strip().lstrip("@") for u in open(a.accounts) if u.strip() and not u.startswith("#")]
history = json.load(open(a.history)) if a.history and os.path.exists(a.history) else []
history = [h for h in history if h["month"] != a.month]
prev = {h["account"]: h for h in history if h["month"] == max((x["month"] for x in history), default="")}
posts_out, acct_rows, skipped = [], [], []
for u in usernames:
info, posts, err = fetch_account(a.token, a.ig_user_id, u, start)
if err:
skipped.append({"account": u, "reason": err.get("error_user_msg") or err.get("message", "unknown")})
print(f"skip {u}: {skipped[-1]['reason']}", file=sys.stderr); continue
fol = info.get("followers_count") or 0
month_posts = []
for p in posts:
ts = datetime.fromisoformat(p["timestamp"].replace("+0000", "+00:00"))
if not (start <= ts < end):
continue
likes, comments, views = p.get("like_count"), p.get("comments_count") or 0, p.get("view_count")
eng = (likes or 0) + comments
month_posts.append({
"id": p["id"], "account": info["username"], "followers": fol,
"date": ts.strftime("%Y-%m-%d"), "weekday": ts.strftime("%a"), "hour_utc": ts.hour,
"format": fmt_type(p), "likes": likes, "likes_hidden": likes is None,
"comments": comments, "views": views,
"engagement": eng, "er": round(eng / fol, 5) if fol else None,
"views_per_follower": round(views / fol, 3) if (views and fol) else None,
"permalink": p.get("permalink"),
"caption": (p.get("caption") or "")[:280],
"thumb_src": p.get("thumbnail_url") or (p.get("media_url") if p.get("media_type") != "VIDEO" else None),
})
ers = [x["er"] for x in month_posts if x["er"] is not None]
med = statistics.median(ers) if ers else None
for x in month_posts:
x["vs_account_median"] = round(x["er"] / med, 2) if (med and x["er"]) else None
reels = [x["views"] for x in month_posts if x["format"] == "Reel" and x["views"]]
p0 = prev.get(info["username"])
acct_rows.append({
"month": a.month, "account": info["username"], "name": info.get("name"),
"followers": fol,
"follower_change": (fol - p0["followers"]) if p0 else None,
"posts": len(month_posts),
"posts_change": (len(month_posts) - p0["posts"]) if p0 else None,
"reels": sum(1 for x in month_posts if x["format"] == "Reel"),
"median_er": round(med, 5) if med else None,
"avg_reel_views": round(statistics.mean(reels)) if reels else None,
"likes_hidden": any(x["likes_hidden"] for x in month_posts),
})
posts_out.extend(month_posts)
# ranks
reels = sorted([p for p in posts_out if p["format"] == "Reel" and p["views"]], key=lambda p: -p["views"])
feed = sorted([p for p in posts_out if p["format"] != "Reel" and p["er"] is not None], key=lambda p: -p["er"])
for i, p in enumerate(reels): p["rank_reels"] = i + 1
for i, p in enumerate(feed): p["rank_feed"] = i + 1
for p in posts_out:
p["month"] = a.month
p.setdefault("rank_reels", None); p.setdefault("rank_feed", None)
# trends for the written insights
by_fmt = defaultdict(list)
for p in posts_out:
if p["er"] is not None: by_fmt[p["format"]].append(p["er"])
by_day = defaultdict(list)
for p in posts_out:
if p["vs_account_median"]: by_day[p["weekday"]].append(p["vs_account_median"])
tags = lambda ps: Counter(t.lower() for p in ps for t in re.findall(r"#(\w+)", p["caption"]))
top = reels[:TOP_N] + feed[:TOP_N]
summary = {
"month": a.month, "accounts_tracked": len(usernames), "accounts_returned": len(acct_rows),
"skipped": skipped, "posts": len(posts_out),
"median_er_by_format": {k: round(statistics.median(v), 5) for k, v in by_fmt.items()},
"posts_by_format": {k: len(v) for k, v in by_fmt.items()},
"relative_er_by_weekday": {k: round(statistics.median(v), 2) for k, v in by_day.items()},
"breakouts": [ # posts doing 3x+ their own account's median
{k: p[k] for k in ("account", "date", "format", "er", "views", "vs_account_median", "permalink", "caption")}
for p in sorted(posts_out, key=lambda p: -(p["vs_account_median"] or 0)) if (p["vs_account_median"] or 0) >= 3
][:15],
"top_hashtags_in_top_posts": tags(top).most_common(15),
"follower_movers": sorted([r for r in acct_rows if r["follower_change"] is not None],
key=lambda r: -abs(r["follower_change"] / max(r["followers"], 1)))[:8],
"posting_changes": [r for r in acct_rows if r["posts_change"] and abs(r["posts_change"]) >= 5],
"top_reels": [{k: p[k] for k in ("account", "views", "views_per_follower", "caption")} for p in reels[:10]],
"top_feed": [{k: p[k] for k in ("account", "format", "er", "caption")} for p in feed[:10]],
}
json.dump(posts_out, open(f"{a.out}/posts.json", "w"), indent=1)
json.dump(history + acct_rows, open(f"{a.out}/accounts.json", "w"), indent=1)
json.dump(summary, open(f"{a.out}/summary.json", "w"), indent=1)
print(f"{a.month}: {len(acct_rows)}/{len(usernames)} accounts, {len(posts_out)} posts, "
f"{len(reels)} reels, {len(feed)} feed posts, {len(skipped)} skipped")
if __name__ == "__main__":
main()
