harden twitter rss forwarder with timeout, rotation, link-only posts

credentials_test.env

@@ -9,7 +9,8 @@ DISCORD_WEBHOOK_URL="yo channel webhook here"
 TELEGRAM_GROUP_ID="yo telegram group id here"
 
 #twitter things
-TWITTER_RSS_URL="yo nitter id here"
+TWITTER_RSS_URL="https://nitter.net/CalabiyauLeaks/rss"
+TWITTER_RSS_FALLBACKS=
 TWITTER_DISCORD_WEBHOOK="yo discord twitter channel here"
 TWITTER_DISCORD_WEBHOOK_MAO="yo discord twitter channel here for mao server"
 

main.py

@@ -33,12 +33,10 @@ if os.getenv('DISCORD_CHANNEL_ID_SHITPOST') and os.getenv('TELEGRAM_GROUP_ID_SHI
     d_to_tg.add_route(int(os.getenv('DISCORD_CHANNEL_ID_SHITPOST')), int(os.getenv('TELEGRAM_GROUP_ID_SHITPOST')))
 
 twitter = TwitterRSSForwarder(
-    rss_url=os.getenv('TWITTER_RSS_URL'),
     webhook_url=os.getenv('TWITTER_DISCORD_WEBHOOK'),
 )
 
 twitter_mao = TwitterRSSForwarder(
-    rss_url=os.getenv('TWITTER_RSS_URL'),
     webhook_url=os.getenv('TWITTER_DISCORD_WEBHOOK_MAO'),
     guid_file='data/last_tweet_mao.txt',
 )
@@ -84,6 +82,8 @@ async def main():
             await forwarder.stop()
             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,39 +3,60 @@ import aiohttp
 import feedparser
 from discord.ext import commands
 
+REQUEST_TIMEOUT = 15
+
+
 class TwitterLeaks(commands.Cog):
     def __init__(self, bot):
         self.bot = bot
 
     @commands.command()
     async def leaks(self, ctx):
-        rss_url = os.getenv('TWITTER_RSS_URL')
+        primary = os.getenv('TWITTER_RSS_URL', '')
-        if not rss_url:
+        fallbacks_raw = os.getenv('TWITTER_RSS_FALLBACKS', '')
+
+        urls = [primary] if primary else []
+        if fallbacks_raw:
+            urls.extend(u.strip() for u in fallbacks_raw.split(',') if u.strip())
+
+        if not urls:
             return await ctx.reply('rss url not configured.')
 
-        try:
+        timeout = aiohttp.ClientTimeout(total=REQUEST_TIMEOUT)
-            headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}
+        headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}
-            async with aiohttp.ClientSession(headers=headers) as session:
-                async with session.get(rss_url) as resp:
-                    text = await resp.text()
 
-            feed = feedparser.parse(text)
+        async with aiohttp.ClientSession(timeout=timeout, headers=headers) as session:
-            if not feed.entries:
+            entries = None
-                return await ctx.reply('no tweets found.')
+            last_error = None
 
-            response = "**Latest 5 Leaks:**\n"
+            for url in urls:
-            for entry in feed.entries[:5]:
+                try:
-                title = entry.get('title', '')
+                    async with session.get(url) as resp:
-                link = entry.get('link', '')
+                        resp.raise_for_status()
+                        text = await resp.text()
+                    feed = feedparser.parse(text)
+                    if feed.entries:
+                        entries = feed.entries
+                        break
+                except Exception as e:
+                    last_error = e
+                    continue
 
-                if 'nitter.net' in link:
+        if not entries:
-                    link = link.replace('nitter.net', 'twitter.com').replace('#m', '')
+            msg = 'no tweets found.'
+            if last_error:
+                msg += f' (last error: {last_error})'
+            return await ctx.reply(msg[:2000])
 
-                response += f"\n{title}\n{link}\n"
+        response = "**Latest 5 Leaks:**\n"
+        for entry in entries[:5]:
+            link = entry.get('link', '')
+            if 'nitter.net' in link:
+                link = link.replace('nitter.net', 'twitter.com').replace('#m', '')
+            response += f'\n{link}\n'
+
+        await ctx.reply(response[:2000])
 
