admin/openclaw-newsroom
1
0
Fork 0
mirror of https://github.com/jacob-bd/openclaw-newsroom.git synced 2026-08-14 00:47:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add Xquik Twitter search provider

Browse Source
This commit is contained in:
kriptoburak
2026-05-17 03:54:38 +03:00
parent 32f019f62b
commit 72ac447393
3 changed files with 199 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+3 -2
View File
@@ -127,6 +127,7 @@ OpenClaw Gateway
| `GH_TOKEN` | Recommended | GitHub API (5000 req/h vs 60/h unauthenticated) | GitHub personal access token (free) |
| `TAVILY_API_KEY` | Optional | Tavily web search for breaking news | 1000 queries/month free |
| `TWITTERAPI_IO_KEY` | Optional | twitterapi.io keyword search supplement | Paid (small monthly fee) |
| `XQUIK_API_KEY` | Optional | Xquik keyword search supplement with `--provider xquik` | Requires Xquik API access |
### Optional Tools
- **bird** — Twitter/X CLI tool (for `scan_twitter_ai.sh`). Install: `npm install -g @steipete/bird` or `brew install steipete/tap/bird` — see [bird.fast](https://bird.fast). If not installed, the Twitter bird CLI source is skipped gracefully.
@@ -306,8 +307,8 @@ Scans official AI company accounts, tech reporters/leakers, and CEO accounts usi
- Tier 2: Reporters and leakers (break news first)
- Tier 3: CEOs (context, not breaking news)
### 5. `fetch_twitter_api.py` — twitterapi.io Keyword Search
Supplements bird CLI with keyword-based search. Uses engagement filtering (50+ likes or 5000+ followers) to cut noise. Properly tags tweet-only stories (no external article URL).
### 5. `fetch_twitter_api.py` - X/Twitter API Keyword Search
Supplements bird CLI with keyword-based search. Defaults to twitterapi.io when `TWITTERAPI_IO_KEY` is set; run with `--provider xquik` and `XQUIK_API_KEY` to use Xquik instead. Uses engagement filtering (50+ likes or 5000+ followers) to cut noise. Properly tags tweet-only stories (no external article URL).
### 6. `github_trending.py` — GitHub Trending + Releases
Three strategies:
scripts/fetch_twitter_api.py
+115 -21
View File
@@ -1,19 +1,20 @@
#!/usr/bin/env python3
"""
Fetch AI tweets via twitterapi.io as a supplement to bird CLI.
Fetch AI tweets via an API supplement to bird CLI.
Uses twitterapi.io's search endpoint for keyword-based Twitter searches.
This catches breaking news that might be missed by the bird CLI's
account-based approach.
Uses a provider search endpoint for keyword-based Twitter searches. This
catches breaking news that might be missed by the bird CLI's account-based
approach.
Output: pipe-delimited TITLE|URL|SOURCE format.
Sources tagged with "(tweet)" when no external article URL exists.
Usage:
python3 fetch_twitter_api.py [--max-queries 3]
python3 fetch_twitter_api.py [--max-queries 3] [--provider auto]
Environment:
TWITTERAPI_IO_KEY — required
TWITTERAPI_IO_KEY - used by the default twitterapi.io path
XQUIK_API_KEY - used when --provider xquik is selected
"""
import json
@@ -22,16 +23,21 @@ import re
import sys
import ssl
import argparse
from datetime import datetime, timezone
from urllib.request import Request, urlopen
from urllib.error import HTTPError, URLError
from urllib.error import HTTPError
_SSL_CTX = ssl.create_default_context()
TIMEOUT = 15
API_BASE = "https://api.twitterapi.io/twitter"
TWITTERAPI_BASE = "https://api.twitterapi.io/twitter"
XQUIK_API_BASE = "https://xquik.com/api/v1"
XQUIK_CONTRACT_DATE = "2026-04-29"
PROVIDER_AUTO = "auto"
PROVIDER_TWITTERAPI = "twitterapi"
PROVIDER_XQUIK = "xquik"
PROVIDER_CHOICES = (PROVIDER_AUTO, PROVIDER_TWITTERAPI, PROVIDER_XQUIK)
# Search queries — focused on breaking AI news
# Search queries - focused on breaking AI news
SEARCH_QUERIES = [
'"breaking" (AI OR "artificial intelligence" OR LLM) -is:retweet lang:en',
'(Anthropic OR Claude OR OpenAI) (announce OR launch OR release) -is:retweet lang:en',
@@ -43,7 +49,27 @@ MIN_LIKES = 50
MIN_FOLLOWERS = 5000
def search_twitter(query, api_key, max_results=10):
def resolve_provider(provider_choice, env=os.environ):
"""Resolve an explicit or automatic provider choice to a provider and key."""
if provider_choice == PROVIDER_XQUIK:
return PROVIDER_XQUIK, env.get("XQUIK_API_KEY", "")
if provider_choice == PROVIDER_TWITTERAPI:
return PROVIDER_TWITTERAPI, env.get("TWITTERAPI_IO_KEY", "")
twitterapi_key = env.get("TWITTERAPI_IO_KEY", "")
if twitterapi_key:
return PROVIDER_TWITTERAPI, twitterapi_key
return PROVIDER_XQUIK, env.get("XQUIK_API_KEY", "")
def search_twitter(query, api_key, provider=PROVIDER_TWITTERAPI, max_results=10):
"""Search tweets via the selected provider."""
if provider == PROVIDER_XQUIK:
return search_xquik(query, api_key, max_results=max_results)
return search_twitterapi(query, api_key, max_results=max_results)
def search_twitterapi(query, api_key, max_results=10):
"""Search tweets via twitterapi.io."""
from urllib.parse import urlencode
@@ -51,7 +77,7 @@ def search_twitter(query, api_key, max_results=10):
"query": query,
"queryType": "Latest",
})
url = f"{API_BASE}/tweet/advanced_search?{params}"
url = f"{TWITTERAPI_BASE}/tweet/advanced_search?{params}"
req = Request(url, headers={
"X-API-Key": api_key,
@@ -75,6 +101,67 @@ def search_twitter(query, api_key, max_results=10):
return []
def search_xquik(query, api_key, max_results=10):
"""Search tweets via Xquik."""
from urllib.parse import urlencode
params = urlencode({
"q": query,
"queryType": "Latest",
"limit": max_results,
})
url = f"{XQUIK_API_BASE}/x/tweets/search?{params}"
req = Request(url, headers={
"x-api-key": api_key,
"xquik-api-contract": XQUIK_CONTRACT_DATE,
"User-Agent": "NewsScanner/1.0",
})
try:
with urlopen(req, timeout=TIMEOUT, context=_SSL_CTX) as resp:
data = json.loads(resp.read().decode('utf-8'))
return data.get("tweets", [])
except HTTPError as e:
if e.code == 401:
print(" Error: Invalid XQUIK_API_KEY", file=sys.stderr)
elif e.code == 402:
print(" Warning: Xquik API access required", file=sys.stderr)
elif e.code == 429:
print(" Warning: Xquik rate limit", file=sys.stderr)
else:
print(f" Warning: Xquik HTTP {e.code}", file=sys.stderr)
return []
except Exception as e:
print(f" Warning: Xquik error: {e}", file=sys.stderr)
return []
def _external_url_from_entity(url_entity):
expanded = (
url_entity.get("expanded_url")
or url_entity.get("expandedUrl")
or url_entity.get("url", "")
)
if (
expanded
and "twitter.com" not in expanded
and "t.co" not in expanded
and "x.com" not in expanded
):
return expanded
return ""
def _author_screen_name(author):
return (
author.get("userName")
or author.get("username")
or author.get("screenName")
or ""
)
def extract_url_from_tweet(tweet):
"""
Extract the first external URL from a tweet.
@@ -86,12 +173,12 @@ def extract_url_from_tweet(tweet):
entities = tweet.get("entities", {})
urls = entities.get("urls", [])
for u in urls:
expanded = u.get("expanded_url", u.get("url", ""))
if expanded and "twitter.com" not in expanded and "t.co" not in expanded and "x.com" not in expanded:
expanded = _external_url_from_entity(u)
if expanded:
return expanded, False
author = tweet.get("author", {})
screen_name = author.get("userName", "")
screen_name = _author_screen_name(author)
tweet_id = tweet.get("id", "")
if screen_name and tweet_id:
return f"https://x.com/{screen_name}/status/{tweet_id}", True
@@ -99,14 +186,18 @@ def extract_url_from_tweet(tweet):
def main():
parser = argparse.ArgumentParser(description="Fetch AI tweets via twitterapi.io")
parser = argparse.ArgumentParser(description="Fetch AI tweets via an API provider")
parser.add_argument('--max-queries', type=int, default=3,
help='Max queries to run (default: 3)')
parser.add_argument('--provider', choices=PROVIDER_CHOICES,
default=PROVIDER_AUTO,
help='Tweet search provider (default: auto)')
args = parser.parse_args()
api_key = os.environ.get("TWITTERAPI_IO_KEY", "")
provider, api_key = resolve_provider(args.provider)
if not api_key:
print(" Warning: TWITTERAPI_IO_KEY not set, skipping", file=sys.stderr)
print(" Warning: TWITTERAPI_IO_KEY or XQUIK_API_KEY not set, skipping",
file=sys.stderr)
return 0
seen_urls = set()
@@ -115,7 +206,7 @@ def main():
queries = SEARCH_QUERIES[:args.max_queries]
for query in queries:
tweets = search_twitter(query, api_key)
tweets = search_twitter(query, api_key, provider=provider)
for tweet in tweets:
likes = tweet.get("likeCount", 0)
author = tweet.get("author", {})
@@ -142,14 +233,17 @@ def main():
continue
seen_urls.add(url)
screen_name = author.get("userName", "unknown")
screen_name = _author_screen_name(author) or "unknown"
source_tag = f"X/@{screen_name} (tweet)" if is_tweet_only else f"X/@{screen_name}"
all_results.append(f"{title}|{url}|{source_tag}")
for line in all_results:
print(line)
print(f" Done: {len(all_results)} tweets from {len(queries)} queries", file=sys.stderr)
print(
f" Done: {len(all_results)} tweets from {len(queries)} queries via {provider}",
file=sys.stderr,
)
if __name__ == "__main__":
scripts/test_components.py
+81
View File
@@ -271,6 +271,87 @@ test("AI as standalone word - should match",
test("API standalone - should match",
is_ai("REST API design best practices for developers"))
# ==============================================================
print("\n" + "=" * 60)
print("COMPONENT 5: fetch_twitter_api.py")
print("=" * 60)
import fetch_twitter_api
print("\n[Provider Selection]")
test("Auto keeps twitterapi.io when both keys exist",
fetch_twitter_api.resolve_provider(
"auto", {"TWITTERAPI_IO_KEY": "tw", "XQUIK_API_KEY": "xq"}
) == ("twitterapi", "tw"))
test("Auto uses Xquik when only Xquik key exists",
fetch_twitter_api.resolve_provider("auto", {"XQUIK_API_KEY": "xq"})
== ("xquik", "xq"))
test("Explicit Xquik reads XQUIK_API_KEY",
fetch_twitter_api.resolve_provider("xquik", {"XQUIK_API_KEY": "xq"})
== ("xquik", "xq"))
class _FakeResp:
def __init__(self, body):
self._body = json.dumps(body).encode("utf-8")
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
return False
def read(self):
return self._body
print("\n[Xquik Search Request]")
captured = {}
original_urlopen = fetch_twitter_api.urlopen
def _fake_urlopen(req, timeout, context):
captured["url"] = req.full_url
captured["headers"] = req.headers
return _FakeResp({"tweets": [{"id": "1", "text": "AI launch"}]})
try:
fetch_twitter_api.urlopen = _fake_urlopen
xquik_tweets = fetch_twitter_api.search_xquik("AI launch", "xq", max_results=7)
finally:
fetch_twitter_api.urlopen = original_urlopen
test("Xquik search returns tweets",
xquik_tweets == [{"id": "1", "text": "AI launch"}])
test("Xquik search hits /x/tweets/search",
"/x/tweets/search?" in captured.get("url", ""))
test("Xquik search passes limit",
"limit=7" in captured.get("url", ""))
test("Xquik search sends API key",
captured.get("headers", {}).get("X-api-key") == "xq")
test("Xquik search sends contract header",
captured.get("headers", {}).get("Xquik-api-contract") == "2026-04-29")
print("\n[Tweet URL Extraction]")
xquik_tweet = {
"id": "123",
"author": {"username": "openai"},
"entities": {"urls": []},
}
tweet_url, tweet_only = fetch_twitter_api.extract_url_from_tweet(xquik_tweet)
test("Xquik author username builds tweet URL",
tweet_url == "https://x.com/openai/status/123" and tweet_only)
external_tweet = {
"id": "123",
"author": {"username": "openai"},
"entities": {"urls": [{"expandedUrl": "https://openai.com/news"}]},
}
external_url, external_only = fetch_twitter_api.extract_url_from_tweet(external_tweet)
test("expandedUrl external link is preferred",
external_url == "https://openai.com/news" and not external_only)
# ==============================================================
print("\n" + "=" * 60)
print("SUMMARY")