Social: the collector's snapshot from the platform reads

Inputs

Outputs

Source

python

# Input preparation
inp = __ivy_ctx__["nodes"][__ivy_node_id__]["input"]

# Compute
# Shared by every social node (built into each by build.mjs): dates, blocks, merging.
import datetime, json, re, statistics

def day(s):
    s = str(s or "")[:10]
    return s if re.fullmatch(r"\d{4}-\d{2}-\d{2}", s) else None

def now_iso():
    return datetime.datetime.utcnow().replace(microsecond=0).isoformat() + "Z"

def ms_of(s):
    if s is None or s == "":
        return None
    if isinstance(s, (int, float)):
        return int(s * 1000) if s < 1e12 else int(s)
    t = str(s).replace("Z", "+00:00")
    t = re.sub(r"([+-]\d{2})(\d{2})$", r"\1:\2", t)
    try:
        return int(datetime.datetime.fromisoformat(t).timestamp() * 1000)
    except ValueError:
        return None

def rows_of(block):
    """A {fields, rows} block as dicts."""
    if not isinstance(block, dict) or not block.get("fields"):
        return []
    f = block["fields"]
    return [dict(zip(f, r)) for r in block.get("rows") or []]

def block_of(dicts, first):
    fields = list(first)
    for d in dicts:
        for k in d:
            if k not in fields:
                fields.append(k)
    return {"fields": fields, "rows": [[d.get(k) for k in fields] for d in dicts]}

def listed(v):
    return [x for x in (v if isinstance(v, list) else [v] if v else []) if x is not None]

def text_of(v, limit=4000):
    s = " ".join(str(v).split()) if isinstance(v, (str, int, float)) else ""
    return s[:limit]

# The collector's schema v1 document from the platform nodes' answers: accounts with their daily rows and posts,
# the headline metric per account with its median over the lift window, each post's lift, the spans, the legend.
SET = {"range_presets": [{"label": "2 weeks", "days": 14}, {"label": "1 month", "days": 30}, {"label": "3 months", "days": 91}, {"label": "6 months", "days": 182}, {"label": "1 year", "days": 365}, {"label": "All", "days": 0}],
       "default_range": "3 months", "posts_default_days": 180, "lift_min_posts": int(inp.get("lift_min_posts") or 10), "marker_min_lift": 2.0, "lift_median_days": 365, "min_reach_for_rate": int(inp.get("min_reach_for_rate") or 50), "index_min_base": 10, "stale_amber_days": 2, "stale_red_days": 7}