-            await ctx.reply(response[:2000])
-        except Exception as e:
-            await ctx.reply(f'failed to fetch leaks: {e}')
 
 async def setup(bot):
     await bot.add_cog(TwitterLeaks(bot))

twitterbridge/rss.py

@@ -1,36 +1,213 @@
-# this is where twitter/x rss-to-discord forwarding lives
+# Twitter/X RSS-to-Discord forwarding via Nitter RSS feeds.
+# Polls RSS endpoints on a timer, auto-rotates through fallback URLs
+# when an instance returns empty or fails, and posts new entries
+# to a Discord webhook.
 import os
 import asyncio
+import logging
 import aiohttp
 import feedparser
 
-POLL_INTERVAL = 300  # 5 minutes
+logger = logging.getLogger(__name__)
-GUID_FILE = 'data/last_tweet.txt'
+
+POLL_INTERVAL = 300          # 5 minutes
+REQUEST_TIMEOUT = 30         # seconds per HTTP call
+MAX_EMPTY_STRIKES = 3        # consecutive empty polls before rotating URL
+
 
 class TwitterRSSForwarder:
-    def __init__(self, rss_url, webhook_url, guid_file='data/last_tweet.txt'):
+    """Polls a Twitter RSS feed and forwards new items to a Discord webhook.
-        self.rss_url = rss_url
+
+    Reads TWITTER_RSS_URL and optional TWITTER_RSS_FALLBACKS from the
+    environment.  Falls back through URLs when the current one fails or
+    returns empty entries.
+    """
+
+    def __init__(self, webhook_url: str, guid_file: str = 'data/last_tweet.txt'):
+        primary = os.getenv('TWITTER_RSS_URL', '')
+        fallbacks_raw = os.getenv('TWITTER_RSS_FALLBACKS', '')
+
+        urls = [primary] if primary else []
+        if fallbacks_raw:
+            urls.extend(u.strip() for u in fallbacks_raw.split(',') if u.strip())
+
+        if not urls:
+            raise ValueError('TWITTER_RSS_URL is not set')
+
+        self.rss_urls = urls
         self.webhook_url = webhook_url
         self.guid_file = guid_file
-        self.last_guid = None
+
+        self.last_guid: str | None = None
         self._running = False
+        self._session: aiohttp.ClientSession | None = None
+
+        self._current_idx = 0
+        self._empty_strikes = 0
+
+    # ------------------------------------------------------------------
+    # Public lifecycle
+    # ------------------------------------------------------------------
 
     async def start(self):
         self._running = True
         self._load_last_guid()
-        print(f'✅ twitter rss forwarder started — polling {self.rss_url}')
+
+        timeout = aiohttp.ClientTimeout(total=REQUEST_TIMEOUT)
+        self._session = aiohttp.ClientSession(
+            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',
+                    len(self.rss_urls), self.rss_urls[self._current_idx])
 
         while self._running:
             try:
-                await self._poll()
+                await self._poll_cycle()
-            except Exception as e:
+            except asyncio.CancelledError:
-                print(f'twitter rss poll failed: {e}')
+                logger.info('twitter rss forwarder cancelled during poll')
+                break
+            except Exception:
+                logger.exception('unhandled error in twitter rss poll loop — will retry next cycle')
 
-            await asyncio.sleep(POLL_INTERVAL)
+            try:
+                await asyncio.sleep(POLL_INTERVAL)
+            except asyncio.CancelledError:
+                self._running = False
+                raise
 
     def stop(self):
         self._running = False
