50 lines
2.3 KiB
Python
50 lines
2.3 KiB
Python
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
|