Integrate Bluesky as alternative to Twitterbot (#2051)

---------

Co-authored-by: Gresille & Siffle <[email protected]>

backend/requirements.txt

@@ -60,4 +60,5 @@ requests==2.32.2
 tweepy==4.14.0
 # dotenv for .env file
 python-dotenv==1.0.0
-
+# AT protocol library for Bluesky
+atproto==0.0.58

backend/twitterbot/admin.py

@@ -8,7 +8,7 @@
 from django.contrib import admin
 from django.utils.html import format_html
 
-from .models.tweeted import TweetInfo
+from .models.history import TweetInfo
 
 
 @admin.register(TweetInfo)
@@ -32,13 +32,12 @@ class TwitterBotAdmin(admin.ModelAdmin):
     list_filter = ["bot_name"]
 
     @staticmethod
-    @admin.display(description="URL of the tweet")
+    @admin.display(description="Post URL")
     def get_twitter_link(obj):
         """Return the URI of the tweet."""
         return format_html(
-            '<a href="https://twitter.com/{}/status/{}" target="_blank">Tweet</a>',
-            obj.bot_name,
-            obj.tweet_id,
+            '<a href="{}" target="_blank">Link</a>',
+            obj.message_url,
         )
 
     @staticmethod

backend/twitterbot/client.py

@@ -0,0 +1,113 @@
+from functools import cached_property
+from pathlib import Path
+from typing import List, Optional
+
+import requests
+from django.conf import settings
+
+from tournesol.models import Entity
+
+
+class TournesolBotClient:
+    def __init__(self, account):
+
+        credentials = settings.TWITTERBOT_CREDENTIALS
+        if account not in credentials:
+            raise ValueError(f"No credentials found for {account} account!")
+
+        self.account_cred = credentials[account]
+        self.language = self.account_cred["LANGUAGE"]
+
+    @cached_property
+    def tweepy_api(self):
+        # Client for Twitter API v1.1
+        # We need this authentication also because to add media it only works with api v1.1
+
+        import tweepy  # pylint:disable=import-outside-toplevel
+        auth = tweepy.OAuth1UserHandler(
+            consumer_key=self.account_cred["CONSUMER_KEY"],
+            consumer_secret=self.account_cred["CONSUMER_SECRET"],
+            access_token=self.account_cred["ACCESS_TOKEN"],
+            access_token_secret=self.account_cred["ACCESS_TOKEN_SECRET"],
+        )
+        return tweepy.API(auth)
+
+    @cached_property
+    def tweepy_client(self):
+        # Client for Twitter API v2
+
+        import tweepy  # pylint:disable=import-outside-toplevel
+        return tweepy.Client(
+            consumer_key=self.account_cred["CONSUMER_KEY"],
+            consumer_secret=self.account_cred["CONSUMER_SECRET"],
+            access_token=self.account_cred["ACCESS_TOKEN"],
+            access_token_secret=self.account_cred["ACCESS_TOKEN_SECRET"],
+        )
+
+    @property
+    def bluesky_handle(self):
+        return self.account_cred["ATPROTO_HANDLE"]
+
+    @cached_property
+    def atproto_client(self):
+        from atproto import Client  # pylint:disable=import-outside-toplevel
+        client = Client()
+        client.login(
+            self.account_cred["ATPROTO_HANDLE"],
+            self.account_cred["ATPROTO_PASSWORD"],
+        )
+        return client
+
+    def create_tweet(self, text: str, media_files: Optional[List[Path]] = None):
+        if media_files is None:
+            media_ids = []
+        else:
+            medias = (self.tweepy_api.media_upload(filepath) for filepath in media_files)
+            media_ids = [media.media_id for media in medias]
+
+        resp = self.tweepy_client.create_tweet(
+            text=text,
+            media_ids=media_ids,
+        )
+        return resp.data["id"]
+
+    def create_bluesky_post(
+        self,
+        text,
+        embed_video: Optional[Entity] = None,
+        image_files: Optional[List[Path]] = None,
+        image_alts: Optional[List[str]] = None,
+    ):
+        from atproto import models  # pylint:disable=import-outside-toplevel
+        if image_files is None:
+            if embed_video is None:
+                embed = None
+            else:
+                preview_response = requests.get(
+                    f"https://api.tournesol.app/preview/entities/{embed_video.uid}",
+                    timeout=10,
+                )
+                preview_response.raise_for_status()
+                img_data = preview_response.content
+                thumb_blob = self.atproto_client.upload_blob(img_data).blob
+                embed = models.AppBskyEmbedExternal.Main(
+                    external=models.AppBskyEmbedExternal.External(
+                        title=embed_video.metadata.get("name", ""),
+                        description=embed_video.metadata.get("uploader", ""),
+                        uri=f"https://tournesol.app/entities/{embed_video.uid}",
+                        thumb=thumb_blob,
+                    )
+                )
+            resp = self.atproto_client.send_post(
+                text=text,
+                embed=embed,
+                langs=[self.language],
+            )
+        else:
+            resp = self.atproto_client.send_images(
+                text=text,
+                langs=[self.language],
+                images=[p.read_bytes() for p in image_files],
+                image_alts=image_alts,
+            )
+        return resp.uri

backend/twitterbot/management/commands/load_tweetinfo.py

@@ -4,7 +4,7 @@
 
 from tournesol.entities.video import YOUTUBE_UID_NAMESPACE
 from tournesol.models.entity import Entity
-from twitterbot.models.tweeted import TweetInfo
+from twitterbot.models.history import TweetInfo
 
 
 class Command(BaseCommand):

backend/twitterbot/management/commands/run_twitterbot.py

@@ -52,4 +52,4 @@ def handle(self, *args, **options):
 
             return
 
-        tweet_video_recommendation(bot_name, assumeyes=options["assumeyes"])
+        tweet_video_recommendation(bot_name, assumeyes=options["assumeyes"], dest=["bluesky"])

backend/twitterbot/management/commands/tweet_top_contributors.py

@@ -26,5 +26,8 @@ def add_arguments(self, parser):
         )
 
     def handle(self, *args, **options):
