Social listening

The problem#

Conversations about a brand or product are spread over networks, forums and news sites, each with its own API, auth and field names. Watching them means one client per source and a merge step that breaks whenever one of them changes.

Onekey gives every source the same call shape and the same fields, so a listening job is a loop over endpoint names.

Endpoints to call#

Each endpoint name links to its reference page with every parameter, the caching window and a recorded response.

Your first call#

This is x.search (2 credits). Replace YOUR_API_KEY with your key; sign up, no card. 1,000 free credits cover the open-platform endpoints. Hard-to-reach platforms (X, TikTok, Instagram, Reddit, and others marked on their page) need a verified email and a purchased pack.

curl "https://api.onekey.dev/v1/x/search?q=nasa" \
  -H "x-api-key: YOUR_API_KEY"

A working script#

Search every source for each tracked term, drop duplicates and print the strongest signals. Run it from cron or any scheduler; the API is request and response, so you decide how often it runs.

import os
import requests

API = "https://api.onekey.dev"
HEADERS = {"x-api-key": os.environ["ONEKEY_KEY"]}


def call(path, **params):
    """GET /v1/<platform>/<endpoint>; returns the data or raises with the error code."""
    r = requests.get(f"{API}/v1/{path}", params=params, headers=HEADERS, timeout=30)
    body = r.json()
    if not body["success"]:
        raise RuntimeError(f"{path}: {body['error']['code']}")
    return body["data"]


def get(item, field):
    """get(item, 'content.title'): a field of the unified schema, or None."""
    for key in field.split("."):
        item = (item or {}).get(key)
    return item


TERMS = ["acme", "acme.com"]
SOURCES = ["x/search", "hackernews/search_by_date", "reddit/search_posts", "google/news"]

seen = set()
for term in TERMS:
    for source in SOURCES:
        try:
            items = call(source, q=term)
        except RuntimeError as err:
            print(err)  # failed calls are refunded; carry on with the other sources
            continue
        for it in items:
            key = get(it, "content.url") or get(it, "content.id")
            if key in seen:
                continue
            seen.add(key)
            text = get(it, "content.title") or get(it, "content.text") or ""
            likes = get(it, "engagement.likes") or 0
            print(f"[{source}] {likes:>5}  {text[:100]}")

What a month costs#

Illustrative volume, not a benchmark: 5 tracked terms, each searched twice a day for 30 days on all four sources (300 calls per source).

About 3,000 credits: the Starter pack (5,000 credits, $9) covers it, at $1.80 per 1,000 credits. Prices are in USD (regional pricing is planned); credits never expire. See pricing.

Every figure above assumes every call is a fresh fetch. These endpoints cache for 2 minutes. Free when served from cache (0.2 credit for a cached response over 100 KB). Repeating a call inside that window costs nothing. Invalid, failed and not-found calls are not charged either.

Good to know#

• Results are public posts only. What a search returns differs by platform: engagement counts, for example, are not available on every source, and missing fields come back as null.
• Reddit endpoints are in beta; the other three are stable.

Keep reading#

• Pagination
• Caching
• Unified schema
• Platform pages: X, Hacker News, Reddit, Google