esefandClaude Opus 5.5
2af2adc780
RSSHub's guid is the timeline entry's own id (a retweet's id), so ids follow the order the account posted, while feed order and pubDate follow the original tweet. A fresh retweet of an old post can sort to the bottom; it is now posted, and only ids older than every known one are treated as deletion slide-ins. !leaks sorts by id too. Co-Authored-By: Claude Opus 5.5 <[email protected]>
64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
import os
|
|
import aiohttp
|
|
import feedparser
|
|
from discord.ext import commands
|
|
|
|
from twitterbridge.rss import embed_link, tweet_id
|
|
|
|
REQUEST_TIMEOUT = 15
|
|
|
|
|
|
class TwitterLeaks(commands.Cog):
|
|
def __init__(self, bot):
|
|
self.bot = bot
|
|
|
|
@commands.command()
|
|
async def leaks(self, ctx):
|
|
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.')
|
|
|
|
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"
|
|
# the feed isn't chronological; status ids are
|
|
entries = sorted(entries, key=lambda e: tweet_id(e.get('id') or e.get('link', '')) or 0, reverse=True)
|
|
for entry in entries[:5]:
|
|
response += f"\n{embed_link(entry.get('link', ''))}\n"
|
|
|
|
await ctx.reply(response[:2000])
|
|
|
|
|
|
async def setup(bot):
|
|
await bot.add_cog(TwitterLeaks(bot))
|