Instagram API get stories: any public account, not yours

import csv
import os
import sys
import time
from datetime import datetime, timezone

import requests

BASE = "https://api.toolzerhub.com/v2/instagram"
HEADERS = {"x-api-key": os.environ.get("TOOLZERHUB_API_KEY", "")}
FIELDS = ["account", "source", "story_id", "posted_at", "expires_at", "type", "seconds",
          "link", "file"]
calls = 0


def cell(value):
    # Spreadsheets run text starting with = + - or @ as a formula; an apostrophe keeps it text.
    if isinstance(value, str) and value[:1] in ("=", "+", "-", "@", "\t"):
        return "'" + value
    return value


def get(path, params, tries=4):
    # 429 and 503 are retried; the wait comes from the error body or the Retry-After header.
    global calls
    for attempt in range(tries):
        r = requests.get(BASE + path, headers=HEADERS, params=params, timeout=90)
        calls += 1
        if r.status_code == 400:  # malformed username: skip this account
            print("bad request:", str(r.json()["error"].get("message")).split("\n")[0])
            return None
        if r.status_code not in (429, 503) or attempt == tries - 1:
            if r.status_code >= 500:  # any 5xx, or a 503 past the retries: skip
                print(f"HTTP {r.status_code}, skipping:", r.text[:120].replace("\n", " "))
                return None
            r.raise_for_status()
            return r.json().get("data")
        try:
            wait = r.json()["error"]["details"]["retryAfter"]
        except (ValueError, KeyError, TypeError):
            wait = None
        try:
            wait = int(wait or r.headers.get("Retry-After") or 5 * (attempt + 1))
        except ValueError:
            wait = 5 * (attempt + 1)
        time.sleep(min(wait, 60))


def when(ts):
    return datetime.fromtimestamp(ts, timezone.utc).isoformat() if ts else ""


def save(item, folder, video):
    # Media URLs carry a signed expiry, so download in the same run.
    path = os.path.join(folder, f"{item['pk']}.{'mp4' if video else 'jpg'}")
    try:
        url = (item["video_versions"][0] if video else item["image_versions2"]["candidates"][0])["url"]
        with requests.get(url, timeout=120, stream=True) as r:
            r.raise_for_status()
            with open(path, "wb") as f:
                for chunk in r.iter_content(1 << 16):
                    f.write(chunk)
        return path
    except (KeyError, IndexError, TypeError, OSError, requests.RequestException) as e:
        print("  download failed:", item.get("pk"), type(e).__name__)
        if os.path.exists(path):
            os.remove(path)  # no half-written files
        return ""


def write(writer, account, source, items, folder, seen):
    for item in items:
        if not item.get("pk") or item["pk"] in seen:
            continue
        seen.add(item["pk"])
        video = bool(item.get("video_versions"))
        links = [(s.get("story_link") or {}).get("url") for s in item.get("story_link_stickers") or []]
        writer.writerow({
            "account": account, "source": cell(source), "story_id": item["pk"],
            "posted_at": when(item.get("taken_at")), "expires_at": when(item.get("expiring_at")),
            "type": "video" if video else "image", "seconds": round(item.get("video_duration") or 0) or "",
            "link": cell(" ".join(u for u in links if u)),
            "file": save(item, folder, video) if folder else "",
        })


def export_account(writer, account, folder, highlights):
    seen = set()
    data = get("/user/stories", {"username_or_id": account})
    stories = ((data or {}).get("reel") or {}).get("items") or []  # no live story: reel is null
    write(writer, account, "story", stories, folder, seen)
    tray = []
    if highlights and data is not None:  # skip highlights for unknown or failed accounts
        tray = ((get("/user/highlights", {"username_or_id": account}) or {}).get("tray")) or []
        for h in tray:
            items = (get("/highlight/stories", {"highlight_id": h.get("id")}) or {}).get("items") or []
            write(writer, account, f"highlight: {h.get('title') or h.get('id')}", items, folder, seen)
            time.sleep(0.3)
    print(f"{account}: {len(stories)} live stories, {len(tray)} highlights, {len(seen)} rows")
    return len(seen)


def main(accounts, highlights="no", folder="", out="stories.csv"):
    if folder:
        os.makedirs(folder, exist_ok=True)
    total = 0
    with open(out, "w", newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, fieldnames=FIELDS)
        writer.writeheader()
        for account in [a.strip().lstrip("@") for a in accounts.split(",") if a.strip()]:
            total += export_account(writer, account, folder, highlights == "yes")
    print(f"wrote {total} rows to {out} in {calls} calls")


if __name__ == "__main__":
    if len(sys.argv) < 2 or (len(sys.argv) > 2 and sys.argv[2] not in ("yes", "no")):
        sys.exit('usage: python instagram_stories.py nasa,nba [yes|no highlights] [media_folder or ""] [file.csv]')
    if not HEADERS["x-api-key"]:
        sys.exit("set TOOLZERHUB_API_KEY first")
    main(*sys.argv[1:])

export TOOLZERHUB_API_KEY=your_key
python instagram_stories.py nasa,nike yes media stories.csv

nasa: 2 live stories, 5 highlights, 91 rows
nike: 0 live stories, 1 highlights, 48 rows
wrote 139 rows to stories.csv in 11 calls