How to monitor mentions of your brand and reply

// watch.mjs: run from cron. Prints mentions not seen before and remembers them.
import { existsSync, readFileSync, writeFileSync } from "node:fs";

const api = "https://api.buzztake.com/v1";
const accountId = process.env.ACCOUNT_ID;
const headers = {
  Authorization: `Bearer ${process.env.BUZZTAKE_API_KEY}`,
  "Content-Type": "application/json",
};
const SEEN_FILE = "seen.json";
const seen = new Set(existsSync(SEEN_FILE) ? JSON.parse(readFileSync(SEEN_FILE, "utf8")) : []);

async function read(body) {
  const res = await fetch(`${api}/accounts/${accountId}/actions`, {
    method: "POST",
    headers,
    body: JSON.stringify({ ...body, wait: 90 }),
  });
  if (!res.ok) throw new Error(`${res.status}: ${await res.text()}`);
  const action = await res.json();
  if (action.status !== "succeeded") throw new Error(`${body.type}: ${action.status}`);
  return action.result.items;
}

const notifications = await read({ type: "read_notifications", limit: 50 });
const posts = await read({ type: "search", query: "acme", kind: "posts", limit: 50 });

const mentions = [
  ...notifications
    .filter((n) => ["mention", "reply", "quote", "comment", "tag"].includes(n.kind))
    .map((n) => ({ key: n.url ?? `${n.actor}|${n.at}|${n.text}`, url: n.url, text: n.text })),
  ...posts
    .filter((p) => p.author_handle !== "acme_demo")
    .map((p) => ({ key: p.url ?? p.id, url: p.url, text: p.text })),
];

for (const mention of mentions) {
  if (!mention.key || seen.has(mention.key)) continue;
  seen.add(mention.key);
  console.log(mention.url ?? "(no url)", mention.text);
}
writeFileSync(SEEN_FILE, JSON.stringify([...seen]));

# watch.py: run from cron. Prints mentions not seen before and remembers them.
import json
import os
import requests

api = "https://api.buzztake.com/v1"
account_id = os.environ["ACCOUNT_ID"]
headers = {"Authorization": f"Bearer {os.environ['BUZZTAKE_API_KEY']}"}
SEEN_FILE = "seen.json"
seen = set(json.load(open(SEEN_FILE))) if os.path.exists(SEEN_FILE) else set()


def read(body):
    resp = requests.post(
        f"{api}/accounts/{account_id}/actions",
        headers=headers,
        json={**body, "wait": 90},
        timeout=100,
    )
    if not resp.ok:
        raise SystemExit(f"{resp.status_code}: {resp.text}")
    action = resp.json()
    if action["status"] != "succeeded":
        raise SystemExit(f"{body['type']}: {action['status']}")
    return action["result"]["items"]


notifications = read({"type": "read_notifications", "limit": 50})
posts = read({"type": "search", "query": "acme", "kind": "posts", "limit": 50})

mentions = [
    {"key": n["url"] or f"{n['actor']}|{n['at']}|{n['text']}", "url": n["url"], "text": n["text"]}
    for n in notifications
    if n.get("kind") in ("mention", "reply", "quote", "comment", "tag")
] + [
    {"key": p["url"] or p["id"], "url": p["url"], "text": p["text"]}
    for p in posts
    if p.get("author_handle") != "acme_demo"
]

for mention in mentions:
    if not mention["key"] or mention["key"] in seen:
        continue
    seen.add(mention["key"])
    print(mention["url"] or "(no url)", mention["text"])

with open(SEEN_FILE, "w") as f:
    json.dump(sorted(seen), f)