Initial commit.
This commit is contained in:
commit
18f4cbb5e9
|
@ -0,0 +1,26 @@
|
|||
# PyConFr Mastodon Bot
|
||||
|
||||
## Fetching the talks
|
||||
|
||||
PonyConf don't have an API so to fetch the talks data, run:
|
||||
|
||||
$ rsync root@deb2.afpy.org:/home/ponyconf/src/db.sqlite3 ./
|
||||
|
||||
|
||||
## Dry run
|
||||
|
||||
To test the pyconfr-to-mastodon bot, just use `--dry-run`, it won't connect to Mastodon.
|
||||
|
||||
Passing a `client_id`, `client_secret`, and `access_token` is mandatory though, so one can use:
|
||||
|
||||
$ python pyconfr-to-mastodon.py _ _ _ --dry-run
|
||||
|
||||
|
||||
To avoid boring "sleeps", just tell it to toot in the past, like `--ahead-days 999`.
|
||||
|
||||
|
||||
## Real use
|
||||
|
||||
Combined with `pass` one can use it like;
|
||||
|
||||
$ python pyconfr-to-mastodon.py --ahead-days 7 "$(pass afpy_mamot_client_id)" "$(pass afpy_mamot_client_secret)" "$(pass afpy_mamot_access_token)"
|
|
@ -0,0 +1,158 @@
|
|||
"""AFPy's bot to tweet about PyConFr."""
|
||||
|
||||
import argparse
|
||||
import logging
|
||||
import re
|
||||
import sqlite3
|
||||
from datetime import datetime, timedelta
|
||||
from time import sleep
|
||||
|
||||
from mastodon import Mastodon
|
||||
|
||||
|
||||
def convert_datetime(val):
|
||||
"""Convert ISO 8601 datetime to datetime.datetime object."""
|
||||
return datetime.fromisoformat(val.decode())
|
||||
|
||||
|
||||
sqlite3.register_converter("datetime", convert_datetime)
|
||||
|
||||
|
||||
def to_twitter_handle(twitter):
|
||||
"""Convert a twitter URL to a twitter handle."""
|
||||
username = re.sub("https://(www.)?twitter.com/", "", twitter, flags=re.I).lstrip(
|
||||
"@"
|
||||
)
|
||||
return f"@{username}@twitter.com"
|
||||
|
||||
|
||||
def to_mastodon_handle(mastodon):
|
||||
"""Convert a mastodon URL to a mastodon handle."""
|
||||
if match := re.match("https?://([^/]+)/@([^/]*)/?$", mastodon):
|
||||
return f"@{match[2]}@{match[1]}"
|
||||
return mastodon
|
||||
|
||||
|
||||
class Talk:
|
||||
def __init__(self, cursor, row):
|
||||
fields = [column[0] for column in cursor.description]
|
||||
for key, value in zip(fields, row):
|
||||
setattr(self, key, value)
|
||||
self.participants = self.participant.split("\x00")
|
||||
self.twitters = self.twitter.split("\x00")
|
||||
self.mastodons = self.mastodon.split("\x00")
|
||||
|
||||
@property
|
||||
def author(self):
|
||||
"""If there's multiple authors, mention them all, regardless how."""
|
||||
participants = []
|
||||
for participant, twitter, mastodon in zip(
|
||||
self.participants, self.twitters, self.mastodons
|
||||
):
|
||||
if mastodon:
|
||||
participants.append(to_mastodon_handle(mastodon))
|
||||
elif twitter:
|
||||
participants.append(to_twitter_handle(twitter))
|
||||
else:
|
||||
participants.append(participant)
|
||||
return ", ".join(participants)
|
||||
|
||||
|
||||
def get_talks(django_site_id=4):
|
||||
con = sqlite3.connect("db.sqlite3", detect_types=sqlite3.PARSE_DECLTYPES)
|
||||
con.row_factory = Talk
|
||||
with con:
|
||||
res = con.execute(
|
||||
f"""
|
||||
SELECT cfp_talk.start_date,
|
||||
cfp_room.name room,
|
||||
GROUP_CONCAT(cfp_participant.name, x'00') participant,
|
||||
GROUP_CONCAT(cfp_participant.mastodon, x'00') mastodon,
|
||||
GROUP_CONCAT(cfp_participant.twitter, x'00') twitter,
|
||||
cfp_talk.title,
|
||||
cfp_talkcategory.name category
|
||||
FROM cfp_talk
|
||||
JOIN cfp_room ON cfp_room.id = cfp_talk.room_id
|
||||
JOIN cfp_talk_speakers ON cfp_talk_speakers.talk_id = cfp_talk.id
|
||||
JOIN cfp_participant ON cfp_participant.id = cfp_talk_speakers.participant_id
|
||||
JOIN cfp_talkcategory ON cfp_talkcategory.id = cfp_talk.category_id
|
||||
WHERE cfp_talk.site_id={django_site_id}
|
||||
AND cfp_talk.accepted = 1
|
||||
AND cfp_talk.confirmed = 1
|
||||
GROUP BY cfp_talk.id
|
||||
ORDER BY cfp_talk.start_date
|
||||
"""
|
||||
)
|
||||
return res.fetchall()
|
||||
|
||||
|
||||
def parse_args():
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--django-site-id", type=int, default=4)
|
||||
parser.add_argument("client_id")
|
||||
parser.add_argument("client_secret")
|
||||
parser.add_argument("access_token")
|
||||
parser.add_argument(
|
||||
"--dry-run", action="store_true", help="Do not actually toot, just print."
|
||||
)
|
||||
parser.add_argument(
|
||||
"--ahead-days",
|
||||
type=int,
|
||||
default=0,
|
||||
help="How many days before the talk we should pouet about it.",
|
||||
)
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def toot_for(talk, ahead_days=0):
|
||||
"""Redact a toot for a given talk."""
|
||||
if ahead_days:
|
||||
return f"Dans {ahead_days} jours exactement, salle {talk.room} : « {talk.title} » présenté par {talk.author}"
|
||||
else:
|
||||
return f"Salle {talk.room} : « {talk.title} » présenté par {talk.author}, c'est maintenant !"
|
||||
|
||||
|
||||
def live_toot(talks, mastodon: Mastodon, ahead_days=0):
|
||||
for talk in talks:
|
||||
if talk.category == "Sprint":
|
||||
continue # We don't toot sprints.
|
||||
delta = talk.start_date - datetime.now() - timedelta(days=ahead_days)
|
||||
if delta.total_seconds() > 10:
|
||||
logging.info("Waiting %s for %s to start.", delta, talk.title)
|
||||
sleep(delta.total_seconds() - 2) # Wait for the talk to start.
|
||||
sleep(2) # This is just a fool guard so we can Ctrl-C easily anytime.
|
||||
mastodon.toot(toot_for(talk, ahead_days))
|
||||
|
||||
|
||||
class DryRunMastodon:
|
||||
"""For the --dry-run option."""
|
||||
|
||||
def __init__(*args, **kwargs):
|
||||
pass
|
||||
|
||||
def account_verify_credentials(self):
|
||||
return {"username": "(Dry run mode)"}
|
||||
|
||||
def toot(self, message):
|
||||
print(message)
|
||||
|
||||
|
||||
def main():
|
||||
args = parse_args()
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
talks = get_talks(args.django_site_id)
|
||||
mastodon_instance = DryRunMastodon if args.dry_run else Mastodon
|
||||
mastodon = mastodon_instance(
|
||||
api_base_url="https://mamot.fr",
|
||||
client_id=args.client_id,
|
||||
client_secret=args.client_secret,
|
||||
access_token=args.access_token,
|
||||
)
|
||||
logging.info(
|
||||
"Mastodon connected as %s", mastodon.account_verify_credentials()["username"]
|
||||
)
|
||||
live_toot(talks, mastodon, args.ahead_days)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
|
@ -0,0 +1 @@
|
|||
[tool.black]
|
|
@ -0,0 +1 @@
|
|||
Mastodon.py
|
Loading…
Reference in New Issue