Social: mentions from the web (RSS)

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]

# Mentions from the web: the items of RSS or Atom feeds (Google News' search feed for a query when no feed is given), each one a mention of source web.
import urllib.request, urllib.parse, xml.etree.ElementTree as ET, email.utils, html


class RssMentions:
    "Social: mentions from the web (RSS)"

    def run(self, inp):
        out = {}
        feeds = listed(inp.get("feeds"))
        q = str(inp.get("query") or "").strip()
        if not feeds and q:
            feeds = ["https://news.google.com/rss/search?" + urllib.parse.urlencode({"q": q, "hl": inp.get("lang") or "en", "gl": inp.get("country") or "US", "ceid": f"{inp.get('country') or 'US'}:{inp.get('lang') or 'en'}"})]
        if not feeds:
            raise ValueError("query or feeds is needed.")
        name = inp.get("name") or q
        limit = max(1, min(int(inp.get("limit") or 50), 200))
        # The cursor is the newest item's time as next_cursor wrote it: digits, which ms_of reads as no date at all.
        cur = str(inp.get("$cursor") or "").strip()
        since = int(cur) if cur.isdigit() else (ms_of(cur) or 0)
        rows, newest, errors = [], since, []
        def strip(s):
            return html.unescape(re.sub(r"<[^>]+>", " ", s or "")).strip()
        for url in feeds:
            try:
                req = urllib.request.Request(url, headers={"User-Agent": str(inp.get("user_agent") or "IVYX social-listen/0.1")})
                with urllib.request.urlopen(req, timeout=float(inp.get("timeout") or 20)) as r:
                    root = ET.fromstring(r.read())
            except Exception as e:
                errors.append(f"{url}: {type(e).__name__}: {e}"[:200])
                continue
            ns = {"a": "http://www.w3.org/2005/Atom"}
            items = root.findall(".//item") or root.findall(".//a:entry", ns)
            for it in items:
                def f(tag):
                    el = it.find(tag) if it.find(tag) is not None else it.find("a:" + tag, ns)
                    return (el.text or "").strip() if el is not None and el.text else ""
                link = f("link") or (it.find("a:link", ns).get("href") if it.find("a:link", ns) is not None else "")
                title = strip(f("title"))
                when = f("pubDate") or f("published") or f("updated")
                try:
                    at = int(email.utils.parsedate_to_datetime(when).timestamp() * 1000) if when and "," in when else ms_of(when)
                except Exception:
                    at = ms_of(when)
                src = it.find("source")
                author = (src.text if src is not None and src.text else "") or f("author") or urllib.parse.urlparse(link).netloc
                mid = f("guid") or f("id") or link or title
                if at and at <= since:
                    continue
                newest = max(newest, at or 0)
                rows.append({"id": "web:" + re.sub(r"\W+", "", mid)[-60:], "query": name, "source": "web", "url": link, "author": strip(author), "text": title + ((" — " + strip(f("description"))[:200]) if f("description") else ""), "at": at, "reach": None})
        rows.sort(key=lambda m: m.get("at") or 0, reverse=True)
        out.update({"mentions": rows[:limit], "count": min(len(rows), limit), "next_cursor": str(newest) if newest else "", "errors": errors})
        return out


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

• feed

  A feed's two items become two mentions of source web, newest first, the cursor at the newest.

• cursor

  A cursor at the newer item leaves only what came after it, which is nothing.

• nothing-to-read

  Neither a query nor a feed is refused, since there is nothing to listen to.