Implement pinning to social networks

This is mostly untested so far, since we need some real posts to pin
before we can do that, but "mostly works".

Author: mhagander

pgweb/news/admin.py

@@ -2,8 +2,11 @@
 from django import forms
 
 from pgweb.util.admin import PgwebAdmin
+from pgweb.util.moderation import ModerationState
 from pgweb.core.models import OrganisationEmail
-from .models import NewsArticle, NewsTag
+from .models import NewsArticle, NewsTag, PinnedNewsArticle
+
+from datetime import datetime, timedelta
 
 
 class NewsArticleAdminForm(forms.ModelForm):
@@ -20,14 +23,36 @@ class NewsArticleAdmin(PgwebAdmin):
     list_filter = ('modstate', )
     filter_horizontal = ('tags', )
     search_fields = ('content', 'title', )
-    exclude = ('modstate', 'firstmoderator', )
+    exclude = ('modstate', 'firstmoderator', 'ispinned', )
     form = NewsArticleAdminForm
 
 
+class PinnedNewsArticleAdminForm(forms.ModelForm):
+    model = PinnedNewsArticle
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.fields['pinnedarticle'].queryset = NewsArticle.objects.filter(modstate=ModerationState.APPROVED, date__gt=datetime.now() - timedelta(days=365)).order_by('-date')
+        self.fields['pinnedarticle'].widget.can_delete_related = False
+        self.fields['pinnedarticle'].widget.can_add_related = False
+
+
+class PinnedNewsArticleAdmin(admin.ModelAdmin):
+    exclude = ('pinnedtoproviders', )
+    form = PinnedNewsArticleAdminForm
+
+    def has_add_permission(self, request, obj=None):
+        return False
+
+    def has_delete_permission(self, request, obj=None):
+        return False
+
+
 class NewsTagAdmin(PgwebAdmin):
     list_display = ('urlname', 'name', 'description')
     filter_horizontal = ('allowed_orgs', )
 
 
 admin.site.register(NewsArticle, NewsArticleAdmin)
 admin.site.register(NewsTag, NewsTagAdmin)
+admin.site.register(PinnedNewsArticle, PinnedNewsArticleAdmin)

pgweb/news/management/commands/social_post.py

@@ -13,7 +13,7 @@
 import time
 
 from pgweb.util.moderation import ModerationState
-from pgweb.news.models import NewsArticle
+from pgweb.news.models import NewsArticle, PinnedNewsArticle
 from pgweb.util.socialposter import get_all_providers
 
 
@@ -50,3 +50,11 @@ def handle(self, *args, **options):
                     if postid is not None:
                         a.postedto[p.name] = postid
                         a.save(update_fields=['postedto'])
+
+        # Pin or unpin any articles as needed
+        pna = PinnedNewsArticle.objects.select_related('pinnedarticle').only('pinnedarticle', 'pinnedtoproviders', 'pinnedarticle__postedto').all()[0]
+        for p in allproviders:
+            if pna.pinnedtoproviders.get(p.name, None) != pna.pinnedarticle.postedto.get(p.name, None):
+                if p.set_pin(pna.pinnedarticle.postedto.get(p.name, None)):
+                    pna.pinnedtoproviders[p.name] = pna.pinnedarticle.postedto.get(p.name, None)
+                    pna.save(update_fields=['pinnedtoproviders'])

pgweb/news/migrations/0009_pinnednewsarticle.py

@@ -0,0 +1,27 @@
+# Generated by Django 5.2.10 on 2026-04-06 20:14
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+def add_record(apps, schema_editor):
+    apps.get_model('news', 'PinnedNewsArticle')(pinnedarticle=None).save()
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('news', '0008_multisocial'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='PinnedNewsArticle',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('pinnedtoproviders', models.JSONField(blank=True, default=dict)),
+                ('pinnedarticle', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, to='news.newsarticle')),
+            ],
+        ),
+        migrations.RunPython(add_record),
+    ]

pgweb/news/models.py

@@ -2,6 +2,7 @@
 from datetime import date
 from pgweb.core.models import Organisation, OrganisationEmail
 from pgweb.core.text import ORGANISATION_HINT_TEXT
+from django.core.validators import ValidationError
 from pgweb.util.moderation import TristateModerateModel, ModerationState, TwoModeratorsMixin
 from django.template.defaultfilters import slugify
 
@@ -113,3 +114,18 @@ def get_field_description(self, f):
             return 'Content preview'
         elif f == 'permanenturl':
             return 'Permanent URL'
+
+
+class PinnedNewsArticle(models.Model):
+    pinnedarticle = models.ForeignKey(NewsArticle, null=True, blank=True, on_delete=models.SET_NULL)
+    pinnedtoproviders = models.JSONField(null=False, blank=True, default=dict)
+
+    def save(self, *args, **kwargs):
+        if not self.pk and PinnedNewsArticle.objects.exists():
+            raise ValidationError("Only one PinnedNewsArticle may exist!")
+        return super().save(*args, **kwargs)
+
+    def __str__(self):
+        if self.pinnedarticle:
+            return str(self.pinnedarticle)
+        return 'No article currently pinned'

