twitterbridge: one poller, per-webhook seen sets, fxtwitter links, outage DM

The RSSHub feed isn't strictly chronological, so 'post everything above the
last guid' could skip tweets. Now a single poller fans out to both webhooks,
each remembering posted guids (data/twitter_seen.json, atomic writes): new
tweets anywhere above the oldest known one are posted oldest-first, old tweets
sliding in after a deletion are ignored, failed posts retry next poll and
Discord 429s are honoured. Links go out as fxtwitter.com for real embeds.
The owner gets a DM after 90 min of broken feed and again on recovery.
Night polling is now 1-2 h.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

main.py

@@ -18,23 +18,25 @@ bot = commands.Bot(command_prefix='!', intents=intents, help_command=None)
 # The Telegram bridge discovers its own routes from credentials.env and
 # starts polling when the extension loads — see tgbridge/config.py.
 
-twitter = TwitterRSSForwarder(
+async def alert_owner(text: str):
-    webhook_url=os.getenv('TWITTER_DISCORD_WEBHOOK'),
+    owner = await bot.fetch_user(int(os.getenv('OWNER_ID', '0')))
-)
+    await owner.send(text)
 
-twitter_mao = TwitterRSSForwarder(
+# one poller for the @CalabiyauLeaks feed, posting to both servers
-    webhook_url=os.getenv('TWITTER_DISCORD_WEBHOOK_MAO'),
+twitter = TwitterRSSForwarder(
-    guid_file='data/last_tweet_mao.txt',
+    webhooks={
+        'main': os.getenv('TWITTER_DISCORD_WEBHOOK'),
+        'mao': os.getenv('TWITTER_DISCORD_WEBHOOK_MAO'),
+    },
+    alert=alert_owner,
 )
 
 @bot.event
 async def on_ready():
     print(f'[+] {bot.user} is online!')
-    # on_ready fires again after reconnects; don't start duplicate pollers
+    # on_ready fires again after reconnects; don't start a duplicate poller
     if not twitter._running:
         asyncio.create_task(twitter.start())
-    if not twitter_mao._running:
-        asyncio.create_task(twitter_mao.start())
     # Clear stale global slash commands
     try:
         await bot.tree.sync()
@@ -71,11 +73,8 @@ async def main():
         finally:
             # bot.close() unloads extensions, so tgbridge.teardown stops the
             # Telegram poller and closes its aiohttp session for us
-            pass
             twitter.stop()
-            twitter_mao.stop()
             await twitter.close()
-            await twitter_mao.close()
 
 logging.basicConfig(level=logging.INFO)
 logging.getLogger('httpx').setLevel(logging.WARNING)

twitterbridge/leaks_cmd.py

@@ -3,6 +3,8 @@ import aiohttp
 import feedparser
 from discord.ext import commands
 
+from twitterbridge.rss import embed_link
+
 REQUEST_TIMEOUT = 15
 
 
@@ -50,10 +52,7 @@ class TwitterLeaks(commands.Cog):
 
         response = "**Latest 5 Leaks:**\n"
         for entry in entries[:5]:
-            link = entry.get('link', '')
+            response += f"\n{embed_link(entry.get('link', ''))}\n"
-            if 'nitter.net' in link:
-                link = link.replace('nitter.net', 'twitter.com').replace('#m', '')
-            response += f'\n{link}\n'
 
         await ctx.reply(response[:2000])
 

twitterbridge/rss.py

@@ -1,37 +1,58 @@
 # Twitter/X RSS-to-Discord forwarding (feed: self-hosted RSSHub, see esefrss).
-# Polls RSS endpoints on a timer, auto-rotates through fallback URLs
+# One poller fetches the feed and fans new entries out to every webhook.
-# when an instance returns empty or fails, and posts new entries
+# Each webhook remembers which tweets it has already posted, so a reordered,
-# to a Discord webhook.
+# partial or briefly broken feed never skips or reposts anything.
 import os
+import json
+import time
 import random
 import asyncio
 import logging
 from datetime import datetime
 from zoneinfo import ZoneInfo
+from urllib.parse import urlsplit, urlunsplit
 import aiohttp
 import feedparser
 
 logger = logging.getLogger(__name__)
 
 # The feed comes from RSSHub on yuuka using a real X account's cookie, so
-# polls are jittered and slower at night to avoid a robotic 24/7 pattern.
+# polls are jittered and much slower at night to avoid a robotic 24/7 pattern.
 POLL_INTERVAL = (240, 480)          # daytime: random 4-8 minutes
-NIGHT_POLL_INTERVAL = (1200, 2100) # night: random 20-35 minutes
+NIGHT_POLL_INTERVAL = (3600, 7200)  # night: random 1-2 hours
 NIGHT_HOURS = range(1, 6)           # 01:00-05:59 Istanbul; the account posts ~06:00-08:00
 LOCAL_TZ = ZoneInfo('Europe/Istanbul')
 REQUEST_TIMEOUT = 30                # seconds per HTTP call
-MAX_EMPTY_STRIKES = 3        # consecutive empty polls before rotating URL
+SEEN_LIMIT = 500                    # remembered guids per webhook (the feed holds 20)
+POST_GAP = 1.5                      # seconds between webhook posts
+POST_ATTEMPTS = 3
+ALERT_AFTER = 90 * 60               # feed broken this long -> DM the owner
+
+EMBED_HOST = 'fxtwitter.com'        # proper Discord embeds; X never sees these
+
+
+def embed_link(link: str) -> str:
+    """Point a tweet link at fxtwitter so Discord shows a real embed."""
+    parts = urlsplit(link.replace('#m', ''))
+    if parts.netloc in ('x.com', 'twitter.com', 'www.twitter.com', 'nitter.net'):
+        parts = parts._replace(netloc=EMBED_HOST)
+    return urlunsplit(parts)
+
+
+def _guid(entry) -> str:
+    return entry.get('id') or entry.get('link', '')
 
 
 class TwitterRSSForwarder:
-    """Polls a Twitter RSS feed and forwards new items to a Discord webhook.
+    """Polls the Twitter RSS feed and forwards new items to Discord webhooks.
 
-    Reads TWITTER_RSS_URL and optional TWITTER_RSS_FALLBACKS from the
+    Reads TWITTER_RSS_URL and optional comma-separated TWITTER_RSS_FALLBACKS.
-    environment.  Falls back through URLs when the current one fails or
+    `webhooks` maps a name (used in the state file) to a webhook URL.
-    returns empty entries.
+    `alert` is an optional coroutine taking a message, called when the feed
+    has been broken for ALERT_AFTER and again when it recovers.
     """
 
-    def __init__(self, webhook_url: str, guid_file: str = 'data/last_tweet.txt'):
+    def __init__(self, webhooks: dict[str, str], state_file: str = 'data/twitter_seen.json', alert=None):
         primary = os.getenv('TWITTER_RSS_URL', '')
         fallbacks_raw = os.getenv('TWITTER_RSS_FALLBACKS', '')
 
@@ -43,15 +64,16 @@ class TwitterRSSForwarder:
             raise ValueError('TWITTER_RSS_URL is not set')
 
         self.rss_urls = urls
-        self.webhook_url = webhook_url
+        self.webhooks = {name: url for name, url in webhooks.items() if url}
-        self.guid_file = guid_file
+        self.state_file = state_file
+        self.alert = alert
 
-        self.last_guid: str | None = None
+        self.seen: dict[str, list[str]] = {}
         self._running = False
         self._session: aiohttp.ClientSession | None = None
 
-        self._current_idx = 0
+        self._bad_since: float | None = None
-        self._empty_strikes = 0
+        self._alerted = False
 
     # ------------------------------------------------------------------
     # Public lifecycle
@@ -59,16 +81,13 @@ class TwitterRSSForwarder:
 
     async def start(self):
         self._running = True
-        self._load_last_guid()
+        self._load_state()
 
         timeout = aiohttp.ClientTimeout(total=REQUEST_TIMEOUT)
-        self._session = aiohttp.ClientSession(
+        self._session = aiohttp.ClientSession(timeout=timeout)
-            timeout=timeout,
-            headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'},
-        )
 
-        logger.info('twitter rss forwarder started — %d URLs, current: %s',
+        logger.info('twitter rss forwarder started — %s -> %s',
-                    len(self.rss_urls), self.rss_urls[self._current_idx])
+                    self.rss_urls[0], ', '.join(self.webhooks) or 'no webhooks')
 
         while self._running:
             try:
@@ -105,143 +124,152 @@ class TwitterRSSForwarder:
     # ------------------------------------------------------------------
 
     async def _poll_cycle(self):
-        start_idx = self._current_idx
+        entries, problem = await self._fetch_any()
-        errors: list[str] = []
-
-        for offset in range(len(self.rss_urls)):
-            idx = (start_idx + offset) % len(self.rss_urls)
-            url = self.rss_urls[idx]
-
-            try:
-                entries = await self._fetch_feed(url)
-            except asyncio.CancelledError:
-                raise
-            except Exception as exc:
-                errors.append(f'{url}: {exc}')
-                logger.warning('twitter rss url failed: %s — %s', url, exc)
-                continue
-
         if not entries:
-                logger.debug('twitter rss url returned empty: %s', url)
+            await self._note_bad(problem or 'feed returned no items')
-                self._empty_strikes += 1
-                if self._empty_strikes >= MAX_EMPTY_STRIKES:
-                    logger.warning('twitter rss %d consecutive empty polls — rotating away from %s',
-                                   self._empty_strikes, url)
-                    self._rotate()
-                else:
-                    logger.debug('twitter rss empty strike %d/%d on %s',
-                                 self._empty_strikes, MAX_EMPTY_STRIKES, url)
             return
+        await self._note_good()
 
-            self._empty_strikes = 0
+        for name, webhook in self.webhooks.items():
-            if idx != self._current_idx:
+            await self._forward(name, webhook, entries)
-                logger.info('twitter rss switched to %s', url)
-                self._current_idx = idx
-
-            await self._process_entries(entries)
-            return
-
-        logger.error('twitter rss all %d URLs failed: %s', len(self.rss_urls), '; '.join(errors))
-        self._rotate()
-
-    async def _fetch_feed(self, url: str) -> list[dict] | None:
-        if self._session is None:
-            raise RuntimeError('start() must be called before fetching')
 
+    async def _fetch_any(self) -> tuple[list[dict], str | None]:
+        problems = []
+        for url in self.rss_urls:
+            try:
                 async with self._session.get(url) as resp:
                     resp.raise_for_status()
                     text = await resp.text()
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:
+                problems.append(f'{url}: {exc}')
+                continue
 
             feed = feedparser.parse(text)
+            if feed.entries:
+                return feed.entries, None
+            problems.append(f'{url}: no items' + (f' ({feed.bozo_exception})' if feed.bozo else ''))
+        return [], '; '.join(problems)
 
-        if feed.bozo and not feed.entries:
+    async def _forward(self, name: str, webhook: str, entries: list[dict]):
-            logger.warning('twitter rss parse error from %s: %s', url, feed.bozo_exception)
+        ids = [_guid(e) for e in entries]
-            return None
+        seen = self.seen.get(name)
 
-        return feed.entries
+        if seen is None:
-
+            # first run for this webhook: remember the current feed, post nothing
-    # ------------------------------------------------------------------
+            self.seen[name] = ids[::-1]
-    # Entry processing
+            self._save_state()
-    # ------------------------------------------------------------------
+            logger.info('twitter rss [%s] first run — remembered %d tweets', name, len(ids))
-
-    async def _process_entries(self, entries: list[dict]):
-        if self.last_guid is None:
-            self.last_guid = entries[0].get('id', '')
-            self._save_last_guid()
-            logger.info('twitter rss first run — recorded guid: %s', self.last_guid)
             return
 
-        # Find the index of the last known GUID in the current feed.
+        seen_set = set(seen)
-        last_idx = None
+        known = [i for i, g in enumerate(ids) if g in seen_set]
-        for i, entry in enumerate(entries):
+        if not known:
-            if entry.get('id', '') == self.last_guid:
+            # nothing in the feed is familiar (long outage or a feed change):
-                last_idx = i
+            # re-baseline instead of flooding the channel with the whole page
-                break
+            logger.warning('twitter rss [%s] no known tweets in feed — re-baselining without posting', name)
-
+            seen.extend(ids[::-1])
-        if last_idx is None:
+            del seen[:-SEEN_LIMIT]
-            # Stored GUID not in this feed — nitter returned a stale/partial
+            self._save_state()
-            # snapshot.  Update to the newest GUID *without* posting, so we
-            # don't spam the channel with duplicates when the full feed returns.
-            logger.warning(
-                'stored guid %s not found in feed (%d entries) — '
-                'updating to %s without posting',
-                self.last_guid, len(entries), entries[0].get('id', '?'),
-            )
-            self.last_guid = entries[0].get('id', '')
-            self._save_last_guid()
             return
 
-        new_entries = entries[:last_idx]  # everything newer than last known
+        # Unseen entries above the lowest known one are new (the feed isn't strictly
-        if not new_entries:
+        # chronological, so they can sit below newer known tweets). Unseen entries
-            return
+        # below every known one are old tweets that slid in when something above
+        # was deleted, so they are only remembered.
+        lowest_known = known[-1]
+        new = [e for e in entries[:lowest_known] if _guid(e) not in seen_set]
+        tail = [g for g in ids[lowest_known:] if g not in seen_set]
+        if tail:
+            seen.extend(tail)
+            del seen[:-SEEN_LIMIT]
+            self._save_state()
 
-        for entry in reversed(new_entries):
+        for entry in reversed(new):  # oldest first
-            await self._post_to_discord(entry)
+            if not await self._post(name, webhook, entry):
-
+                break  # leave the rest unseen; retried next poll
-        self.last_guid = new_entries[0].get('id', '')
+            seen.append(_guid(entry))
-        self._save_last_guid()
+            del seen[:-SEEN_LIMIT]
+            self._save_state()
+            await asyncio.sleep(POST_GAP)
 
     # ------------------------------------------------------------------
     # Discord webhook
     # ------------------------------------------------------------------
 
-    async def _post_to_discord(self, entry):
+    async def _post(self, name: str, webhook: str, entry) -> bool:
-        link = entry.get('link', '')
+        link = embed_link(entry.get('link', ''))
-        if 'nitter.net' in link:
+        payload = {'username': '@CalabiyauLeaks', 'content': link}
-            link = link.replace('nitter.net', 'twitter.com').replace('#m', '')
 
-        payload = {
+        for _ in range(POST_ATTEMPTS):
-            'username': '@CalabiyauLeaks',
+            try:
-            'content': link,
+                async with self._session.post(webhook, json=payload) as resp:
-        }
+                    if resp.status in (200, 204):
-
+                        logger.info('twitter rss [%s] posted: %s', name, link)
-        async with self._session.post(self.webhook_url, json=payload) as resp:
+                        return True
-            if resp.status not in (200, 204):
+                    if resp.status == 429:
+                        data = await resp.json(content_type=None)
+                        wait = float(data.get('retry_after', 2))
+                        logger.info('twitter rss [%s] rate limited, waiting %.1fs', name, wait)
+                        await asyncio.sleep(min(wait, 60))
+                        continue
                     body = await resp.text()
-                logger.warning('twitter webhook post failed: %s %s', resp.status, body[:200])
+                    logger.warning('twitter rss [%s] webhook post failed: %s %s', name, resp.status, body[:200])
-            else:
+                    if resp.status < 500:
-                logger.info('twitter rss posted: %s', link)
+                        return False
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:
+                logger.warning('twitter rss [%s] webhook post error: %s', name, type(exc).__name__)
+            await asyncio.sleep(2)
+        return False
 
     # ------------------------------------------------------------------
-    # Rotation
+    # Outage alerts
     # ------------------------------------------------------------------
 
-    def _rotate(self):
+    async def _note_bad(self, problem: str):
-        self._current_idx = (self._current_idx + 1) % len(self.rss_urls)
+        now = time.monotonic()
-        self._empty_strikes = 0
+        if self._bad_since is None:
-        logger.info('twitter rss rotated to URL %d/%d: %s',
+            self._bad_since = now
-                    self._current_idx + 1, len(self.rss_urls), self.rss_urls[self._current_idx])
+        logger.warning('twitter rss feed problem: %s', problem)
+
+        broken_for = now - self._bad_since
+        if not self._alerted and broken_for >= ALERT_AFTER:
+            self._alerted = True
+            await self._send_alert(
+                f'⚠️ esefrss: the @CalabiyauLeaks feed has been broken for {broken_for / 60:.0f} min '
+                f'({problem[:300]}).\nIf X locked @esefos2: log in on arona, pass the check, then put a fresh '
+                f'auth_token in ~/esefrss/.env and `docker compose up -d --force-recreate` (see esefrss README).'
+            )
+
+    async def _note_good(self):
+        if self._alerted:
+            await self._send_alert('✅ esefrss: the @CalabiyauLeaks feed is working again.')
+        self._bad_since = None
+        self._alerted = False
+
+    async def _send_alert(self, text: str):
+        if self.alert is None:
+            return
+        try:
+            await self.alert(text)
+        except Exception:
+            logger.exception('twitter rss alert failed')
 
     # ------------------------------------------------------------------
     # Persistence
     # ------------------------------------------------------------------
 
-    def _load_last_guid(self):
+    def _load_state(self):
-        if os.path.exists(self.guid_file):
+        if os.path.exists(self.state_file):
-            with open(self.guid_file, 'r') as f:
+            with open(self.state_file, 'r') as f:
-                self.last_guid = f.read().strip() or None
+                self.seen = {k: list(v) for k, v in json.load(f).items()}
 
-    def _save_last_guid(self):
+    def _save_state(self):
-        os.makedirs('data', exist_ok=True)
+        os.makedirs(os.path.dirname(self.state_file) or '.', exist_ok=True)
-        with open(self.guid_file, 'w') as f:
+        tmp = self.state_file + '.tmp'
-            f.write(self.last_guid or '')
+        with open(tmp, 'w') as f:
+            json.dump(self.seen, f)
+        os.replace(tmp, self.state_file)