Social: pack the mentions

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 MentionsPack:
    "Social: pack the mentions"

    def run(self, inp):
        out = {}
        # The search nodes' mentions as one answer of kind mentions: the model's sentiments attached by id, the competitors' sizes beside.
        sent = {}
        for s in listed((inp.get("sentiments") or {}).get("items") if isinstance(inp.get("sentiments"), dict) else inp.get("sentiments")):
            if isinstance(s, dict) and s.get("id") and s.get("sentiment") in ("positive", "neutral", "negative"):
                sent[str(s["id"])] = s["sentiment"]
        seen, rows = set(), []
        name = inp.get("name") or ""
        for key in ("mentions", "mentions_a", "mentions_b", "mentions_c"):
            for m in listed(inp.get(key)):
                if not isinstance(m, dict) or not m.get("id") or m["id"] in seen:
                    continue
                seen.add(m["id"])
                m = dict(m)
                m.setdefault("query", name)
                if m["id"] in sent:
                    m["sentiment"] = sent[m["id"]]
                rows.append(m)
        comps = []
        for key in ("competitor_a", "competitor_b", "competitor_c"):
            c = inp.get(key)
            if isinstance(c, dict) and (c.get("account") or {}).get("handle"):
                comps.append({"query": name, "handle": c["account"]["handle"], "platform": c["account"].get("platform") or "x", "followers": c.get("followers"), "posts_per_week": c.get("posts_per_week"), "at": int(datetime.datetime.utcnow().timestamp() * 1000)})
        out.update({"kind": "mentions", "mentions": rows, "competitors": comps, "count": len(rows), "negative": sum(1 for m in rows if m.get("sentiment") == "negative"), "client": inp.get("client") or ""})
        return out


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

• sentiment

  Two nodes' mentions keep the model's sentiment by id, and a competitor's size rides beside them.

• unknown-sentiment

  A sentiment outside positive, neutral and negative is not attached, and a repeated id counts once.

• empty

  No mention is an empty answer of the same kind.