discord-patreon/main.py

#!/usr/bin/env python3
from datetime import datetime
from typing import List
import requests
import discord
import time
import os

# According to the docs, on_ready may be called multiple times.
# We use this bool to globally restrict our feed loop to run only once.
running = False

dango = 416184109790920705
windbeutel = 416184132473847810
cupcake = 416184150404628480
taiyaki = 416184208470310922
cookie = 416184227672096780
role_tags = {
    'Dango': [dango, windbeutel, cupcake, taiyaki, cookie],
    'Windbeutel': [windbeutel, cupcake, taiyaki, cookie],
    'Vanilla Cupcake': [cupcake, taiyaki, cookie],
    'Taiyaki': [taiyaki, cookie],
    'Erdnussbutterkeks': [cookie],
}
everyone_mention = '<@&239093443505291274>'
campaign_id = '187753'

class MyClient(discord.Client):
    async def on_ready(self):
        global running
        if not running:
            running = True
            target = await self.fetch_channel(os.environ['DISCORD_PATREON_CHANNEL'])
            print(f'Got channel {target}')
            await feed(target)

class PatreonPost():
    url: str
    title: str
    tags: List[str]

    def __init__(self, raw_data: dict):
        self.url = raw_data['attributes']['url']
        self.title = raw_data['attributes']['title']
        self.tags = [name_from_tag(tag) for tag in raw_data['relationships']['user_defined_tags']['data'] if tag['type'] == 'post_tag' and 'user_defined;' in tag['id'] and name_from_tag(tag) in role_tags.keys()]


    def __eq__(self, other) -> bool:
        return self.url == other.url


    def __ne__(self, other) -> bool:
        return self.url != other.url


    def __repr__(self) -> str:
        return f'{self.title} at {self.url} with tags {self.tags}'


    def pings(self) -> str:
        return ' '.join([f'<@&{role}>' for role in role_tags[self.tags[0]]]) if len(self.tags) > 0 else everyone_mention


    def to_embed(self) -> discord.Embed:
        embed = discord.Embed(color=0x30bdbe, timestamp = datetime.now())
        embed.set_thumbnail(url='https://images-ext-2.discordapp.net/external/Du9_FHQdYAfN4oygZb8caVpDAY4mmuVqZsFB2749kyE/https/cdn.discordapp.com/icons/239093443505291274/a_cb5ad90faf788f877b1b8b9691a55d28.png')
        embed.add_field(name=self.title, value=self.url)
        return embed

def name_from_tag(tag: dict) -> str:
    return tag['id'].split(';')[1]

async def feed(target):
    latest = await get_post(0)
    print(latest)
    while True:
        new = await get_post(0)
        if new != latest:
            print(new)
            await target.send(new.pings(), embed = new.to_embed())
            latest = new
        time.sleep(60)

async def get_post(n: int) -> dict:
    url = f'https://www.patreon.com/api/posts?sort=-published_at&filter[campaign_id]={campaign_id}&filter[contains_exclusive_posts]=true&json-api-version=2.0'
    response = requests.get(url)
    return PatreonPost(response.json()['data'][n])

MyClient().run(os.environ['DISCORD_PATREON_SECRET'])