From 300f2fa8f041275ed3a21255666b03acbfaf545b Mon Sep 17 00:00:00 2001 From: esefxdz Date: Thu, 16 Jul 2026 12:26:32 +0300 Subject: [PATCH] harden twitter rss forwarder with timeout, rotation, link-only posts --- credentials_test.env | 3 +- main.py | 4 +- twitterbridge/leaks_cmd.py | 73 ++++++---- twitterbridge/rss.py | 264 ++++++++++++++++++++++++++----------- 4 files changed, 240 insertions(+), 104 deletions(-) diff --git a/credentials_test.env b/credentials_test.env index 9466dc4..f7f7261 100644 --- a/credentials_test.env +++ b/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" diff --git a/main.py b/main.py index 62bf6cb..c931080 100644 --- a/main.py +++ b/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) diff --git a/twitterbridge/leaks_cmd.py b/twitterbridge/leaks_cmd.py index e129c7f..0255256 100644 --- a/twitterbridge/leaks_cmd.py +++ b/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') - if not rss_url: + 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: return await ctx.reply('rss url not configured.') - - try: - 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) - if not feed.entries: - return await ctx.reply('no tweets found.') - - response = "**Latest 5 Leaks:**\n" - for entry in feed.entries[:5]: - title = entry.get('title', '') - link = entry.get('link', '') - - if 'nitter.net' in link: - link = link.replace('nitter.net', 'twitter.com').replace('#m', '') - - response += f"\n{title}\n{link}\n" - - await ctx.reply(response[:2000]) - except Exception as e: - await ctx.reply(f'failed to fetch leaks: {e}') + + timeout = aiohttp.ClientTimeout(total=REQUEST_TIMEOUT) + headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'} + + async with aiohttp.ClientSession(timeout=timeout, headers=headers) as session: + entries = None + last_error = None + + for url in urls: + try: + async with session.get(url) as resp: + 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 not entries: + msg = 'no tweets found.' + if last_error: + msg += f' (last error: {last_error})' + return await ctx.reply(msg[:2000]) + + 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]) + async def setup(bot): await bot.add_cog(TwitterLeaks(bot)) diff --git a/twitterbridge/rss.py b/twitterbridge/rss.py index 0e0d534..b01c414 100644 --- a/twitterbridge/rss.py +++ b/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 -GUID_FILE = 'data/last_tweet.txt' +logger = logging.getLogger(__name__) + +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'): - self.rss_url = rss_url + """Polls a Twitter RSS feed and forwards new items to a Discord webhook. + + 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() - except Exception as e: - print(f'twitter rss poll failed: {e}') + await self._poll_cycle() + except asyncio.CancelledError: + 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]}...') \ No newline at end of file