Social: lay a publish request out per platform

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]


class PublishPlan:
    "Social: lay a publish request out per platform"

    def run(self, inp):
        out = {}
        # One publish request (a content item with a variant per target) laid out per platform, so the agent's if-branches run the right publisher.
        targets = [t for t in listed(inp.get("targets")) if isinstance(t, dict) and t.get("platform")]
        by = {}
        for t in targets:
            by.setdefault(t["platform"], t)
        def tags(t):
            return [str(x).lstrip("#") for x in listed(t.get("hashtags"))]
        ig = by.get("instagram"); li = by.get("linkedin"); x = by.get("x"); fb = by.get("facebook")
        out["has_instagram"] = bool(ig and ig.get("connection") and ig.get("ig_user_id") and listed(ig.get("media_urls")))
        out["has_linkedin"] = bool(li and li.get("connection") and li.get("author"))
        out["has_x"] = bool(x and x.get("connection"))
        out["ig"] = {"connection": ig.get("connection"), "ig_user_id": ig.get("ig_user_id"), "account": ig.get("account"), "text": ig.get("text") or "", "hashtags": tags(ig), "first_comment": ig.get("first_comment") or "", "media_urls": listed(ig.get("media_urls")), "format": ig.get("format") or inp.get("format") or ""} if ig else {}
        out["li"] = {"connection": li.get("connection"), "author": li.get("author"), "account": li.get("account"), "text": li.get("text") or "", "hashtags": tags(li), "link": li.get("link") or "", "link_title": li.get("link_title") or ""} if li else {}
        out["x"] = {"connection": x.get("connection"), "account": x.get("account"), "handle": x.get("handle") or "", "texts": listed(x.get("texts")) or ([x.get("text")] if x.get("text") else []), "hashtags": tags(x)} if x else {}
        out["content_item"] = inp.get("content_item") or ""
        out["count"] = sum(1 for k in ("has_instagram", "has_linkedin", "has_x") if out[k])
        out["skipped"] = [t.get("platform") for t in targets if t["platform"] not in ("instagram", "linkedin", "x") or not out.get("has_" + t["platform"])]
        return out


result = PublishPlan().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

• two

  X and Instagram with its media are published; YouTube has no publisher and is listed as skipped.

• instagram-without-media

  Instagram without a public media URL cannot publish and is skipped.

• no-targets

  An empty list of targets publishes nothing.