feat(news): add opt-in Xquik search source

backend/services/fetchers/news.py

@@ -5,11 +5,13 @@ import time
 import logging
 import calendar
 import concurrent.futures
+
 import requests
 import feedparser
 from services.network_utils import fetch_with_curl
 from services.fetchers._store import latest_data, _data_lock, _mark_fresh
 from services.fetchers.retry import with_retry
+from services.fetchers.xquik_news import fetch_xquik_entries
 from services.oracle_service import enrich_news_items, compute_global_threat_level, detect_breaking_events
 
 
@@ -200,18 +202,23 @@ def fetch_news():
         source_name, url = item
         try:
             xml_data = fetch_with_curl(url, timeout=10).text
-            return source_name, feedparser.parse(xml_data)
+            return source_name, feedparser.parse(xml_data).entries
         except (requests.RequestException, ConnectionError, TimeoutError, ValueError, KeyError, OSError) as e:
             logger.warning(f"Feed {source_name} failed: {e}")
-            return source_name, None
+            return source_name, []
 
     with concurrent.futures.ThreadPoolExecutor(max_workers=min(len(feeds), 6)) as pool:
         feed_results = list(pool.map(_fetch_feed, feeds.items()))
 
-    for source_name, feed in feed_results:
-        if not feed:
+    for entry in fetch_xquik_entries():
+        source_name = entry["source"]
+        source_weights[source_name] = 3
+        feed_results.append((source_name, [entry]))
+
+    for source_name, entries in feed_results:
+        if not entries:
             continue
-        for entry in feed.entries[:5]:
+        for entry in entries[:5]:
             # Drop articles older than the max-age threshold so the
             # threat feed doesn't show stale stories across cycles.
             pp = entry.get("published_parsed")

backend/services/fetchers/xquik_news.py

@@ -0,0 +1,163 @@
+"""Opt-in X post search for the shared news and threat feed."""
+
+from __future__ import annotations
+
+import logging
+import os
+import re
+import threading
+import time
+from datetime import datetime, timezone
+from typing import Any
+
+import requests
+
+from services.network_utils import outbound_user_agent
+
+
+logger = logging.getLogger("services.data_fetcher")
+
+_SEARCH_URL = "https://xquik.com/api/v1/x/tweets/search"
+_TWEET_ID_RE = re.compile(r"^[0-9]{1,32}$")
+_USERNAME_RE = re.compile(r"^[A-Za-z0-9_]{1,15}$")
+_MAX_RESULTS = 100
+_MAX_TEXT_LENGTH = 500
+
+_cache_lock = threading.Lock()
+_cache_signature: tuple[str, int] | None = None
+_cache_entries: list[dict[str, Any]] = []
+_cache_fetched_at = 0.0
+
+
+def xquik_fetch_enabled() -> bool:
+    """Return whether the operator enabled Xquik search enrichment."""
+    return str(os.environ.get("XQUIK_ENABLED", "false")).strip().lower() in {
+        "1",
+        "true",
+        "yes",
+        "on",
+    }
+
+
+def _bounded_int(name: str, default: int, minimum: int, maximum: int) -> int:
+    try:
+        value = int(str(os.environ.get(name, default)).strip())
+    except (TypeError, ValueError):
+        return default
+    return max(minimum, min(maximum, value))
+
+
+def _published_parts(value: object) -> time.struct_time | None:
+    if not isinstance(value, str) or not value.strip():
+        return None
+    try:
+        parsed = datetime.fromisoformat(value.strip().replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    if parsed.tzinfo is None:
+        return None
+    return parsed.astimezone(timezone.utc).timetuple()
+
+
+def _normalize_tweet(tweet: object) -> dict[str, Any] | None:
+    if not isinstance(tweet, dict):
+        return None
+    tweet_id = str(tweet.get("id") or "").strip()
+    text = " ".join(str(tweet.get("text") or "").split())
+    author = tweet.get("author")
+    username = str(author.get("username") or "").strip() if isinstance(author, dict) else ""
+    if not _TWEET_ID_RE.fullmatch(tweet_id) or not _USERNAME_RE.fullmatch(username) or not text:
+        return None
+
+    created_at = tweet.get("createdAt")
+    published_parts = _published_parts(created_at)
+    if published_parts is None:
+        return None
+    entry: dict[str, Any] = {
+        "title": text[:_MAX_TEXT_LENGTH],
+        "summary": "",
+        "link": f"https://x.com/{username}/status/{tweet_id}",
+        "published": created_at,
+        "published_parsed": published_parts,
+        "source": f"Xquik/@{username}",
+    }
+    return entry
+
+
+def _request_entries(api_key: str, query: str, limit: int) -> list[dict[str, Any]] | None:
+    timeout = _bounded_int("XQUIK_SEARCH_TIMEOUT_S", 10, 1, 30)
+    try:
+        response = requests.get(
+            _SEARCH_URL,
+            headers={
+                "User-Agent": outbound_user_agent("xquik-search"),
+                "x-api-key": api_key,
+            },
+            params={
+                "q": query,
+                "queryType": "Latest",
+                "limit": limit,
+                "replies": "exclude",
+                "retweets": "exclude",
+            },
+            timeout=(5, timeout),
+            allow_redirects=False,
+        )
+        if 300 <= response.status_code < 400:
+            raise requests.HTTPError("Xquik search redirected")
+        response.raise_for_status()
+        payload = response.json()
+    except (requests.RequestException, ValueError) as exc:
+        logger.warning("Xquik search failed: %s", type(exc).__name__)
+        return None
+
+    tweets = payload.get("tweets") if isinstance(payload, dict) else None
+    if not isinstance(tweets, list):
+        logger.warning("Xquik search returned an invalid response")
+        return None
+
+    entries = [entry for tweet in tweets[:limit] if (entry := _normalize_tweet(tweet))]
+    logger.info("Xquik search returned %d usable posts", len(entries))
+    return entries
+
+
+def fetch_xquik_entries() -> list[dict[str, Any]]:
+    """Return cached, normalized X posts for the configured search query."""
+    if not xquik_fetch_enabled():
+        return []
+
+    api_key = str(os.environ.get("XQUIK_API_KEY", "")).strip()
+    query = str(os.environ.get("XQUIK_SEARCH_QUERY", "")).strip()
+    if not api_key or not query:
+        logger.warning("Xquik search requires XQUIK_API_KEY and XQUIK_SEARCH_QUERY")
+        return []
+
+    limit = _bounded_int("XQUIK_SEARCH_LIMIT", 20, 1, _MAX_RESULTS)
+    interval_seconds = _bounded_int("XQUIK_SEARCH_INTERVAL_MINUTES", 30, 5, 1440) * 60
+    signature = (query, limit)
+
+    global _cache_entries, _cache_fetched_at, _cache_signature
+    with _cache_lock:
+        if (
+            _cache_signature == signature
+            and _cache_fetched_at
+            and time.monotonic() - _cache_fetched_at < interval_seconds
+        ):
+            return [dict(entry) for entry in _cache_entries]
+
+        entries = _request_entries(api_key, query, limit)
+        if entries is not None:
+            _cache_signature = signature
+            _cache_entries = entries
+            _cache_fetched_at = time.monotonic()
+        if _cache_signature == signature:
+            return [dict(entry) for entry in _cache_entries]
+        return []
+
+
+def _reset_cache_for_tests() -> None:
+    global _cache_entries, _cache_fetched_at, _cache_signature
+    with _cache_lock:
+        _cache_signature = None
+        _cache_entries = []
+        _cache_fetched_at = 0.0