HEADLINE = {"instagram": ["lifetime_reach", "lifetime_views"], "youtube": ["lifetime_viewCount"], "linkedin": ["lifetime_impressions", "range_impressions"], "x": ["lifetime_public.impression_count"], "facebook": ["lifetime_reach"], "tiktok": ["lifetime_views"]}
LEGEND = {
    "instagram": {"reach": ("accounts", "day, lifetime", "Unique accounts that saw any content"), "views": ("views", "day, lifetime", "Plays and views across content"), "follower_count": ("accounts", "day", "New followers per day (gross)"), "followers_count": ("accounts", "instant", "Followers (profile field)"), "total_interactions": ("interactions", "day, lifetime", "Likes, comments, shares, saves"), "accounts_engaged": ("accounts", "day", "Accounts that interacted with content"), "profile_links_taps": ("taps", "day", "Taps on the profile's links"), "likes": ("likes", "lifetime", "Likes"), "comments": ("comments", "lifetime", "Comments"), "shares": ("shares", "lifetime", "Shares"), "saved": ("saves", "lifetime", "Saves"), "ig_reels_avg_watch_time": ("ms", "lifetime", "Average watch time of a reel")},
    "youtube": {"views": ("views", "day, lifetime", "Views"), "estimatedMinutesWatched": ("minutes", "day", "Minutes watched"), "averageViewDuration": ("seconds", "day", "Average view duration"), "averageViewPercentage": ("percent", "day", "Average percentage of a video watched"), "subscribersGained": ("accounts", "day", "Subscribers gained"), "subscribersLost": ("accounts", "day", "Subscribers lost"), "likes": ("likes", "day", "Likes"), "comments": ("comments", "day", "Comments"), "shares": ("shares", "day", "Shares"), "subscriberCount": ("accounts", "instant", "Subscribers"), "viewCount": ("views", "instant, lifetime", "Views"), "likeCount": ("likes", "lifetime", "Likes"), "commentCount": ("comments", "lifetime", "Comments")},
    "linkedin": {"impressions_total": ("impressions", "day", "Impressions of the page's posts"), "unique_impressions_organic": ("impressions", "day", "Unique organic impressions"), "clicks_total": ("clicks", "day", "Clicks"), "reactions_total": ("reactions", "day", "Reactions"), "comments_total": ("comments", "day", "Comments"), "reposts_total": ("reposts", "day", "Reposts"), "engagement_rate_total": ("ratio", "day", "LinkedIn's engagement rate"), "new_total_followers": ("accounts", "day", "New followers"), "new_organic_followers": ("accounts", "day", "New organic followers"), "new_sponsored_followers": ("accounts", "day", "New sponsored followers"), "total_followers": ("accounts", "instant", "Followers"), "total_page_views_total": ("visits", "day", "Page views, all pages"), "overview_page_views_total": ("visits", "day", "Overview page views"), "jobs_page_views_total": ("visits", "day", "Jobs page views"), "total_unique_visitors_total": ("visitors", "day", "Unique visitors, all pages"), "impressions": ("impressions", "lifetime", "Impressions"), "clicks": ("clicks", "lifetime", "Clicks"), "likes": ("likes", "lifetime", "Reactions"), "comments": ("comments", "lifetime", "Comments"), "reposts": ("reposts", "lifetime", "Reposts"), "engagement": ("ratio", "lifetime", "Engagement rate")},
    "x": {"public.followers_count": ("accounts", "instant", "Followers"), "public.following_count": ("accounts", "instant", "Following"), "public.tweet_count": ("posts", "instant", "Posts"), "public.listed_count": ("lists", "instant", "Listed"), "public.impression_count": ("impressions", "lifetime", "Impressions"), "public.like_count": ("likes", "lifetime", "Likes"), "public.reply_count": ("replies", "lifetime", "Replies"), "public.retweet_count": ("reposts", "lifetime", "Reposts"), "public.quote_count": ("quotes", "lifetime", "Quotes"), "public.bookmark_count": ("bookmarks", "lifetime", "Bookmarks"), "non_public.impression_count": ("impressions", "lifetime", "Impressions (private)"), "non_public.engagements": ("engagements", "lifetime", "Engagements (private)"), "non_public.user_profile_clicks": ("clicks", "lifetime", "Profile clicks (private)"), "non_public.url_link_clicks": ("clicks", "lifetime", "Link clicks (private)")},
    "facebook": {"page_impressions": ("impressions", "day", "Impressions of the Page"), "page_fan_adds": ("accounts", "day", "New fans"), "fan_count": ("accounts", "instant", "Fans"), "followers_count": ("accounts", "instant", "Followers")},
}


