1.0.0: Working, but not perfect

.dockerignore

@@ -0,0 +1,8 @@
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/

.env.example

@@ -0,0 +1,11 @@
+LOG_LEVEL=INFO
+LOG_DISCORD_WEBHOOK_URL=https://discord.com/api/webhooks/YYYYYYYY/YYYYYYYY
+LOG_DISCORD_WEBHOOK_LEVEL=WARNING
+USERS_ALL=Mxtive,spectatorindex,Breaking911
+USERS_TOP=X,XData
+USERS_MEDIA=archillect
+COOLDOWN_MAX_TIME=60
+X_CSRF_TOKEN=XXXXXXXX
+X_AUTH_TOKEN=XXXXXXXX
+X_BEARER_TOKEN=XXXXXXXX
+DISCORD_WEBHOOK_URL=https://discord.com/api/webhooks/XXXXXXXX/XXXXXXXX

.github/FUNDING.yml

@@ -0,0 +1 @@
+custom: ["https://cash.app/$EthanChrisp", "https://venmo.com/u/Mxtive"]

.github/workflows/ci.yml

@@ -0,0 +1,39 @@
+name: ci
+
+on:
+  push:
+    branches:
+      - "main"
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      -
+        name: Checkout
+        uses: actions/checkout@v3
+      -
+        name: Login to Docker Hub
+        uses: docker/login-action@v2
+        with:
+          username: ${{ secrets.DOCKER_HUB_USERNAME }}
+          password: ${{ secrets.DOCKER_HUB_ACCESS_TOKEN }}
+      -
+        name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v2
+      -
+        name: Build and push
+        uses: docker/build-push-action@v3
+        with:
+          context: .
+          file: ./Dockerfile
+          push: true
+          tags: ${{ secrets.DOCKER_HUB_USERNAME }}/bluebird:latest
+      -
+        name: Docker Hub Description
+        uses: peter-evans/dockerhub-description@v3
+        with:
+          username: ${{ secrets.DOCKER_HUB_USERNAME }}
+          password: ${{ secrets.DOCKER_HUB_ACCESS_TOKEN }}
+          repository: ${{ secrets.DOCKER_HUB_USERNAME }}/bluebird
+          short-description: ${{ github.event.repository.description }}

Dockerfile

@@ -0,0 +1,16 @@
+FROM python:3.12.2-slim-bullseye
+
+WORKDIR /bluebird
+
+# Install and configure Poetry
+# https://github.com/python-poetry/poetry
+RUN pip install poetry
+RUN poetry config virtualenvs.create false
+
+# Install dependencies
+COPY pyproject.toml pyproject.toml
+RUN poetry install --no-root
+
+COPY . .
+
+CMD [ "python", "bluebird.py" ]

README.md

@@ -1,2 +1,56 @@
 # Bluebird
