pyconfr-to-mastodon/pyconfr-to-mastodon.py

"""AFPy's bot to tweet about PyConFr."""

import argparse
import logging
import re
import sqlite3
from datetime import datetime, timedelta, timezone
from time import sleep
from pathlib import Path
from zoneinfo import ZoneInfo
from contextlib import suppress

from mastodon import Mastodon
from jinja2 import Environment, FileSystemLoader
from html2image import Html2Image


def generate_file(output_path, title, author, type, date, salle):
    environment = Environment(loader=FileSystemLoader("template/"))
    template = environment.get_template("template.html")

    rend = template.render(
        title=title, author=author, type=type, date=date, salle=salle
    )

    hti = Html2Image()
    hti.load_file("template/illustration_1.png")
    hti.load_file("template/logo_pyconfr_1.png")
    hti.screenshot(
        html_str=rend,
        css_file="template/styles.css",
        save_as=output_path,
        size=(1200, 675),
    )


def convert_datetime(val):
    """Convert ISO 8601 datetime to datetime.datetime object."""
    return datetime.fromisoformat(val.decode() + "+00:00")


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.id,
           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(timezone.utc) - timedelta(days=ahead_days)
        )
        if delta.total_seconds() < -300:
            logging.info("Skipping %s (already happened).", talk.title)
            continue
        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.
        png_name = f"talk-{talk.id}.png"
        with suppress(FileNotFoundError):
            Path(png_name).unlink()
        generate_file(
            output_path=png_name,
            title=talk.title,
            author=", ".join(talk.participants),
            type=talk.category.split()[0],
            date=talk.start_date.astimezone(ZoneInfo("Europe/Paris")).strftime(
                "%A %d %B %HH%M"
            ),
            salle="Salle " + talk.room.split("/")[0],
        )
        media = mastodon.media_post("rend.png")
        mastodon.status_post(toot_for(talk, ahead_days), media_ids=[media])


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 media_post(self, *args, **kwargs):
        print("Would upload media to Mastodon.")

    def status_post(self, *args, **kwargs):
        print("Would tooting", args, kwargs)


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()