import sentry_sdk import twitter from django.db import models from djconfig import config from telegram import Bot, InputMediaPhoto, InputMediaVideo from feeds.models import FeedModuleConfig class TwitterFeedModuleConfig(FeedModuleConfig): screen_name = models.CharField(max_length=256) MODULE_NAME = 'Twitter feed' def execute(self, bot: Bot, chat_id, last_id): config._reload_maybe() if last_id is None: last_id = 0 api = twitter.Api(config.twitter_consumer_api_key, config.twitter_consumer_api_secret, config.twitter_access_token, config.twitter_access_token_secret) for status in reversed(api.GetUserTimeline(screen_name=self.screen_name, since_id=last_id)): try: text = status.text # type: str for url in status.urls: text = text.replace(url.url, url.expanded_url) media = list(filter(lambda m: m.type in ('photo', 'animated_gif', 'video'), status.media)) if not len(media): bot.send_message(chat_id, text) elif len(media) == 1: if media[0].type == 'photo': bot.send_photo(chat_id, media[0].media_url_https, text) elif media[0].type == 'animated_gif': bot.send_document(chat_id, media[0].video_info['variants'][-1]['url'], caption=text) else: bot.send_video(chat_id, media[0].video_info['variants'][-1]['url'], caption=text) else: converted_media = [] for i, m in enumerate(media): if m.type == 'photo': converted_media.append(InputMediaPhoto(m.media_url_https, text if not i else None)) elif m.type in ('animated_gif', 'video'): converted_media.append(InputMediaVideo(m.video_info['variants'][-1]['url'], caption=text if not i else None)) bot.send_media_group(chat_id, converted_media) except Exception as e: sentry_sdk.capture_exception(e) yield status.id