← All ivy-nodes
IVYXSTUDIO · IVY NODE
X
X: recent posts that match a search, as mentions
ivy.node.x-search · v0.1.0
ivyx✓
Searches the last seven days of X for a query (keywords, a brand name, a hashtag) through the X API v2 with an X connection, as the mentions IVYX Social's listening shows: the post, who wrote it, when, and its impressions as reach. Recent search is a paid-tier endpoint with a monthly cap, so the query stays narrow. As a trigger's read it resumes from the newest post's id.
#x#twitter#search#listening#social
Inputs
| Field | Type | Description |
|---|---|---|
| connectionrequired | Not declared | The X account as the stored secret connection.<id>, which IVYX Flows keeps and any other host reads as an ordinary stored secret of that name. |
| queryrequired | string | X's search query: ivyx OR (net ranks) lang:tr. |
| name | string | The listening query's name in IVYX Social, which each mention names. |
| limit | integer | At most this many, 100 at most. |
| include_reposts | boolean | Keep reposts. |
| $cursor | string | Where the last read ended, handed in by a trigger that resumes (the service's poll cursor); absent on a first read. |
| $start | string | On a trigger's first read: now (only what comes next) or all (what is there). |
Outputs
| Field | Type | Description |
|---|---|---|
| mentionsrequired | array | Mentions: id, query, source, url, author, text, at (ms), reach. |
| countrequired | integer | Not declared |
| next_cursor | string | Not declared |
| statusrequired | object | Not declared |
| skippedrequired | boolean | Not declared |
Source
python
# Input preparation
inp = __ivy_ctx__["nodes"][__ivy_node_id__]["input"]
# Compute
# ivy_connect, embedded from IVYX Flows' connectors/sdk by tools/author-connector-nodes.py; never edited here.
import sys as _sys, types as _types
_IVY_CONNECT = {'errors': '"""The errors a connector node raises, one per thing a person would do about it."""\n\nfrom typing import Any, Optional\n\n\nclass ConnectorError(Exception):\n """An app answered with an error, or could not be reached. ``status`` is 0 when nothing answered."""\n\n def __init__(self, message: str, status: int = 0, body: Any = None):\n super().__init__(message)\n self.status = status\n self.body = body\n\n\nclass AuthError(ConnectorError):\n """The app refused the credential (401/403): the connection needs a look, or a new sign-in."""\n\n\nclass NotFound(ConnectorError):\n """404: what the step asked for is not there."""\n\n\nclass RateLimited(ConnectorError):\n """429 after every retry: the app wants fewer calls."""\n\n\nclass Invalid(ConnectorError):\n """400, 409 or 422: the app refused what the step sent, and says why."""\n\n\nclass Unavailable(ConnectorError):\n """5xx after every retry, a timeout, or no answer at all."""\n\n\ndef error_for(status: int, message: str, body: Any = None) -> ConnectorError:\n if status in (401, 403):\n return AuthError(message, status, body)\n if status == 404:\n return NotFound(message, status, body)\n if status == 429:\n return RateLimited(message, status, body)\n if status in (400, 409, 410, 412, 415, 422):\n return Invalid(message, status, body)\n if status == 0 or status >= 500:\n return Unavailable(message, status, body)\n return ConnectorError(message, status, body)\n\n\ndef app_message(body: Any) -> Optional[str]:\n """The app\'s own words for an error, from the shapes APIs answer with."""\n if not isinstance(body, dict):\n return None\n err = body.get("error")\n if isinstance(err, dict) and isinstance(err.get("message"), str):\n return err["message"][:300]\n for key in ("error_description", "message", "error", "detail"):\n if isinstance(body.get(key), str) and body[key]:\n return body[key][:300]\n errors = body.get("errors")\n if isinstance(errors, list) and errors and isinstance(errors[0], dict) and isinstance(errors[0].get("message"), str):\n return errors[0]["message"][:300]\n return None\n', 'client': '"""The client a connector node calls its app with."""\n\nimport email.utils\nimport json as jsonlib\nimport os\nimport random\nimport threading\nimport time\nimport urllib.error\nimport urllib.parse\nimport urllib.request\nfrom typing import Any, Callable, Dict, Iterator, List, Optional, Union\n\nfrom .errors import AuthError, ConnectorError, Unavailable, app_message, error_for\n\nIDEMPOTENT = ("GET", "HEAD", "OPTIONS", "PUT", "DELETE")\nRETRY_STATUS = (500, 502, 503, 504)\nMAX_WAIT = 60.0\n\n\nclass Response:\n """What an app answered: the status, the headers (lower-case names), and the body as text, bytes or JSON."""\n\n def __init__(self, status: int, headers: Dict[str, str], content: bytes, url: str):\n self.status = status\n self.headers = headers\n self.content = content\n self.url = url\n\n @property\n def ok(self) -> bool:\n return self.status < 400\n\n @property\n def text(self) -> str:\n return self.content.decode("utf-8", errors="replace")\n\n def json(self) -> Any:\n if not self.content.strip():\n return None\n return jsonlib.loads(self.text)\n\n def maybe_json(self) -> Any:\n try:\n return self.json()\n except ValueError:\n return None\n\n\nclass _Bucket:\n """At most ``per_minute`` calls a minute per connection, shared by every client of it in this process."""\n\n def __init__(self, per_minute: float):\n self.rate = per_minute / 60.0\n self.capacity = max(1.0, min(per_minute, 10.0))\n self.tokens = self.capacity\n self.at = time.monotonic()\n self.lock = threading.Lock()\n\n def take(self, sleep: Callable[[float], None]) -> None:\n with self.lock:\n now = time.monotonic()\n self.tokens = min(self.capacity, self.tokens + (now - self.at) * self.rate)\n self.at = now\n wait = 0.0 if self.tokens >= 1 else (1 - self.tokens) / self.rate\n self.tokens -= 1\n if wait > 0:\n sleep(wait)\n\n\n_buckets: Dict[str, _Bucket] = {}\n\n\ndef _dig(value: Any, path: str) -> Any:\n for key in path.split("."):\n if isinstance(value, dict):\n value = value.get(key)\n elif isinstance(value, list) and key.isdigit() and int(key) < len(value):\n value = value[int(key)]\n else:\n return None\n return value\n\n\ndef _next_link(header: str) -> Optional[str]:\n """The ``rel="next"`` URL of a Link header, as GitHub and others page with."""\n for part in header.split(","):\n bits = part.split(";")\n if len(bits) > 1 and any(b.strip().replace(" ", "") in (\'rel="next"\', "rel=next") for b in bits[1:]):\n return bits[0].strip().strip("<>")\n return None\n\n\ndef _retry_after(value: Optional[str]) -> Optional[float]:\n if not value:\n return None\n try:\n return max(0.0, float(value))\n except ValueError:\n pass\n try:\n when = email.utils.parsedate_to_datetime(value)\n return max(0.0, when.timestamp() - time.time())\n except (TypeError, ValueError):\n return None\n\n\nclass Client:\n """\n A connection\'s client. Every request carries the connection\'s headers and\n query; a path is under its base URL, a full URL is used as it is.\n """\n\n def __init__(\n self,\n credential: Dict[str, Any],\n timeout: float = 30.0,\n retries: int = 4,\n broker_url: Optional[str] = None,\n broker_key: Optional[str] = None,\n sleep: Callable[[float], None] = time.sleep,\n base: Optional[str] = None,\n ):\n if not isinstance(credential, dict) or not isinstance(credential.get("headers"), dict):\n raise ConnectorError(\'The connection input is not a connection; give the step {"$secret": "connection.<id>"}.\')\n self.credential = credential\n # One of the connector\'s other roots (``bases``), for an app spread over hosts: Google\'s gmail. and sheets.\n self.base = base\n if base is not None and base not in (credential.get("bases") or {}):\n raise ConnectorError(f"The connection {credential.get(\'id\')} has no base called {base}; its connector names {\', \'.join(sorted(credential.get(\'bases\') or {})) or \'none\'}.")\n self.timeout = timeout\n self.retries = retries\n self.sleep = sleep\n self.broker_url = broker_url if broker_url is not None else os.environ.get("IVYX_CONNECTIONS_URL", "")\n self.broker_key = broker_key if broker_key is not None else os.environ.get("IVYX_CONNECTIONS_KEY", "")\n per_minute = (credential.get("rateLimit") or {}).get("perMinute")\n self._bucket = None\n if per_minute:\n self._bucket = _buckets.setdefault(str(credential.get("id")), _Bucket(float(per_minute)))\n\n # ---------- who and where ----------\n\n @property\n def name(self) -> str:\n return str(self.credential.get("connector") or self.credential.get("id") or "the app")\n\n @property\n def base_url(self) -> str:\n if self.base is not None:\n return str((self.credential.get("bases") or {}).get(self.base) or "")\n return str(self.credential.get("baseUrl") or "")\n\n @property\n def account(self) -> str:\n """Who the connection signs in as, from its test: an address, a workspace."""\n return str(self.credential.get("account") or "")\n\n def url(self, path: str) -> str:\n if path.lower().startswith(("http://", "https://")):\n return path\n if not self.base_url:\n raise ConnectorError(f"{path} is relative and the connection {self.credential.get(\'id\')} has no base URL; give a full URL.")\n return self.base_url.rstrip("/") + "/" + path.lstrip("/")\n\n # ---------- tokens ----------\n\n def _secrets(self) -> List[str]:\n """Every value that signs in, so no message ever carries one."""\n out = [str(v) for v in (self.credential.get("values") or {}).values() if v]\n out.extend(str(t.get("access_token")) for t in (self.credential.get("tokens") or {}).values() if isinstance(t, dict) and t.get("access_token"))\n for v in (self.credential.get("headers") or {}).values():\n out.extend(p for p in str(v).split(" ") if len(p) > 8)\n return [s for s in out if len(s) >= 6]\n\n def redact(self, text: str) -> str:\n for s in self._secrets():\n text = text.replace(s, "[REDACTED]")\n return text\n\n def refresh(self) -> bool:\n """Ask the broker for a fresh credential; False when there is no broker to ask."""\n if not self.broker_url or not self.broker_key or not self.credential.get("id"):\n return False\n req = urllib.request.Request(\n f"{self.broker_url.rstrip(\'/\')}/{urllib.parse.quote(str(self.credential[\'id\']))}/credentials?fresh=1",\n headers={"Authorization": f"Bearer {self.broker_key}", "Accept": "application/json"},\n )\n try:\n with urllib.request.urlopen(req, timeout=30) as res:\n fresh = jsonlib.loads(res.read().decode("utf-8"))\n except urllib.error.HTTPError as err:\n with err:\n said = app_message(_loads(err.read())) or f"the broker answered {err.code}"\n raise AuthError(f"The connection {self.credential.get(\'id\')} could not be refreshed: {said}", err.code)\n except (urllib.error.URLError, OSError, ValueError) as err:\n raise Unavailable(f"The connections broker did not answer: {err}")\n if isinstance(fresh, dict) and isinstance(fresh.get("headers"), dict):\n self.credential = fresh\n return True\n return False\n\n def token(self, audience: Optional[str] = None) -> str:\n """\n The access token, for a protocol that is not HTTP (IMAP\'s XOAUTH2).\n ``audience`` names a token for another resource the connector asks\n for (Microsoft\'s ``imap``), handed out beside the main one.\n """\n self._ahead()\n if audience is None:\n values = self.credential.get("values") or {}\n found = values.get("access_token") or values.get("token")\n if not found:\n raise AuthError(f"The connection {self.credential.get(\'id\')} signs in with {self.credential.get(\'type\')}, not a token.")\n return str(found)\n got = (self.credential.get("tokens") or {}).get(audience)\n if not isinstance(got, dict) or not got.get("access_token"):\n raise AuthError(f"The connection {self.credential.get(\'id\')} has no token for {audience}: sign in to it again, so the app is asked for that access too.")\n return str(got["access_token"])\n\n def _ahead(self) -> None:\n expires = self.credential.get("expiresAt")\n if self.credential.get("type") == "oauth2" and isinstance(expires, (int, float)) and expires / 1000.0 - time.time() < 60:\n self.refresh()\n\n # ---------- requests ----------\n\n def request(\n self,\n method: str,\n path: str,\n params: Optional[Dict[str, Any]] = None,\n json: Any = None,\n data: Union[bytes, str, Dict[str, Any], None] = None,\n headers: Optional[Dict[str, str]] = None,\n timeout: Optional[float] = None,\n retry_unsafe: bool = False,\n check: bool = True,\n ) -> Response:\n """\n One call, tried again on 429 always and on 5xx or no answer when the\n method is safe to repeat (or ``retry_unsafe``), waiting as the app\n asks or longer each time. A refused token is refreshed once. With\n ``check`` an error status raises, else the response comes back as it is.\n """\n method = method.upper()\n self._ahead()\n repeatable = method in IDEMPOTENT or retry_unsafe\n refreshed = False\n attempt = 0\n while True:\n if self._bucket:\n self._bucket.take(self.sleep)\n try:\n res = self._send(method, path, params, json, data, headers, timeout)\n except Unavailable:\n if repeatable and attempt < self.retries:\n self.sleep(self._backoff(attempt))\n attempt += 1\n continue\n raise\n if res.status == 401 and not refreshed and self.credential.get("type") == "oauth2" and self.refresh():\n refreshed = True\n continue\n if attempt < self.retries and (res.status == 429 or (res.status in RETRY_STATUS and repeatable)):\n wait = _retry_after(res.headers.get("retry-after"))\n self.sleep(min(MAX_WAIT, wait if wait is not None else self._backoff(attempt)))\n attempt += 1\n continue\n if check and not res.ok:\n raise self._error(method, path, res)\n return res\n\n def _backoff(self, attempt: int) -> float:\n return min(30.0, 0.5 * (2 ** attempt)) + random.uniform(0, 0.25)\n\n def _send(self, method, path, params, json, data, headers, timeout) -> Response:\n url = self.url(path)\n query = dict(self.credential.get("query") or {})\n query.update({k: v for k, v in (params or {}).items() if v is not None})\n if query:\n sep = "&" if urllib.parse.urlsplit(url).query else "?"\n url = url + sep + urllib.parse.urlencode(query, doseq=True)\n sent = {"Accept": "application/json"}\n sent.update(self.credential.get("headers") or {})\n body = None\n if json is not None and data is not None:\n raise ConnectorError("Give json or data, not both.")\n if json is not None:\n body = jsonlib.dumps(json).encode("utf-8")\n sent["Content-Type"] = "application/json"\n elif isinstance(data, dict):\n body = urllib.parse.urlencode(data, doseq=True).encode("utf-8")\n sent["Content-Type"] = "application/x-www-form-urlencoded"\n elif isinstance(data, str):\n body = data.encode("utf-8")\n elif data is not None:\n body = data\n sent.update(headers or {})\n req = urllib.request.Request(url, data=body, method=method, headers=sent)\n try:\n with urllib.request.urlopen(req, timeout=timeout or self.timeout) as res:\n return Response(res.status, {k.lower(): v for k, v in res.headers.items()}, res.read(), res.geturl())\n except urllib.error.HTTPError as err:\n with err:\n return Response(err.code, {k.lower(): v for k, v in (err.headers or {}).items()}, err.read() or b"", url)\n except (urllib.error.URLError, OSError) as err:\n reason = getattr(err, "reason", err)\n raise Unavailable(self.redact(f"{self.name} could not be reached at {self.url(path)}: {reason}"))\n\n def _error(self, method: str, path: str, res: Response) -> ConnectorError:\n body = res.maybe_json()\n said = app_message(body) or res.text.strip()[:200] or "no reason given"\n where = path if not path.lower().startswith("http") else urllib.parse.urlsplit(path).path\n return error_for(res.status, self.redact(f"{res.status} from {self.name} {method} {where}: {said}"), body)\n\n def get(self, path: str, **kw) -> Response:\n return self.request("GET", path, **kw)\n\n def post(self, path: str, **kw) -> Response:\n return self.request("POST", path, **kw)\n\n def put(self, path: str, **kw) -> Response:\n return self.request("PUT", path, **kw)\n\n def patch(self, path: str, **kw) -> Response:\n return self.request("PATCH", path, **kw)\n\n def delete(self, path: str, **kw) -> Response:\n return self.request("DELETE", path, **kw)\n\n # ---------- lists ----------\n\n def paginate(\n self,\n method: str,\n path: str,\n items: str,\n style: str = "cursor",\n next: Optional[str] = None,\n param: Optional[str] = None,\n params: Optional[Dict[str, Any]] = None,\n json: Any = None,\n page_size: Optional[int] = None,\n size_param: str = "limit",\n limit: Optional[int] = None,\n max_pages: int = 1000,\n ) -> Iterator[Any]:\n """\n Every item of a list the app hands out in pages. ``items`` is the dot\n path to a page\'s list. ``style`` is how the next page is asked for:\n\n - ``cursor``: ``next`` is the dot path to the next cursor in a page,\n sent back as ``param`` (``cursor`` by default; in the JSON body of a\n POST when there is one). An empty cursor ends it.\n - ``link``: the ``Link`` header\'s ``rel="next"`` URL.\n - ``offset``: ``param`` (``offset``) grows by the page\'s length; a page\n shorter than ``page_size`` ends it.\n - ``page``: ``param`` (``page``) counts up from 1 until a page is empty.\n """\n params = dict(params or {})\n if page_size:\n params[size_param] = page_size\n body = dict(json) if isinstance(json, dict) else json\n given = 0\n url = path\n position: Any = None\n if style == "offset":\n param = param or "offset"\n position = int(params.get(param, 0))\n elif style == "page":\n param = param or "page"\n position = int(params.get(param, 1))\n elif style == "cursor":\n param = param or "cursor"\n if not next:\n raise ConnectorError("A cursor page names where its next cursor is: next=\'…\'.")\n elif style != "link":\n raise ConnectorError(f"A page style is cursor, link, offset or page, not {style}.")\n for _ in range(max_pages):\n if style in ("offset", "page"):\n params[param] = position\n res = self.request(method, url, params=params, json=body)\n page = res.json()\n found = _dig(page, items)\n if not isinstance(found, list):\n raise ConnectorError(f"{self.name} answered a page with no list at {items}.")\n for item in found:\n yield item\n given += 1\n if limit is not None and given >= limit:\n return\n if style == "cursor":\n cursor = _dig(page, next)\n if not cursor:\n return\n if isinstance(body, dict) and method.upper() != "GET":\n body[param] = cursor\n else:\n params[param] = cursor\n elif style == "link":\n link = _next_link(res.headers.get("link", ""))\n if not link:\n return\n url, params = link, {}\n elif style == "offset":\n if not found or (page_size and len(found) < page_size):\n return\n position += len(found)\n else:\n if not found:\n return\n position += 1\n\n # ---------- files ----------\n\n def download(self, path: str, dest: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:\n """A file from the app into the workspace, at ``dest``; folders on the way are made."""\n res = self.request("GET", path, params=params, headers={"Accept": "*/*"})\n folder = os.path.dirname(dest)\n if folder:\n os.makedirs(folder, exist_ok=True)\n with open(dest, "wb") as f:\n f.write(res.content)\n return {"path": dest, "bytes": len(res.content), "content_type": res.headers.get("content-type", "")}\n\n def upload(self, path: str, src: str, method: str = "POST", content_type: str = "application/octet-stream", params: Optional[Dict[str, Any]] = None) -> Response:\n """A workspace file\'s bytes as the body of one request."""\n with open(src, "rb") as f:\n content = f.read()\n return self.request(method, path, params=params, data=content, headers={"Content-Type": content_type})\n\n\ndef _loads(raw: bytes) -> Any:\n try:\n return jsonlib.loads(raw.decode("utf-8"))\n except ValueError:\n return None\n\n\ndef xoauth2(user: str, token: str) -> bytes:\n """The SASL XOAUTH2 string IMAP and SMTP sign in with: ``imap.authenticate("XOAUTH2", lambda _: xoauth2(user, token))``."""\n return f"user={user}\\x01auth=Bearer {token}\\x01\\x01".encode("utf-8")\n\n\ndef client(connection: Union[str, Dict[str, Any]], **kw) -> Client:\n """The client for a step\'s connection input: the credential\'s JSON text as the service resolved it, or the object."""\n if isinstance(connection, str):\n try:\n connection = jsonlib.loads(connection)\n except ValueError:\n raise ConnectorError(\'The connection input is not a connection; give the step {"$secret": "connection.<id>"}.\')\n return Client(connection, **kw)\n', '__init__': '"""\nivy_connect: what every connector node shares (docs/connectors.md, module 4).\n\nA node gets its connection as a stored secret, ``{"$secret": "connection.<id>"}``,\nwhich the service resolves when the step starts into the credential\'s JSON:\nthe headers and query to send, the base URL, the raw values. This library\nturns that into a client that signs every request, waits and tries again when\nthe app is busy, pages through lists, says errors in the app\'s own words, and\nnever writes a token into a message.\n\n from ivy_connect import client\n\n slack = client(inp["connection"])\n for ch in slack.paginate("GET", "conversations.list", items="channels", next="response_metadata.next_cursor"):\n ...\n slack.post("chat.postMessage", json={"channel": inp["channel"], "text": inp["text"]})\n\nAn app spread over hosts names its other roots in the manifest\'s ``bases``;\n``client(inp["connection"], base="sheets")`` calls under one of them. A\nprotocol that is not HTTP takes ``c.token()`` (or ``c.token("imap")`` for a\ntoken of another resource) and ``xoauth2(user, token)``.\n\nThe standard library only, and Python 3.9, like the nodes it runs in.\n"""\n\nfrom .client import Client, Response, client, xoauth2\nfrom .errors import AuthError, ConnectorError, Invalid, NotFound, RateLimited, Unavailable\n\n__all__ = ["client", "xoauth2", "Client", "Response", "ConnectorError", "AuthError", "NotFound", "RateLimited", "Invalid", "Unavailable"]\n__version__ = "0.2.0"\n'}
if getattr(_sys.modules.get('ivy_connect'), '__embedded__', None) != 'd8e4d4dcdcae1a17':
_pkg = _types.ModuleType('ivy_connect')
_pkg.__path__ = []
_pkg.__package__ = 'ivy_connect'
_sys.modules['ivy_connect'] = _pkg
for _name in ('errors', 'client'):
_mod = _types.ModuleType('ivy_connect.' + _name)
_mod.__package__ = 'ivy_connect'
_sys.modules['ivy_connect.' + _name] = _mod
exec(compile(_IVY_CONNECT[_name], '<ivy_connect/' + _name + '.py>', 'exec'), _mod.__dict__)
setattr(_pkg, _name, _mod)
exec(compile(_IVY_CONNECT['__init__'], '<ivy_connect/__init__.py>', 'exec'), _pkg.__dict__)
_pkg.__embedded__ = 'd8e4d4dcdcae1a17'
del _IVY_CONNECT
_EMPTY = {'mentions': [], 'count': 0}
import datetime, json, re, time
from ivy_connect import client, ConnectorError
def day(s):
s = str(s or "")[:10]
return s if re.fullmatch(r"\d{4}-\d{2}-\d{2}", s) else None
def iso_day(ts):
return datetime.datetime.utcfromtimestamp(ts).strftime("%Y-%m-%d")
def today():
return datetime.datetime.utcnow().strftime("%Y-%m-%d")
def days_back(n):
return (datetime.datetime.utcnow() - datetime.timedelta(days=n)).strftime("%Y-%m-%d")
def ms_of(s):
"""An ISO stamp (Z or +0000) or epoch seconds to milliseconds."""
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 table(rows_by_day, first=("date",)):
"""{day: {field: value}} to the collector's block, fields in first-seen order after the leading ones."""
fields = list(first)
for d in sorted(rows_by_day):
for k in rows_by_day[d]:
if k not in fields:
fields.append(k)
rows = [[d] + [rows_by_day[d].get(k) for k in fields[1:]] for d in sorted(rows_by_day)]
return {"fields": fields, "rows": rows}
def posts_block(items, first):
fields = list(first)
for it in items:
for k in it:
if k not in fields:
fields.append(k)
return {"fields": fields, "rows": [[it.get(k) for k in fields] for it in items]}
def skipped_answer(out):
out.update({k: v for k, v in _EMPTY.items() if k not in out})
out["skipped"] = True
out["status"] = {"status": "no_access", "reason": "no connection named", "recorded_at": datetime.datetime.utcnow().isoformat() + "Z"}
def done(out, **kw):
out["skipped"] = False
out["status"] = {"status": kw.pop("status", "ok"), "recorded_at": datetime.datetime.utcnow().isoformat() + "Z", **({"reason": kw.pop("reason")} if "reason" in kw else {})}
out.update(kw)
class XSearch:
"X: recent posts that match a search, as mentions"
def run(self, inp):
out = {}
if not inp.get("connection"):
__skipped__ = True
else:
__skipped__ = False
if __skipped__:
skipped_answer(out)
out["mentions"] = []
out["count"] = 0
else:
x = client(inp["connection"], base="v2")
q = str(inp.get("query") or "").strip()
if not q:
raise ValueError("query is empty.")
params = {"query": q + (" -is:retweet" if not inp.get("include_reposts") else ""), "max_results": max(10, min(int(inp.get("limit") or 50), 100)), "tweet.fields": "created_at,author_id,public_metrics", "expansions": "author_id", "user.fields": "username,public_metrics"}
if inp.get("$cursor"):
params["since_id"] = inp["$cursor"]
data = x.get("tweets/search/recent", params=params).json()
users = {u["id"]: u for u in ((data.get("includes") or {}).get("users") or [])}
name = str(inp.get("name") or q)
rows = []
for t in data.get("data") or []:
a = users.get(t.get("author_id")) or {}
pm = t.get("public_metrics") or {}
rows.append({"id": "x:" + t["id"], "query": name, "source": "x", "url": f"https://x.com/i/web/status/{t['id']}", "author": a.get("username") or t.get("author_id"), "text": t.get("text"), "at": ms_of(t.get("created_at")), "reach": pm.get("impression_count") or (a.get("public_metrics") or {}).get("followers_count")})
done(out, mentions=rows, count=len(rows), next_cursor=(data.get("meta") or {}).get("newest_id") or inp.get("$cursor") or "")
return out
result = XSearch().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
- one
One matching post is one mention with its impressions as reach.
- no-connection
An empty connection skips the platform and says so, so an agent can wire every platform and run the ones a brand has.
- refused
A credential the app refuses is said with the app's status, never as an empty answer.