-        print('twitter rss forwarder stopped')
+        logger.info('twitter rss forwarder stop requested')
+
+    async def close(self):
+        if self._session is not None:
+            await self._session.close()
+            self._session = None
+            logger.debug('twitter rss http session closed')
+
+    # ------------------------------------------------------------------
+    # Poll cycle
+    # ------------------------------------------------------------------
+
+    async def _poll_cycle(self):
+        start_idx = self._current_idx
+        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)
+                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
+
+            self._empty_strikes = 0
+            if idx != self._current_idx:
+                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 with self._session.get(url) as resp:
+            resp.raise_for_status()
+            text = await resp.text()
+
+        feed = feedparser.parse(text)
+
+        if feed.bozo and not feed.entries:
+            logger.warning('twitter rss parse error from %s: %s', url, feed.bozo_exception)
+            return None
+
+        return feed.entries
+
+    # ------------------------------------------------------------------
+    # Entry processing
+    # ------------------------------------------------------------------
+
+    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
+
+        new_entries = []
+        for entry in entries:
+            guid = entry.get('id', '')
+            if guid == self.last_guid:
+                break
+            new_entries.append(entry)
+
+        if not new_entries:
+            return
+
+        for entry in reversed(new_entries):
+            await self._post_to_discord(entry)
+
+        self.last_guid = new_entries[0].get('id', '')
+        self._save_last_guid()
+
+    # ------------------------------------------------------------------
+    # Discord webhook
+    # ------------------------------------------------------------------
+
+    async def _post_to_discord(self, entry):
+        link = entry.get('link', '')
+        if 'nitter.net' in link:
+            link = link.replace('nitter.net', 'twitter.com').replace('#m', '')
+
+        payload = {
+            'username': '@CalabiyauLeaks',
+            'content': link,
+        }
+
+        async with self._session.post(self.webhook_url, json=payload) as resp:
+            if resp.status not in (200, 204):
+                body = await resp.text()
+                logger.warning('twitter webhook post failed: %s %s', resp.status, body[:200])
+            else:
+                logger.info('twitter rss posted: %s', link)
+
+    # ------------------------------------------------------------------
+    # Rotation
+    # ------------------------------------------------------------------
+
+    def _rotate(self):
+        self._current_idx = (self._current_idx + 1) % len(self.rss_urls)
+        self._empty_strikes = 0
+        logger.info('twitter rss rotated to URL %d/%d: %s',
+                    self._current_idx + 1, len(self.rss_urls), self.rss_urls[self._current_idx])
+
+    # ------------------------------------------------------------------
+    # Persistence
+    # ------------------------------------------------------------------
 
     def _load_last_guid(self):
         if os.path.exists(self.guid_file):
@@ -41,66 +218,3 @@ class TwitterRSSForwarder:
         os.makedirs('data', exist_ok=True)
         with open(self.guid_file, 'w') as f:
             f.write(self.last_guid or '')
-
-    async def _poll(self):
-        headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}
-        async with aiohttp.ClientSession(headers=headers) as session:
-            async with session.get(self.rss_url) as resp:
-                text = await resp.text()
-
-        feed = feedparser.parse(text)
-
-        if feed.bozo and not feed.entries:
-            print(f'twitter rss feed error: {feed.bozo_exception}')
-            return
-
-        if not feed.entries:
-            return
-
-        # first run — just record the latest guid, don't post anything
-        if self.last_guid is None:
-            self.last_guid = feed.entries[0].get('id', '')
-            self._save_last_guid()
-            print(f'twitter rss first run — recorded guid: {self.last_guid}')
-            return
-
-        # collect new entries (entries are newest-first in rss)
-        new_entries = []
-        for entry in feed.entries:
-            guid = entry.get('id', '')
-            if guid == self.last_guid:
-                break
-            new_entries.append(entry)
-
-        if not new_entries:
-            return
-
-        # post oldest first so discord order matches timeline
-        for entry in reversed(new_entries):
-            await self._post_to_discord(entry)
-
-        # update last seen guid to the newest
-        self.last_guid = new_entries[0].get('id', '')
-        self._save_last_guid()
-
-    async def _post_to_discord(self, entry):
-        title = entry.get('title', '')
-        link = entry.get('link', '')
-
-        # nitter links -> real twitter links
-        if 'nitter.net' in link:
-            link = link.replace('nitter.net', 'twitter.com').replace('#m', '')
-
-        content = f'{title}\n{link}'.strip()
-
-        payload = {
-            'username': '@CalabiyauLeaks',
-            'content': content[:2000],
-        }
-
-        async with aiohttp.ClientSession() as session:
-            async with session.post(self.webhook_url, json=payload) as resp:
-                if resp.status not in (200, 204):
-                    print(f'twitter webhook post failed: {resp.status} {await resp.text()}')
-                else:
-                    print(f'twitter rss posted: {title[:60]}...')