Implement social monitoring and Slack notifications

.env-dist

@@ -4,4 +4,6 @@ DJANGO_DEBUG=true
 SENDY_API_KEY=
 SENDY_ENDPOINT_URL=
 SLACK_OAUTH_TOKEN=
+SLACK_CHANNEL_ID=
 TITO_SECURITY_TOKEN=
+MASTODON_ACCESS_TOKEN=

config/settings.py

@@ -71,6 +71,7 @@
     "tickets",
     "titowebhooks",
     "travel_safety",
+    "social_monitor",
 ]
 
 MIDDLEWARE = [
@@ -236,6 +237,7 @@
 # Slack settings
 
 SLACK_OAUTH_TOKEN = env("SLACK_OAUTH_TOKEN", default="")
+SLACK_CHANNEL_ID = env("SLACK_CHANNEL_ID", default="")
 
 # Conference settings
 
@@ -253,3 +255,8 @@
 TAILWIND_CLI_PATH = env.str("TAILWIND_CLI_PATH", default="~/.local/bin/")
 TAILWIND_CLI_SRC_CSS = env.str("TAILWIND_CLI_SRC_CSS", default="frontend/index.css")
 TAILWIND_CLI_VERSION = env.str("TAILWIND_CLI_VERSION", default="4.1.18")
+
+# Social monitor
+
+MASTODON_ACCESS_TOKEN = env("MASTODON_ACCESS_TOKEN", default="")
+MASTODON_API_BASE_URL = env("MASTODON_API_BASE_URL", default="https://mastodon.social")

pyproject.toml

@@ -24,6 +24,8 @@ dependencies = [
     "requests",
     "rich",
     "whitenoise",
+    "mastodon-py>=2.1.4",
+    "slack-sdk>=3.41.0",
 ]
 
 [tool.bumpver]

social_monitor/admin.py

@@ -0,0 +1,22 @@
+from django.contrib import admin
+
+from .models import PlatformHashTag, SocialPlatform
+
+
+@admin.register(SocialPlatform)
+class SocialPlatformAdmin(admin.ModelAdmin):
+    list_display = (
+        "name",
+        "last_seen",
+        "get_mentions",
+    )
+
+
+@admin.register(PlatformHashTag)
+class PlatformHashTagAdmin(admin.ModelAdmin):
+    list_display = (
+        "platform",
+        "query",
+        "last_seen",
+        "is_active",
+    )

social_monitor/apps.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class SocialMonitorConfig(AppConfig):
+    default_auto_field = 'django.db.models.BigAutoField'
+    name = 'social_monitor'

social_monitor/management/commands/social.py

@@ -0,0 +1,108 @@
+import djclick as click
+from django_q.models import Schedule
+from rich import print
+
+from social_monitor.models import PlatformHashTag, SocialPlatform
+
+
+@click.group()
+def command():
+    """Social monitor command"""
+    pass
+
+
+@command.command()
+def monitor():
+    """Set up a periodic task to monitor for mentions and hashtags"""
+    name = "social-monitor"
+    schedule, created = Schedule.objects.update_or_create(
+        name=name,
+        defaults={
+            "func": "social_monitor.mastodon_client.collect_social_activity",
+            "schedule_type": Schedule.HOURLY,
+        }
+    )
+
+    action = "Created" if created else "Updated"
+    print(f"[green]{action} schedule for {name}[/green]")
+
+@command.command()
+@click.option("--fetch", "fetch_flag", is_flag=True, help="fetch all platforms")
+@click.option("--add", "add_platform_name", help="add new platform")
+@click.option("--remove", "remove_platform_name", help="remove new platform")
+@click.option("--fetch-mentions", "fetch_mentions", default=True, help="fetch mentions")
+def platform(fetch_flag, add_platform_name, remove_platform_name, fetch_mentions):
+    """add social platform"""
+
+    if fetch_flag:
+        platforms = SocialPlatform.objects.all()
+        if platforms:
+            for i, p in enumerate(platforms):
+                print(f"{i+1}) [green]{p}[/green]")
+        else:
+            print("[red]No platforms found![/red]")
+            return
+
+    if add_platform_name:
+        p, created = SocialPlatform.objects.get_or_create(
+            name=add_platform_name,
+            get_mentions=fetch_mentions
+        )
+
+        if created:
+            print(f"[green]Platform `{add_platform_name}` added![/green]")
+        else:
+            print(f"[yellow]Platform `{add_platform_name}` already exists.[/yellow]")
+
+    if remove_platform_name:
+        try:
+            p = SocialPlatform.objects.get(name=remove_platform_name)
+            p.delete()
+            print(f"[green]Platform `{remove_platform_name} removed![/green]`")
+        except SocialPlatform.DoesNotExist:
+            print(f"[red]Platform `{remove_platform_name}` does not exists![/red]`")
+
+@command.command()
+@click.argument('platform_name')
+@click.argument("query_str", required=False)
+@click.option("--add", "add_flag", is_flag=True, help="add query to social platform")
+@click.option("--remove", "remove_flag", is_flag=True, help="remove query from social platform")
+@click.option("--fetch", "fetch_query", is_flag=True, help="fetch queries")
+def query(platform_name, query_str, add_flag, remove_flag, fetch_query):
+    """add social query"""
+    try:
+        p = SocialPlatform.objects.get(name=platform_name)
+    except SocialPlatform.DoesNotExist:
+        print(f"[red]Platform `{platform_name}` does not exist![/red]")
+        return
+
+    if add_flag:
+        if PlatformHashTag.objects.filter(platform=p, query=query_str).exists():
+            print(f"[yellow]Query `{query_str}` already exists for {platform_name}.[/yellow]")
+        else:
+            PlatformHashTag.objects.create(platform=p, query=query_str)
+            print(f"[green]Query `{query_str}` added to {platform_name}![/green]")
+
+    if remove_flag:
+        try:
+            query_obj = PlatformHashTag.objects.get(platform=p, query=query_str)
+            query_obj.delete()
+            print(f"[green]Query `{query_str}` removed from {platform_name}![/green]")
+        except PlatformHashTag.DoesNotExist:
+            print(f"[red]Query `{query_str}` not found for {platform_name}![/red]")
+            return
+
+    if fetch_query:
+        try:
+            query_obj = PlatformHashTag.objects.filter(platform=p)
+            for q in query_obj:
+                print(f"[green]{q}[/green]")
+            return
+        except PlatformHashTag.DoesNotExist:
+            print(f"[red]Query `{query_str}` not found for {platform_name}![/red]")
+
+
+    if not add_flag and not remove_flag:
+        print(f"[blue]No action specified. Use --add or --remove for `{query_str}` on {platform_name}.[/blue]")
+
+

social_monitor/mastodon_client.py

@@ -0,0 +1,113 @@
+import logging
+from typing import List
+
+from django.conf import settings
+from mastodon import Mastodon
+
+from social_monitor.models import PlatformHashTag, SocialPlatform
+from social_monitor.notification import send_slack_notification
+from social_monitor.schemas import ItemType, SocialItem
+
+PLATFORM_NAME = "mastodon"
+
+
+mastodon = Mastodon(
+    access_token=settings.MASTODON_ACCESS_TOKEN,
+    api_base_url=settings.MASTODON_API_BASE_URL,
+)
+
+
+def _get_hashtags():
+    platform = SocialPlatform.objects.get(name=PLATFORM_NAME)
+    query_obj = PlatformHashTag.objects.filter(platform=platform)
+    return query_obj
+
+
+def fetch_mentions(limit: int = 5) -> List[SocialItem]:
+    """fetch the most recent mentions of the current Mastodon account."""
+    mastodon_mentions = []
+    try:
+        platform = SocialPlatform.objects.get(name=PLATFORM_NAME)
+
+        if platform.get_mentions:
+            mentions = mastodon.notifications(types=['mention'], limit=limit, since_id=0 if platform.last_seen is None else int(platform.last_seen))
+            for mention in mentions:
+                mastodon_mentions.append(
+                    SocialItem(
+                        id=int(mention.id),
+                        platform=PLATFORM_NAME,
+                        type=ItemType.MENTION,
+                        author=mention.account.acct,
+                        content=mention.status.content,
+                        url=mention.status.url,
+                        tag=ItemType.MENTION.name,
+                        created_at=mention.status.created_at
+                    )
+                )
+        else:
+            logging.info(f"mentions for `{platform.name}` are not being collected. Please enable it first.")
+    except Exception as e:
+        logging.error(e)
+
+    return mastodon_mentions
+
+
+def fetch_posts(limit: int = 5) -> List[SocialItem]:
+    """fetch the most recent posts containing a specific hashtag from Mastodon."""
+    mastodon_posts = []
+    try:
+        queries = _get_hashtags()
+        for query in queries:
+            if query.is_active:
+                posts = mastodon.timeline_hashtag(query.query, limit=limit, since_id=0 if query.last_seen is None else int(query.last_seen))
+                for post in posts:
+                    mastodon_posts.append(
+                        SocialItem(
+                            id=int(post.id),
+                            platform=PLATFORM_NAME,
+                            type=ItemType.HASHTAG,
+                            author=post.account.acct,
+                            content=post.content,
+                            url=post.url,
+                            tag=query.query,
+                            created_at=post.created_at,
+                        )
+                    )
+            else:
+                logging.info(f"`{query.query}` not active")
+    except Exception as e:
+        logging.error(e)
+
+    return mastodon_posts
+
+
+
+
+def collect_social_activity():
+
+    mentions = fetch_mentions()
+    posts = fetch_posts()
+
+    all_activities = mentions + posts
+    last_seen_marked = []
+
+    if all_activities:
+        platform = SocialPlatform.objects.get(name=PLATFORM_NAME)
+        for activity in all_activities:
+            notified = send_slack_notification(activity)
+
+            if notified:
+                if activity.type == ItemType.MENTION and ItemType.MENTION not in last_seen_marked:
+                    platform.last_seen = activity.id
+                    platform.save()
+                    last_seen_marked.append(ItemType.MENTION)
+
+                if activity.type == ItemType.HASHTAG and activity.tag not in last_seen_marked:
+                    social_query = PlatformHashTag.objects.get(platform=platform, query=activity.tag)
+                    social_query.last_seen = activity.id
+                    social_query.save()
+                    last_seen_marked.append(activity.tag)
+            else:
+                logging.error(f"error while sending notification for `{activity.id}` `{activity.tag}`")
+
+

social_monitor/migrations/0001_initial.py

@@ -0,0 +1,40 @@
+# Generated by Django 5.1.15 on 2026-03-31 18:43
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='SocialPlatform',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('name', models.CharField(max_length=100, unique=True)),
+                ('get_mentions', models.BooleanField(default=True)),
+                ('last_seen', models.CharField(blank=True, help_text='Last seen timestamp for mentions', max_length=200, null=True)),
+                ('created_at', models.DateTimeField(auto_now_add=True)),
+            ],
+        ),
+        migrations.CreateModel(
+            name='PlatformHashTag',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('query', models.CharField(max_length=200)),
+                ('last_seen', models.CharField(blank=True, help_text='Last seen timestamp for the query', max_length=200, null=True)),
+                ('is_active', models.BooleanField(default=True)),
+                ('created_at', models.DateTimeField(auto_now_add=True)),
+                ('updated_at', models.DateTimeField(auto_now=True)),
+                ('platform', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='social_monitor.socialplatform')),
+            ],
+            options={
+                'unique_together': {('platform', 'query')},
+            },
+        ),
+    ]

social_monitor/models.py

@@ -0,0 +1,38 @@
+from django.db import models
+
+
+class SocialPlatform(models.Model):
+    name = models.CharField(max_length=100, unique=True)
+    get_mentions = models.BooleanField(default=True)
+    last_seen = models.CharField(
+        max_length=200,
+        blank=True,
+        null=True,
+        help_text="Last seen timestamp for mentions",
+    )
+    created_at = models.DateTimeField(auto_now_add=True)
+
+    def __str__(self):
+        return self.name
+
+
+class PlatformHashTag(models.Model):
+    platform = models.ForeignKey(SocialPlatform, on_delete=models.CASCADE)
+
+    query = models.CharField(max_length=200)
+    last_seen = models.CharField(
+        max_length=200,
+        blank=True,
+        null=True,
+        help_text="Last seen timestamp for the query",
+    )
+    is_active = models.BooleanField(default=True)
+
+    created_at = models.DateTimeField(auto_now_add=True)
+    updated_at = models.DateTimeField(auto_now=True)
+
+    class Meta:
+        unique_together = ('platform', 'query')
+
+    def __str__(self):
+        return f"{self.platform.name} -> #{self.query}"