src/backend/tweets.py

   1 #!/usr/bin/python3
   2
   3 import datetime
   4 import logging
   5 import twython
   6
   7 from .misc import Object
   8
   9 class Tweets(Object):
  10         async def tweet(self):
  11                 """
  12                         Sends a random promotional tweet
  13                 """
  14                 # Do not tweet too often
  15                 if self.has_had_recent_activity(days=3):
  16                         logging.debug("Won't tweet because we recently did it")
  17                         return
  18
  19                 # Do not tweet when there was a blog post
  20                 if self.backend.blog.has_had_recent_activity(hours=24):
  21                         logging.debug("Won't tweet because the blog has had activity")
  22                         return
  23
  24                 # Select a tweet
  25                 tweet = self._get_random_tweet()
  26                 if not tweet:
  27                         logging.warning("Could not find anything to tweet")
  28                         return
  29
  30                 # Tweet the tweet
  31                 with self.db.transaction():
  32                         self._tweet(tweet)
  33
  34         def has_had_recent_activity(self, **kwargs):
  35                 t = datetime.timedelta(**kwargs)
  36
  37                 res = self.db.get("SELECT COUNT(*) AS count FROM tweets \
  38                         WHERE last_tweeted_at IS NOT NULL AND last_tweeted_at >= NOW() - %s", t)
  39
  40                 if res and res.count > 0:
  41                         return True
  42
  43                 return False
  44
  45         def _get_random_tweet(self):
  46                 res = self.db.get(
  47                         "WITH candidate_tweets AS (SELECT id, \
  48                                 (CURRENT_TIMESTAMP - COALESCE(last_tweeted_at, '1970-01-01')) * RANDOM() AS age \
  49                                 FROM tweets \
  50                                 WHERE (last_tweeted_at IS NULL OR last_tweeted_at <= CURRENT_TIMESTAMP - INTERVAL '1 month') \
  51                         ) \
  52                         SELECT tweets.* FROM candidate_tweets \
  53                                 LEFT JOIN tweets ON candidate_tweets.id = tweets.id \
  54                                 ORDER BY age DESC LIMIT 1")
  55
  56                 return res
  57
  58         def _tweet(self, tweet):
  59                 logging.debug("Tweeting: %s" % tweet.message)
  60
  61                 # Update database status
  62                 self.db.execute("UPDATE tweets \
  63                         SET last_tweeted_at = CURRENT_TIMESTAMP, total_tweets = total_tweets + 1 \
  64                         WHERE id = %s", tweet.id)
  65
  66                 # Connect to twitter
  67                 twitter = twython.Twython(
  68                         self.settings.get("twitter_consumer_key"),
  69                         self.settings.get("twitter_consumer_secret"),
  70                         self.settings.get("twitter_%s_access_token" % tweet.account),
  71                         self.settings.get("twitter_%s_access_token_secret" % tweet.account),
  72                 )
  73
  74                 # Update status
  75                 twitter.update_status(status=tweet.message)