"""Instagram and Pinterest collection via Apify actors. One file for both platforms: the API mechanics are identical and only the input shape and the output field names differ. Actor output schemas drift between versions and between competing store actors, so every field read goes through pick() with several candidate names rather than assuming one layout. Runs are async (start -> poll -> page the dataset) instead of run-sync, because scraping a few dozen profiles routinely outlives run-sync's timeout. python collect/apify.py --platform instagram --niche finance python collect/apify.py --platform pinterest --actor some~pinterest-actor """ import argparse import sys import time import requests import yaml sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) from common import NICHES, RAW, ROOT, append_jsonl, need, seen_ids # noqa: E402 BASE = "https://api.apify.com/v2" # apify~instagram-scraper is the store's maintained IG actor. Pinterest has no # equally canonical one, so pass --actor with whichever you pick from the store. DEFAULT_ACTORS = { "instagram": "apify~instagram-scraper", "pinterest": None, } def pick(d, *names, default=None): """First present, non-empty value among several candidate key names.""" for n in names: if n in d and d[n] not in (None, "", [], {}): return d[n] return default class Apify: def __init__(self, token): self.token = token self.session = requests.Session() def run(self, actor, payload, poll=10, timeout=3600): r = self.session.post(f"{BASE}/acts/{actor}/runs", params={"token": self.token}, json=payload, timeout=60) if r.status_code >= 300: raise RuntimeError(f"start {actor}: {r.status_code} {r.text[:300]}") run = r.json()["data"] rid, ds = run["id"], run["defaultDatasetId"] print(f" run {rid} started") deadline = time.time() + timeout while time.time() < deadline: time.sleep(poll) s = self.session.get(f"{BASE}/actor-runs/{rid}", params={"token": self.token}, timeout=60) status = s.json()["data"]["status"] if status in ("SUCCEEDED", "FAILED", "ABORTED", "TIMED-OUT"): break print(f" ...{status}") else: raise RuntimeError(f"run {rid} exceeded {timeout}s") if status != "SUCCEEDED": print(f" run finished {status} — keeping whatever landed in the dataset", file=sys.stderr) return self.dataset(ds) def dataset(self, ds_id): items, offset = [], 0 while True: r = self.session.get(f"{BASE}/datasets/{ds_id}/items", params={"token": self.token, "offset": offset, "limit": 1000, "clean": "true"}, timeout=120) r.raise_for_status() page = r.json() if not page: break items.extend(page) offset += len(page) if len(page) < 1000: break return items def ig_profile_input(handles): return {"directUrls": [f"https://www.instagram.com/{h.lstrip('@')}/" for h in handles], "resultsType": "details", "resultsLimit": 1} def ig_posts_input(handles, limit): return {"directUrls": [f"https://www.instagram.com/{h.lstrip('@')}/" for h in handles], "resultsType": "posts", "resultsLimit": limit, "addParentData": True} def norm_instagram(it, niche, followers): code = pick(it, "shortCode", "shortcode", "code", "id") if not code: return None handle = str(pick(it, "ownerUsername", "username", "owner_username", default="")).lower() likes = int(pick(it, "likesCount", "likes_count", "likeCount", default=0) or 0) comments = int(pick(it, "commentsCount", "comments_count", default=0) or 0) views = int(pick(it, "videoPlayCount", "videoViewCount", "playCount", default=0) or 0) caption = pick(it, "caption", "text", "edge_media_to_caption", default="") or "" if not caption.strip(): return None # nothing to learn from a caption-less post return { "id": f"ig:{code}", "platform": "instagram", "niche": niche, "creator": handle, "creator_name": pick(it, "ownerFullName", "fullName", default=handle), "creator_size": int(followers.get(handle, 0) or 0), "published_at": pick(it, "timestamp", "taken_at", "takenAt", default=""), "duration_s": pick(it, "videoDuration", "duration"), "url": pick(it, "url", default=f"https://www.instagram.com/p/{code}/"), "text": { "caption": caption, "hashtags": pick(it, "hashtags", default=[]) or [], }, "metrics": {"likes": likes, "comments": comments, "views": views}, } def norm_pinterest(it, niche, followers): pid = pick(it, "id", "pinId", "pin_id") if not pid: return None handle = str(pick(it, "ownerUsername", "username", "boardOwner", default="")).lower() title = pick(it, "title", "gridTitle", "grid_title", default="") or "" desc = pick(it, "description", "closeupDescription", "alt_text", default="") or "" if not (title.strip() or desc.strip()): return None return { "id": f"pin:{pid}", "platform": "pinterest", "niche": niche, "creator": handle, "creator_name": pick(it, "ownerFullName", "boardName", default=handle), "creator_size": int(followers.get(handle, 0) or 0), "published_at": pick(it, "createdAt", "created_at", "timestamp", default=""), "duration_s": None, "url": pick(it, "url", "link", default=f"https://www.pinterest.com/pin/{pid}/"), "text": {"title": title, "description": desc}, "metrics": { "saves": int(pick(it, "saveCount", "repinCount", "saves", default=0) or 0), "comments": int(pick(it, "commentCount", "comments", default=0) or 0), "reactions": int(pick(it, "reactionCount", "reactions", default=0) or 0), }, } NORMALIZERS = {"instagram": norm_instagram, "pinterest": norm_pinterest} def collect(client, platform, actor, niche, cfg, limit): handles = [str(c[platform]).lstrip("@").lower() for c in (cfg.get("creators") or []) if c.get(platform)] if not handles: print(f" {niche}: no {platform} handles in creators.yaml — skipping") return 0 out = RAW / f"{platform}.jsonl" have = seen_ids(out) followers = {} if platform == "instagram": print(f" {niche}: profile pass for {len(handles)} handles") for prof in client.run(actor, ig_profile_input(handles)): h = str(pick(prof, "username", "ownerUsername", default="")).lower() if h: followers[h] = pick(prof, "followersCount", "followers_count", default=0) payload = ig_posts_input(handles, limit) else: # Store Pinterest actors generally take a list of profile/board urls. payload = {"startUrls": [{"url": f"https://www.pinterest.com/{h}/"} for h in handles], "maxItems": limit * len(handles)} print(f" {niche}: post pass ({limit}/profile)") items = client.run(actor, payload) norm = NORMALIZERS[platform] rows = [] for it in items: rec = norm(it, niche, followers) if rec and rec["id"] not in have: rows.append(rec) have.add(rec["id"]) append_jsonl(out, rows) print(f" {niche}: +{len(rows)} of {len(items)} raw items -> {out}") return len(rows) def main(): ap = argparse.ArgumentParser() ap.add_argument("--platform", choices=list(NORMALIZERS), required=True) ap.add_argument("--niche", choices=list(NICHES), help="default: all") ap.add_argument("--actor", help="Apify actor id, e.g. apify~instagram-scraper") ap.add_argument("--limit", type=int, default=150, help="posts per profile") args = ap.parse_args() actor = args.actor or DEFAULT_ACTORS[args.platform] if not actor: sys.exit(f"No default actor for {args.platform}. Pick one from the Apify " f"store and pass --actor .") cfg = yaml.safe_load((ROOT / "creators.yaml").read_text()) or {} client = Apify(need("APIFY_TOKEN")) total = 0 for n in ([args.niche] if args.niche else list(NICHES)): try: total += collect(client, args.platform, actor, n, cfg.get(n) or {}, args.limit) except Exception as e: print(f" {n}: failed ({e}) — continuing", file=sys.stderr) print(f"\n{total} new {args.platform} records") if __name__ == "__main__": main()