class SocialSnapshotBuild:
    "Social: the collector's snapshot from the platform reads"

    def run(self, inp):
        out = {}
        generated_at = str(inp.get("generated_at") or now_iso())
        today = generated_at[:10]
        window_start = (datetime.date.fromisoformat(today) - datetime.timedelta(days=SET["lift_median_days"])).isoformat()
        accounts, platforms, flags, gaps, warnings = [], {}, [], [], []
        breakdowns = {"youtube": [], "instagram": {}, "linkedin": {}}
        for key in ("account_a", "account_b", "account_c", "account_d", "account_e", "account_f", "account_g", "account_h"):
            blk = inp.get(key)
            if not isinstance(blk, dict) or not isinstance(blk.get("account"), dict):
                continue
            a = dict(blk["account"])
            platform = a.get("platform") or "other"
            status = blk.get("status") or {"status": "ok", "recorded_at": generated_at}
            daily = blk.get("daily") if isinstance(blk.get("daily"), dict) and blk["daily"].get("rows") else None
            # several blocks of one account (an X account's counts and its posts) merge on the key
            existing = next((x for x in accounts if x["key"] == a.get("key")), None)
            acct = existing or {"key": a.get("key"), "platform": platform, "handle": a.get("handle"), "kind": a.get("kind") or "company", "status": status, "daily_span": {}}
            if a.get("name"):
                acct["name"] = a["name"]
            if daily:
                merged = {r["date"]: r for r in rows_of(acct.get("daily"))} if acct.get("daily") else {}
                for r in rows_of(daily):
                    merged.setdefault(r["date"], {"date": r["date"]}).update({k: v for k, v in r.items() if v is not None})
                acct["daily"] = block_of([merged[d] for d in sorted(merged)], ["date"])
                for f in acct["daily"]["fields"][1:]:
                    dates = [r["date"] for r in rows_of(acct["daily"]) if r.get(f) is not None]
                    if dates:
                        acct["daily_span"][f] = [dates[0], dates[-1]]
                p = platforms.setdefault(platform, {"first_date": None, "last_date": None})
                ds = sorted(merged)
                if ds:
                    p["first_date"] = min(x for x in (p["first_date"], ds[0]) if x)
                    p["last_date"] = max(x for x in (p["last_date"], ds[-1]) if x)
            if isinstance(blk.get("posts"), dict) and blk["posts"].get("rows"):
                acct["posts"] = blk["posts"]
                acct["headline_metric_field"] = blk.get("headline_metric")
            if isinstance(blk.get("post_daily"), dict) and blk["post_daily"]:
                acct["post_daily"] = blk["post_daily"]
            if blk.get("paid_split"):
                acct["paid_split"] = blk["paid_split"]
            for b in listed(blk.get("breakdowns")):
                if platform == "youtube":
                    breakdowns["youtube"].append(b)
                else:
                    breakdowns.setdefault(platform, {}).setdefault(acct["key"], {})[b.get("key") or "breakdown"] = b
            if status.get("status") and status["status"] != "ok":
                gaps.append({"account_key": acct["key"], "scope": "account", "status": status["status"], "reason": status.get("reason") or "", "detail": ""})
            if not existing:
                accounts.append(acct)
        # the headline metric, its median over the window, and each post's lift
        for acct in accounts:
            posts = rows_of(acct.get("posts"))
            cands = [("lifetime_" + acct["headline_metric_field"]) if acct.get("headline_metric_field") and not str(acct["headline_metric_field"]).startswith(("lifetime_", "range_")) else acct.get("headline_metric_field")] if acct.get("headline_metric_field") else []
            cands += HEADLINE.get(acct["platform"], [])
            field = next((f for f in cands if f and any(p.get(f) is not None for p in posts)), None)
            acct.pop("headline_metric_field", None)
            if not posts:
                acct.pop("posts", None)
                continue
            if not field:
                acct["posts"] = block_of(posts, ["post_key", "url", "title", "text", "media_type", "published_at", "first_seen_at", "thumbnail"])
                continue
            acct["headline_metric"] = re.sub(r"^(lifetime|range)_", "", field)
            for p in posts:
                p["publish_day"] = day(p.get("published_at"))
                p["headline"] = p.get(field)
                p["sponsored"] = bool(p.get("sponsored") or p.get("paid_delivery") == "sponsored")
            inwin = [p["headline"] for p in posts if p.get("publish_day") and p["publish_day"] >= window_start and p.get("headline") is not None]
            med = statistics.median(inwin) if inwin else None
            acct["headline_median"] = med
            acct["headline_n"] = len(inwin)
            acct["lift_window"] = {"start": window_start, "end": today}
            ok = med and len(inwin) >= SET["lift_min_posts"]
            for p in posts:
                older = bool(p.get("publish_day") and p["publish_day"] < window_start)
                p["lift"] = (round(p["headline"] / med, 4) if ok and p.get("headline") is not None and med else None)
                p["lift_basis"] = "older" if older else "window"
            acct["posts"] = block_of(posts, ["post_key", "url", "title", "text", "media_type", "published_at", "first_seen_at", "thumbnail"])
        if not accounts:
            warnings.append("No platform answered: the brand has no connection, or every read failed.")
        out.update({"schema_version": 1, "generated_at": generated_at, "settings": SET,
                    "metrics": {p: {m: {"unit": u, "period_kind": k, "description": d, "additive": u not in ("percent", "ratio", "seconds", "minutes", "ms")} for m, (u, k, d) in legend.items()} for p, legend in LEGEND.items()},
                    "platforms": platforms, "accounts": accounts, "breakdowns": breakdowns, "flags": flags, "gaps": gaps, "annotations": [], "warnings": warnings,
                    "client": inp.get("client") or "", "count": len(accounts)})
        return out


result = SocialSnapshotBuild().run(inp)

# Output collection (the runner reads it here)
out = __ivy_ctx__["nodes"][__ivy_node_id__]["output"]
out.update(result)

Tests

Requires: python:3.9

• lift

  Instagram's daily rows and posts make one account whose median over the year is 600 from three posts; the unreadable Facebook Page is a gap.

• no-platform

  No platform answering is a document with no account and a warning that says why.

• bad-date

  A generated_at that is not an ISO date is refused rather than guessed.