Use Python's internal asyncio stuff instead of Tornado's

[ipfire.org.git] / src / backend / tweets.py

#!/usr/bin/python3

import datetime
import logging
import twython

from .misc import Object

class Tweets(Object):
	async def tweet(self):
		"""
			Sends a random promotional tweet
		"""
		# Do not tweet too often
		if self.has_had_recent_activity(days=4):
			logging.debug("Won't tweet because we recently did it")
			return

		# Do not tweet when there was a blog post
		if self.backend.blog.has_had_recent_activity(hours=24):
			logging.debug("Won't tweet because the blog has had activity")
			return

		# Select a tweet
		tweet = self._get_random_tweet()
		if not tweet:
			logging.warning("Could not find anything to tweet")
			return

		# Tweet the tweet
		with self.db.transaction():
			self._tweet(tweet)

	def has_had_recent_activity(self, **kwargs):
		t = datetime.timedelta(**kwargs)

		res = self.db.get("SELECT COUNT(*) AS count FROM tweets \
			WHERE last_tweeted_at IS NOT NULL AND last_tweeted_at >= NOW() - %s", t)

		if res and res.count > 0:
			return True

		return False

	def _get_random_tweet(self):
		res = self.db.get(
			"WITH candidate_tweets AS (SELECT id, \
				(CURRENT_TIMESTAMP - COALESCE(last_tweeted_at, '1970-01-01')) * RANDOM() AS age \
				FROM tweets \
				WHERE (last_tweeted_at IS NULL OR last_tweeted_at <= CURRENT_TIMESTAMP - INTERVAL '1 month') \
			) \
			SELECT tweets.* FROM candidate_tweets \
				LEFT JOIN tweets ON candidate_tweets.id = tweets.id \
				ORDER BY age DESC LIMIT 1")

		return res

	def _tweet(self, tweet):
		logging.debug("Tweeting: %s" % tweet.message)

		# Update database status
		self.db.execute("UPDATE tweets \
			SET last_tweeted_at = CURRENT_TIMESTAMP, total_tweets = total_tweets + 1 \
			WHERE id = %s", tweet.id)

		# Connect to twitter
		twitter = twython.Twython(
			self.settings.get("twitter_consumer_key"),
			self.settings.get("twitter_consumer_secret"),
			self.settings.get("twitter_%s_access_token" % tweet.account),
			self.settings.get("twitter_%s_access_token_secret" % tweet.account),
		)

		# Update status
		twitter.update_status(status=tweet.message)