-
-        tweet_top_contributor_graph(options["bot_name"], assumeyes=options["assumeyes"])
+        tweet_top_contributor_graph(
+            options["bot_name"],
+            assumeyes=options["assumeyes"],
+            dest=["bluesky"],
+        )

backend/twitterbot/migrations/0003_tweetinfo_atproto_uri_alter_tweetinfo_tweet_id.py

@@ -0,0 +1,30 @@
+# Generated by Django 4.2.11 on 2025-01-27 22:37
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("twitterbot", "0002_rename_tweetedvideo_tweetinfo"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="tweetinfo",
+            name="atproto_uri",
+            field=models.CharField(
+                default=None,
+                help_text="URI of the post on the AT protocol network",
+                max_length=255,
+                null=True,
+            ),
+        ),
+        migrations.AlterField(
+            model_name="tweetinfo",
+            name="tweet_id",
+            field=models.CharField(
+                default=None, help_text="Tweet ID from Twitter URL", max_length=22, null=True
+            ),
+        ),
+    ]

backend/twitterbot/migrations/0004_alter_tweetinfo_bot_name_and_more.py

@@ -0,0 +1,46 @@
+# Generated by Django 4.2.18 on 2025-02-06 14:55
+
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("tournesol", "0061_comparisoncriteriascore_score_max_and_more"),
+        ("twitterbot", "0003_tweetinfo_atproto_uri_alter_tweetinfo_tweet_id"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="tweetinfo",
+            name="bot_name",
+            field=models.CharField(
+                blank=True,
+                choices=[
+                    ("@TournesolBot", "@TournesolBot"),
+                    ("@TournesolBotFR", "@TournesolBotFR"),
+                ],
+                help_text="Name of the bot",
+                max_length=200,
+                null=True,
+            ),
+        ),
+        migrations.AlterField(
+            model_name="tweetinfo",
+            name="datetime_tweet",
+            field=models.DateTimeField(
+                auto_now_add=True, help_text="Time when the video was posted", null=True
+            ),
+        ),
+        migrations.AlterField(
+            model_name="tweetinfo",
+            name="video",
+            field=models.ForeignKey(
+                help_text="Posted video",
+                on_delete=django.db.models.deletion.CASCADE,
+                related_name="tweets",
+                to="tournesol.entity",
+            ),
+        ),
+    ]

backend/twitterbot/models/history.py

@@ -0,0 +1,81 @@
+"""
+Models related to the Twitter bot.
+"""
+import re
+
+from django.db import models
+
+from tournesol.models import Entity
+
+BOT_NAME = [
+    ("@TournesolBot", "@TournesolBot"),
+    ("@TournesolBotFR", "@TournesolBotFR"),
+]
+
+
+class TweetInfo(models.Model):
+    """One tweeted video."""
+
+    video = models.ForeignKey(
+        Entity,
+        on_delete=models.CASCADE,
+        related_name="tweets",
+        help_text="Posted video",
+    )
+
+    tweet_id = models.CharField(
+        null=True,
+        default=None,
+        max_length=22,
+        help_text="Tweet ID from Twitter URL",
+    )
+
+    atproto_uri = models.CharField(
+        null=True,
+        default=None,
+        max_length=255,
+        help_text="URI of the post on the AT protocol network",
+    )
+
+    datetime_tweet = models.DateTimeField(
+        auto_now_add=True,
+        help_text="Time when the video was posted",
+        null=True,
+        blank=True,
+    )
+
+    bot_name = models.CharField(
+        null=True,
+        blank=True,
+        max_length=200,
+        help_text="Name of the bot",
+        choices=BOT_NAME,
+    )
+
+    def __str__(self):
+        return f"{self.video.uid} posted at {self.datetime_tweet}"
+
+    @property
+    def tweet_url(self):
+        if not self.tweet_id:
+            return None
+        return f"https://twitter.com/{self.bot_name}/status/{self.tweet_id}"
+
+    @property
+    def bluesky_url(self):
+        if not self.atproto_uri:
+            return None
+        match = re.match(
+            r"at://(?P<authority>.+)/(?P<collection>.+)/(?P<key>.+)",
+            self.atproto_uri,
+        )
+        if not match or match.group("collection") != "app.bsky.feed.post":
+            return None
+        return f"https://bsky.app/profile/{match.group('authority')}/post/{match.group('key')}"
+
+    @property
+    def message_url(self):
+        bluesky_url = self.bluesky_url
+        if bluesky_url:
+            return self.bluesky_url
+        return self.tweet_url