Social: pack the texts as a content draft

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 model's variants as one answer of kind content-draft, keyed by account, each text held to its platform's length.
LIMITS = {"instagram": 2200, "linkedin": 3000, "x": 280, "youtube": 5000, "facebook": 63206, "tiktok": 2200}


class ContentPack:
    "Social: pack the texts as a content draft"

    def run(self, inp):
        out = {}
        plat = {a["key"]: a.get("platform") for a in listed(inp.get("accounts")) if isinstance(a, dict) and a.get("key")}
        spec = inp.get("spec") if isinstance(inp.get("spec"), dict) else {}
        raw = spec.get("variants") if isinstance(spec.get("variants"), (list, dict)) else inp.get("variants")
        variants = {}
        items = raw.items() if isinstance(raw, dict) else [((v or {}).get("account"), v) for v in listed(raw)]
        for key, v in items:
            if not key or not isinstance(v, dict) or not text_of(v.get("text")):
                continue
            limit = LIMITS.get(plat.get(key), 3000)
            # The model writes them as one string of words (copy-brief's schema); a list is taken as it is.
            raw_tags = v.get("hashtags")
            words = re.split(r"[\s,]+", raw_tags) if isinstance(raw_tags, str) else listed(raw_tags)
            tags = [re.sub(r"^#", "", str(t)).strip() for t in words if re.sub(r"^#", "", str(t)).strip()]
            variants[str(key)] = {"text": str(v["text"]).strip()[:limit], **({"hashtags": tags[:30]} if tags else {}), **({"first_comment": str(v["first_comment"]).strip()[:2200]} if text_of(v.get("first_comment")) else {})}
        out.update({"kind": "content-draft", "request_id": inp.get("request_id") or "", "client": inp.get("client") or "", "title": inp.get("title") or spec.get("title") or "", "format": inp.get("format") or "", "pillar": inp.get("pillar") or "", "brief": inp.get("brief") or "", "variants": variants, "count": len(variants)})
        return out


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

• limits

  X's text is cut at 280 characters, the hashtags lose their #, a variant with no account is dropped.

• keyed

  Variants keyed by account are taken as they are, an unknown platform held to 3000 characters.

• tags-as-text

  Hashtags written as one string of words are split into words, the # taken off.

• empty

  No variant is an empty draft of the same kind.