pgweb/util/socialposter.py

@@ -22,6 +22,9 @@ def post(self):
     def register(self, clientname):
         raise NotImplementedError
 
+    def set_pin(self, postid):
+        raise NotImplementedError
+
 
 class Mastodon(SocialPoster):
     name = 'mastodon'
@@ -42,6 +45,57 @@ def post(self, text):
 
         return r.json()['id']
 
+    def set_pin(self, postid):
+        # First we have to see if there is an existing pin that has to be removed. To do that, we
+        # have to fetch our account id.
+        r = requests.get(
+            '{}/api/v1/accounts/verify_credentials '.format(self.settings.MASTODON_BASEURL),
+            headers={'Authorization': 'Bearer {}'.format(self.settings.MASTODON_TOKEN)},
+            timeout=10,
+        )
+        if r.status_code != 200:
+            print("Failed to get mastodon credentials: {}".format(r.text))
+            return False
+
+        # Next, get the currently pinned ones
+        r = requests.get(
+            '{}/api/v1/accounts/{}/statuses'.format(self.settings.MASTODON_BASEURL, r.json()['id']),
+            headers={'Authorization': 'Bearer {}'.format(self.settings.MASTODON_TOKEN)},
+            params={'pinned': 'true'},
+            timeout=10,
+        )
+        if r.status_code != 200:
+            print("Failed to get list of mastodon pins: {}".format(r.text))
+            return False
+
+        found = False
+        for p in r.json():
+            if p['id'] == postid:
+                # Already pinned!
+                found = True
+            else:
+                # This should be unpinned
+                r2 = requests.post(
+                    '{}/api/v1/statuses{}/unpin'.format(self.settings.MASTODON_BASEURL, p['id']),
+                    headers={'Authorization': 'Bearer {}'.format(self.settings.MASTODON_TOKEN)},
+                    timeout=10,
+                )
+                if r.status_code != 200:
+                    print("Failed to unpin from mastodon: {}".format(r.text))
+
+        if not found:
+            # Not already pinned, so pin!
+            r2 = requests.post(
+                '{}/api/v1/statuses{}/pin'.format(self.settings.MASTODON_BASEURL, postid),
+                headers={'Authorization': 'Bearer {}'.format(self.settings.MASTODON_TOKEN)},
+                timeout=10,
+            )
+            if r.status_code != 200:
+                print("Failed to pin to mastodon: {}".format(r.text))
+                return False
+
+        return True
+
     def register(self, clientname):
         toadd = io.StringIO()
 
@@ -156,6 +210,55 @@ def post(self, text):
 
         return r.json()['uri']
 
+    def set_pin(self, postid):
+        # Get our profile, so we can "patch" it
+        r = requests.get(
+            'https://bsky.social/xrpc/com.atproto.repo.getRecord',
+            headers={'Authorization': 'Bearer {}'.format(self.token)},
+            params={'repo': self.repo, 'collection': 'app.bsky.actor.profile', 'rkey': 'self'},
+            timeout=10,
+        )
+        if r.status_code != 200:
+            print("Failed to get bluesky profile: {}".format(r.text))
+            return False
+        record = r.json()['value']
+        if postid:
+            # We have the uri, but we need both the uri and the cid, so fetch the post
+            r2 = requests.get(
+                'https://bsky.social/xrpc/app.bsky.feed.getPosts',
+                headers={'Authorization': 'Bearer {}'.format(self.token)},
+                params={'uris': postid},
+                timeout=10,
+            )
+            if r.status_code != 200:
+                print("Failed to read back bluesky post {}: {}".format(postid, r.text))
+                return False
+
+            cid = r2.json()['posts'][0]['cid']
+
+            record['pinnedPost'] = {
+                'uri': postid,
+                'cid': cid,
+            }
+        else:
+            if 'pinnedPost' in record:
+                del record['pinnedPost']
+
+        if record != r.json()['value']:
+            # Some changes, so we need to update
+            r = requests.post(
+                'https://bsky.social/xrpc/com.atproto.repo.putRecord',
+                headers={'Authorization': 'Bearer {}'.format(self.token)},
+                json={'repo': self.repo, 'collection': 'app.bsky.actor.profile', 'rkey': 'self', 'record': record},
+                timeout=10,
+            )
+            if r.status_code != 200:
+                print("Failed to save pack bluesky profile for pinned posts: {}".format(r.text))
+                return False
+            return True
+
+        return True
+
     def register(self, clientname):
         if getattr(self.settings, 'BLUESKY_USER', None) is None or getattr(self.settings, 'BLUESKY_PASSWORD', None) is None:
             return "For bluesky, add an 'app password' from 'Settings' -> 'Privacy and Security'.\nRegister the account email as BLUESKY_USER and the app password as BLUESKY_PASSWORD.\n"