src/backend/resolver.py \
src/backend/settings.py \
src/backend/talk.py \
- src/backend/tweets.py \
+ src/backend/toots.py \
src/backend/util.py \
src/backend/wiki.py \
src/backend/zeiterfassung.py
from . import resolver
from . import settings
from . import talk
-from . import tweets
+from . import toots
from . import util
from . import wiki
from . import zeiterfassung
"send-message" : self.messages.send_cli,
"send-all-messages" : self.messages.queue.send_all,
"test-ldap" : self.accounts.test_ldap,
- "tweet" : self.tweets.tweet,
+ "toot" : self.toots.toot,
"update-blog-feeds" : self.blog.update_feeds,
}
return resolver.Resolver(tries=2, timeout=2, domains=[])
@lazy_property
- def tweets(self):
- return tweets.Tweets(self)
+ def toots(self):
+ return toots.Toots(self)
async def cleanup(self):
# Cleanup message queue
--- /dev/null
+#!/usr/bin/python3
+
+import datetime
+import logging
+import mastodon
+
+from .misc import Object
+
+class Toots(Object):
+ async def toot(self):
+ """
+ Sends a random promotional toot
+ """
+ # Do not toot too often
+ if self.has_had_recent_activity(days=3):
+ logging.debug("Won't toot because we recently did it")
+ return
+
+ # Do not toot when there was a blog post
+ if self.backend.blog.has_had_recent_activity(hours=24):
+ logging.debug("Won't toot because the blog has had activity")
+ return
+
+ # Select a toot
+ toot = self._get_random_toot()
+ if not toot:
+ logging.warning("Could not find anything to toot")
+ return
+
+ # Toot the toot!
+ with self.db.transaction():
+ self._toot(toot)
+
+ def has_had_recent_activity(self, **kwargs):
+ t = datetime.timedelta(**kwargs)
+
+ res = self.db.get("SELECT COUNT(*) AS count FROM toots \
+ WHERE last_tooted_at IS NOT NULL AND last_tooted_at >= NOW() - %s", t)
+
+ if res and res.count > 0:
+ return True
+
+ return False
+
+ def _get_random_toot(self):
+ res = self.db.get(
+ "WITH candidate_toots AS (SELECT id, \
+ (CURRENT_TIMESTAMP - COALESCE(last_tooted_at, '1970-01-01')) * RANDOM() AS age \
+ FROM toots \
+ WHERE (last_tooted_at IS NULL OR last_tooted_at <= CURRENT_TIMESTAMP - INTERVAL '1 month') \
+ ) \
+ SELECT toots.* FROM candidate_toots \
+ LEFT JOIN toots ON candidate_toots.id = toots.id \
+ ORDER BY age DESC LIMIT 1")
+
+ return res
+
+ def _toot(self, toot):
+ logging.debug("Posting: %s" % toot.message)
+
+ # Update database status
+ self.db.execute("UPDATE toots \
+ SET last_tooted_at = CURRENT_TIMESTAMP, total_toots = total_toots + 1 \
+ WHERE id = %s", toot.id)
+
+ # Connect to Mastodon
+ conn = mastodon.Mastodon(
+ client_id=self.settings.get("mastodon-client-key"),
+ client_secret=self.settings.get("mastodon-client-secret"),
+ access_token=self.settings.get("mastodon-access-token"),
+ api_base_url="https://social.ipfire.org",
+ )
+
+ # Toot!
+ conn.toot(toot.message)
+++ /dev/null
-#!/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=3):
- 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)
# Check mirrors once every 30 min
#*/30 * * * * nobody ipfire.org --logging=error check-mirrors
-# Tweet once a week
-#0 8 * * * nobody sleep ${RANDOM} && ipfire.org tweet
+# Toot once a day
+#0 8 * * * nobody sleep ${RANDOM} && ipfire.org toot
projects / ipfire.org.git / commitdiff
