implement feed
parent
416017cde1
commit
9243798bc8
@ -0,0 +1 @@
|
||||
secret
|
@ -0,0 +1,90 @@
|
||||
#!/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
|
||||
debug = 356421819860779018
|
||||
role_tags = {
|
||||
'Dango': [dango],
|
||||
'Windbeutel': [dango, windbeutel, debug],
|
||||
'Vanilla Cupcake': [dango, windbeutel, cupcake],
|
||||
'Taiyaki': [dango, windbeutel, cupcake, taiyaki],
|
||||
'Erdnussbutterkeks': [dango, windbeutel, cupcake, taiyaki, cookie],
|
||||
}
|
||||
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'
|
||||
|
||||
|
||||
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:
|
||||
print('refreshing')
|
||||
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'])
|
Loading…
Reference in new issue