Discord alerts for Raspberry Pi stock in a given region. Gets stock updates from rpilocator.com.

Dockerfile

from python:alpine
env PYTHONUNBUFFERED=1
run pip install --no-cache-dir feedparser requests
copy rpilocator-discord.py .
cmd ["python", "/rpilocator-discord.py"]

rpilocator-discord.py

from datetime import datetime
import feedparser
import requests
import time

# Replace the following URL with your Discord webhook URL
WEBHOOK_URL = "https://discord.com/api/webhooks/1234567890/abcdefgh"

# Region filter for rpilocator feed
REGION = 'UK'

# User IDs to mention. Nobody is mentioned if left empty.
# Example: MENTION_USERS = ['251776415735873537', '1028454186792259616']
MENTION_USERS = []

# Message to be sent when there are new entries in the RSS feed
DISCORD_STOCK_MESSAGE = "**New Pi stock**\n{entry.title} - Posted {entry.published}\nLink: {entry.link}"

# rpilocator.com RSS feed URL
FEED_URL = "https://rpilocator.com/feed"

# Set the start time as current time to use as the latest published date
START_TIME = time.time()

# Set the update interval to 1 minute
UPDATE_INTERVAL = 60

# Update the Discord webhook message with mentioned user IDs
if MENTION_USERS:
    for user_id in MENTION_USERS:
        DISCORD_STOCK_MESSAGE = f"<@{user_id}> {DISCORD_STOCK_MESSAGE}"


def sort_entries(dict):
    return dict['published_parsed']


def update_feed(url: str) -> list:
    return feedparser.parse(url)


def get_new_entries(feed: list) -> list:
    new_entries = []
    for entry in feed.entries:
        if any(tag.term == REGION for tag in entry.tags):
            published_date = time.mktime(entry.published_parsed)
            if published_date > START_TIME:
                new_entries.append(entry)
    return sorted(new_entries, key=sort_entries)


def send_message(url: str, message: str) -> int:
    response = requests.post(url, json={"content": message})
    return response.status_code

def get_timestamp() -> str:
    return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())


def main():
    global START_TIME

    # Update the RSS feed
    try:
        feed = update_feed(FEED_URL)
    except Exception as e:
        print(f"[{get_timestamp()}] Failed to update feed: {str(e)}")
        return

    # Build a list of new entries in the RSS feed
    new_entries = get_new_entries(feed)

    # If there are new entries, send messages to the Discord webhook
    if new_entries:
        print(f"[{get_timestamp()}] Found {len(new_entries)} new entries")
        send_success = True
        for entry in new_entries:
            message = DISCORD_STOCK_MESSAGE.format(entry=entry)
            response_status = send_message(WEBHOOK_URL, message)
            if response_status == 204:
                print(f"[{get_timestamp()}] Sent message for '{entry.title}'")
                send_success = True
            else:
                print(f"[{get_timestamp()}] Failed to send message for '{entry.title}'")
                send_success = False

        if send_success:
            START_TIME = time.mktime(new_entries[-1].published_parsed)

    # Print a status message if there are no new entries
    else:
        print(f"[{get_timestamp()}] Updated feed - No new entries")

if __name__ == "__main__":
    while True:
        main()

        # Wait for the next update interval
        time.sleep(UPDATE_INTERVAL)