-TODO
+
+![GitHub Workflow Status](https://img.shields.io/github/actions/workflow/status/EthanC/Bluebird/ci.yml?branch=main) ![Docker Pulls](https://img.shields.io/docker/pulls/ethanchrisp/bluebird?label=Docker%20Pulls) ![Docker Image Size (tag)](https://img.shields.io/docker/image-size/ethanchrisp/bluebird/latest?label=Docker%20Image%20Size)
+
+Bluebird monitors users on X and reports new posts via Discord.
+
+## Setup
+
+Although not required, a [Discord Webhook](https://support.discord.com/hc/en-us/articles/228383668-Intro-to-Webhooks) is recommended for notifications.
+
+An X account is required for API credentials. It is recommended to use a throwaway account due to use of the internal API.
+
+**Environment Variables:**
+
+-   `LOG_LEVEL`: [Loguru](https://loguru.readthedocs.io/en/stable/api/logger.html) severity level to write to the console.
+-   `LOG_DISCORD_WEBHOOK_URL`: [Discord Webhook](https://support.discord.com/hc/en-us/articles/228383668-Intro-to-Webhooks) URL to receive log events.
+-   `LOG_DISCORD_WEBHOOK_LEVEL`: Minimum [Loguru](https://loguru.readthedocs.io/en/stable/api/logger.html) severity level to forward to Discord.
+-   `USERS_ALL`: Comma-separated list of [X](https://x.com/) usernames to monitor for all posts.
+-   `USERS_TOP`: Comma-separated list of [X](https://x.com/) usernames to monitor for top-level posts only.
+-   `USERS_MEDIA`: Comma-separated list of [X](https://x.com/) usernames to monitor for media posts only.
+-   `COOLDOWN_MAX_TIME`: Maximum randomized cooldown time between checking for new posts (default is 60).
+-   `X_CSRF_TOKEN`: CSRF Token obtained via request inspection on X.
+-   `X_AUTH_TOKEN`: Cookie Auth Token obtained via request inspection on X.
+-   `X_BEARER_TOKEN`: Authentication Bearer Token obtained via request inspection on X.
+-   `DISCORD_WEBHOOK_URL`: [Discord Webhook](https://support.discord.com/hc/en-us/articles/228383668-Intro-to-Webhooks) URL to receive available username notifications.
+
+### Docker (Recommended)
+
+Modify the following `docker-compose.yml` example file, then run `docker compose up`.
+
+```yml
+services:
+  bluebird:
+    container_name: bluebird
+    image: ethanchrisp/bluebird:latest
+    environment:
+      LOG_LEVEL: INFO
+      LOG_DISCORD_WEBHOOK_URL: https://discord.com/api/webhooks/YYYYYYYY/YYYYYYYY
+      LOG_DISCORD_WEBHOOK_LEVEL: WARNING
+      USERS_ALL: Mxtive,spectatorindex,Breaking911
+      USERS_TOP: X,XData
+      USERS_MEDIA: archillect
+      COOLDOWN_MAX_TIME: 60
+      X_CSRF_TOKEN: XXXXXXXX
+      X_AUTH_TOKEN: XXXXXXXX
+      X_BEARER_TOKEN: XXXXXXXX
+      DISCORD_WEBHOOK_URL: https://discord.com/api/webhooks/XXXXXXXX/XXXXXXXX
+```
+
+### Standalone
+
+Bluebird is built for [Python 3.12](https://www.python.org/) or greater.
+
+1. Install required dependencies using [Poetry](https://python-poetry.org/): `poetry install --no-root`
+2. Rename `.env.example` to `.env`, then provide the environment variables.
+3. Start Bluebird: `python bluebird.py`

bluebird.py

@@ -0,0 +1,256 @@
+import logging
+from os import environ
+from random import randint
+from sys import exit, stdout
+from threading import Thread
+from time import sleep
+from typing import Self
+
+import dotenv
+from discord_webhook import DiscordEmbed, DiscordWebhook
+from loguru import logger
+from loguru_discord import DiscordSink
+
+from handlers import Intercept
+from services import X
+
+
+class Bluebird:
+    """
+    Monitor users on X and report new posts via Discord.
+
+    https://github.com/EthanC/Bluebird
+    """
+
+    def Initialize(self: Self) -> None:
+        """Initialize Bluebird and begin functionality."""
+
+        logger.info("Bluebird")
+        logger.info("https://github.com/EthanC/Bluebird")
+
+        if dotenv.load_dotenv():
+            logger.success("Loaded environment variables")
+
+        self.state: dict[str, int] = {}
+
+        # Reroute standard logging to Loguru
+        logging.basicConfig(handlers=[Intercept()], level=0, force=True)
+
+        if level := environ.get("LOG_LEVEL"):
+            logger.remove()
+            logger.add(stdout, level=level)
+
+            logger.success(f"Set console logging level to {level}")
+
+        if url := environ.get("LOG_DISCORD_WEBHOOK_URL"):
+            logger.add(
+                DiscordSink(url),
+                level=environ.get("LOG_DISCORD_WEBHOOK_LEVEL"),
+                backtrace=False,
+            )
+
+            logger.success("Enabled logging to Discord webhook")
+            logger.trace(url)
+
+        usersAll: list[str] = []
+        usersTop: list[str] = []
+        usersMedia: list[str] = []
+
+        if value := environ.get("USERS_ALL"):
+            usersAll = value.split(",")
+
+        if value := environ.get("USERS_TOP"):
+            usersTop = value.split(",")
+
+        if value := environ.get("USERS_MEDIA"):
+            usersMedia = value.split(",")
+
+        for username in usersAll:
+            watcher: Thread = Thread(target=Bluebird.WatchPosts, args=(self, username))
+
+            watcher.daemon = True
+            watcher.start()
+
+        for username in usersTop:
+            watcher: Thread = Thread(
+                target=Bluebird.WatchPosts,
+                args=(self, username),
+                kwargs={"replies": False},
+            )
+
+            watcher.daemon = True
+            watcher.start()
+
+        for username in usersMedia:
+            watcher: Thread = Thread(
+                target=Bluebird.WatchPosts,
+                args=(self, username),
+                kwargs={"media": True},
+            )
+
+            watcher.daemon = True
+            watcher.start()
+
+        # Keep the parent thread alive while the child threads run.
+        while True:
+            sleep(1)
+
+    def WatchPosts(
+        self: Self,
+        username: str,
+        replies: bool = True,
+        reposts: bool = True,
+        media: bool = False,
+    ) -> None:
+        """
+        Begin a loop for a single user that fires a notification upon
+        new post detection.
+        """
+
+        # Stagger the watcher threads to ease ratelimit concerns.
+        delay: int = randint(1, 30)
+        firstRun: bool = True
+
+        logger.info(f"[@{username}] Delayed watcher thread by {delay:,}s")
+
+        sleep(delay)
+
+        while True:
+            if not firstRun:
+                # Randomize cooldown to mimic natural behavior.
+                cooldownMax: int = int(environ.get("COOLDOWN_MAX_TIME", 60))
+                cooldown: int = randint(int(cooldownMax / 2), cooldownMax)
+
+                logger.info(
+                    f"[@{username}] Waiting {cooldown:,}s before checking for new posts"
+                )
+
+                sleep(cooldown)
+
+            firstRun = False
+
+            posts: list[dict[str, int | str]] = X.GetUserPosts(
+                username,
+                includeReplies=replies,
+                includeReposts=reposts,
+                onlyMedia=media,
+            )
+
+            # We didn't get any posts. Sleep then try again.
+            if len(posts) <= 0:
+                continue
+
+            # We don't have the latest post timestamp saved. This is
+            # likely a fresh run of the script. Set the latest post
+            # timestamp, then restart the loop.
+            if not self.state.get(username):
+                self.state[username] = posts[-1]["timestamp"]
+
+                logger.info(
+                    f"[@{username}] Watching for new posts after {posts[-1]["postId"]} ({posts[-1]["timestamp"]})"
+                )
+                logger.debug(f"https://x.com/{username}/status/{posts[-1]["postId"]}")
+
+                continue
+
+            for post in posts:
+                if self.state[username] >= post["timestamp"]:
+                    logger.debug(
+                        f"[@{username}] Skipped post {post["postId"]} due to timestamp ({self.state[username]} >= {post["timestamp"]})"
+                    )
+                    logger.debug(f"https://x.com/{username}/status/{post["postId"]}")
+
+                    continue
+
+                logger.success(
+                    f"[@{username}] Detected new post {post["postId"]} ({post["timestamp"]})"
+                )
+                logger.debug(f"https://x.com/{username}/status/{post["postId"]}")
+
+                details: dict = X.GetPost(username, post["postId"])
+
+                if details:
+                    Bluebird.NotifyPost(username, details)
+
+            self.state[username] = posts[-1]["timestamp"]
+
+    def NotifyPost(username: str, post: dict) -> None:
+        """Send a Discord Embed object for the specified X post."""
+
+        if not (webhook := environ.get("DISCORD_WEBHOOK_URL")):
+            return
+
+        embeds: list[DiscordEmbed] = []
+
+        primary: DiscordEmbed = DiscordEmbed()
+        extras: list[DiscordEmbed] = []
+
+        postUrl: str = (
+            f"https://x.com/{post["author"]["screen_name"]}/status/{post["id"]}"
+        )
+
+        primary.set_color("1D9BF0")
+        primary.set_author(
+            f"{post["author"]["name"]} (@{post["author"]["screen_name"]})",
+            url=f"https://x.com/{post["author"]["screen_name"]}",
+            icon_url=post["author"]["avatar_url"],
+        )
+        primary.set_title("Post on X")
+        primary.set_url(postUrl)
+        primary.set_footer(post["source"], icon_url="https://i.imgur.com/hZbC8my.png")
+        primary.set_timestamp(post["created_timestamp"])
+
+        if (post["text"]) and (len(post["text"]) > 0):
+            primary.set_description(f">>> {post["text"]}")
+
+        if media := post.get("media"):
+            idx: int = 0
+            assets: list[dict] = media.get("all", [])
+
+            # External media is not included in the all array.
+            if media.get("external"):
+                assets.append(media["external"])
+
+            for asset in assets:
+                extra: DiscordEmbed = DiscordEmbed()
+
+                extra.set_url(postUrl)
+
+                match asset["type"]:
+                    case "photo":
+                        if idx == 0:
+                            primary.set_image(asset["url"])
+                        else:
+                            extra.set_image(asset["url"])
+                    case "gif":
+                        if idx == 0:
+                            primary.set_image(asset["thumbnail_url"])
+                        else:
+                            extra.set_image(asset["thumbnail_url"])
+                    case "video":
+                        if idx == 0:
+                            primary.set_image(asset["thumbnail_url"])
+                        else:
+                            extra.set_image(asset["thumbnail_url"])
+                    case _:
+                        logger.warning(
+                            f"[@{username}] Unknown media asset type {asset["type"]} for post {post["id"]}"
+                        )
+                        logger.debug(postUrl)
+
+                if idx > 0:
+                    extras.append(extra)
+
+                idx += 1
+
+        embeds.append(primary)
+        embeds.extend(extras)
+
+        DiscordWebhook(url=webhook, embeds=embeds, rate_limit_retry=True).execute()
+
+
+if __name__ == "__main__":
+    try:
+        Bluebird.Initialize(Bluebird)
+    except KeyboardInterrupt:
+        exit()

handlers/__init__.py

@@ -0,0 +1,2 @@
+# ruff: noqa: F401
+from .intercept import Intercept