You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

90 lines
2.9 KiB

#!/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 =
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)
while True:
new = await get_post(0)
if new != latest:
await target.send(new.pings(), embed = new.to_embed())
latest = new
async def get_post(n: int) -> dict:
url = f'[campaign_id]={campaign_id}&filter[contains_exclusive_posts]=true&json-api-version=2.0'
response = requests.get(url)
return PatreonPost(response.json()['data'][n])