212 lines
6.5 KiB
Python
212 lines
6.5 KiB
Python
import discum
|
|
from typing import List
|
|
import webbrowser
|
|
import re
|
|
from bs4 import BeautifulSoup
|
|
import requests
|
|
from dotenv import load_dotenv
|
|
from os.path import realpath, dirname
|
|
from os import getenv
|
|
from yaml import load
|
|
from pathlib import Path
|
|
|
|
CONFIG_FILE = Path(f"{dirname(realpath(__file__))}/config.yaml")
|
|
URL_REGEX = re.compile(
|
|
"(?:(?:https?|ftp):\/\/|\b(?:[a-z\d]+\.))(?:(?:[^\s()<>]+|\((?:[^\s()<>]+|(?:\([^\s()<>]+\)))?\))+(?:\((?:[^\s()<>]+|(?:\(?:[^\s()<>]+\)))?\)|[^\s`!()\[\]{};:'.,<>?«»“”‘’]))?")
|
|
|
|
# consoles
|
|
PS5_CHANNELS = [713321461124694056, 721009876893040682,
|
|
781965674838753331, 802674800786145311,
|
|
780190141599580161, 780190104357961759]
|
|
|
|
# gpus
|
|
FOUNDERS_CHANNELS = [802674384120446996]
|
|
RTX3060_CHANNELS = [809731473669619730] + FOUNDERS_CHANNELS
|
|
RTX3060TI_CHANNELS = [783682409635250187, 783425011116539964,
|
|
802674527850725377, 796451904401834015] + FOUNDERS_CHANNELS
|
|
RTX3070_CHANNELS = [755728127069519913, 761002102804709448,
|
|
802674552541806662, 796451880224948274] + FOUNDERS_CHANNELS
|
|
RTX3070TI_CHANNELS = [839507735531749446,
|
|
849924994686910504] + FOUNDERS_CHANNELS
|
|
RTX3080_CHANNELS = [755727814912901170, 758224323843850250,
|
|
802674584473567303, 796451854808121414] + FOUNDERS_CHANNELS
|
|
RTX3080TI_CHANNELS = [833700313819119676,
|
|
849924965842419722] + FOUNDERS_CHANNELS
|
|
RTX3090_CHANNELS = [755728368413966387, 760577787332788315,
|
|
796451832968642561] + FOUNDERS_CHANNELS
|
|
|
|
MONITORED_CHANNELS = []
|
|
|
|
# load .env
|
|
load_dotenv()
|
|
|
|
########################################
|
|
# Globals
|
|
########################################
|
|
|
|
token = getenv("TOKEN")
|
|
bot = discum.Client(token=token, log=False)
|
|
|
|
########################################
|
|
# Functions
|
|
########################################
|
|
|
|
|
|
def get_soup(url: str):
|
|
r = requests.get(url)
|
|
return BeautifulSoup(r.text, features="lxml")
|
|
|
|
|
|
def get_stockinformer_url(url: str) -> str:
|
|
bs = get_soup(url)
|
|
|
|
for a in bs.find_all("a"):
|
|
if "view at" in a.text.lower():
|
|
url = a.get("href")
|
|
|
|
if "stockinformer" in url:
|
|
return f"https://stockinformer.co.uk/{url}"
|
|
elif "amazon" in url:
|
|
return f"{url.split('?')[0]}"
|
|
|
|
return url
|
|
|
|
return None
|
|
|
|
|
|
def get_partalert_url(url: str) -> str:
|
|
ret_url = None
|
|
bs = get_soup(url)
|
|
|
|
for a in bs.find_all("a"):
|
|
if "amazon" in a.text.lower():
|
|
# remove tags and referrals
|
|
amazon_url = a.get("href").split("?")[0]
|
|
|
|
ret_url = amazon_url
|
|
|
|
return ret_url
|
|
|
|
|
|
def check_urls(urls: List[str]):
|
|
for url in urls:
|
|
if "partalert" in url:
|
|
url = get_partalert_url(url)
|
|
|
|
if not url:
|
|
continue
|
|
elif "stockinformer" in url:
|
|
url = get_stockinformer_url(url)
|
|
|
|
if not url:
|
|
continue
|
|
|
|
print(f'Opening {url}')
|
|
webbrowser.open(url)
|
|
|
|
|
|
def identify_channel(channel_id: int):
|
|
# checking for founders first as this contains multiple models
|
|
if channel_id in FOUNDERS_CHANNELS:
|
|
print("Found a FE")
|
|
elif channel_id in PS5_CHANNELS:
|
|
print("Found a PS5")
|
|
elif channel_id in RTX3060_CHANNELS:
|
|
print("Found a RTX 3060")
|
|
elif channel_id in RTX3060TI_CHANNELS:
|
|
print("Found a RTX 3060 Ti")
|
|
elif channel_id in RTX3070_CHANNELS:
|
|
print("Found a RTX 3070")
|
|
elif channel_id in RTX3070TI_CHANNELS:
|
|
print("Found a RTX 3070 Ti")
|
|
elif channel_id in RTX3080_CHANNELS:
|
|
print("Found a RTX 3080")
|
|
elif channel_id in RTX3080TI_CHANNELS:
|
|
print("Found a RTX 3080 Ti")
|
|
elif channel_id in RTX3090_CHANNELS:
|
|
print("Found a RTX 3090")
|
|
|
|
|
|
@bot.gateway.command
|
|
def on_message(resp):
|
|
urls = []
|
|
|
|
if resp.event.ready_supplemental:
|
|
bot.gateway.subscribeToGuildEvents(wait=1)
|
|
|
|
if resp.event.message:
|
|
m = resp.parsed.auto()
|
|
|
|
channel_id = int(m['channel_id'])
|
|
content = m['content']
|
|
embeds = m['embeds']
|
|
|
|
if channel_id in MONITORED_CHANNELS:
|
|
# search for urls in message text
|
|
urls.extend(URL_REGEX.findall(content))
|
|
|
|
# search for urls in embeds
|
|
for e in embeds:
|
|
for f in [x['value'] for x in e['fields']]:
|
|
urls.extend(URL_REGEX.findall(f))
|
|
|
|
if (urls):
|
|
identify_channel(channel_id)
|
|
check_urls(urls)
|
|
|
|
|
|
def main():
|
|
global MONITORED_CHANNELS
|
|
|
|
if not token:
|
|
print("Could not load environmental variables. Make sure TOKEN is set in .env")
|
|
exit(0)
|
|
|
|
if not CONFIG_FILE.exists():
|
|
print(f"Could not find {CONFIG_FILE}.")
|
|
exit(0)
|
|
|
|
print(f"Loading configuration \"{CONFIG_FILE}\"")
|
|
|
|
with open(CONFIG_FILE, "r") as f:
|
|
config_data = load(f.read())
|
|
|
|
if "3060" in config_data and config_data["3060"] == True:
|
|
print("[*] Monitoring RTX 3060")
|
|
MONITORED_CHANNELS.extend(RTX3060_CHANNELS)
|
|
if "3060ti" in config_data and config_data["3060ti"] == True:
|
|
print("[*] Monitoring RTX 3060 Ti")
|
|
MONITORED_CHANNELS.extend(RTX3060TI_CHANNELS)
|
|
if "3070" in config_data and config_data["3070"] == True:
|
|
print("[*] Monitoring RTX 3070")
|
|
MONITORED_CHANNELS.extend(RTX3070_CHANNELS)
|
|
if "3070ti" in config_data and config_data["3070ti"] == True:
|
|
print("[*] Monitoring RTX 3070 Ti")
|
|
MONITORED_CHANNELS.extend(RTX3070TI_CHANNELS)
|
|
if "3080" in config_data and config_data["3080"] == True:
|
|
print("[*] Monitoring RTX 3080")
|
|
MONITORED_CHANNELS.extend(RTX3080_CHANNELS)
|
|
if "3080ti" in config_data and config_data["3080ti"] == True:
|
|
print("[*] Monitoring RTX 3080 Ti")
|
|
MONITORED_CHANNELS.extend(RTX3080TI_CHANNELS)
|
|
if "3090" in config_data and config_data["3090"] == True:
|
|
print("[*] Monitoring RTX 3090")
|
|
MONITORED_CHANNELS.extend(RTX3090_CHANNELS)
|
|
if "founders" in config_data and config_data["founders"] == True:
|
|
print("[*] Monitoring Founders Editions")
|
|
MONITORED_CHANNELS.extend(FOUNDERS_CHANNELS)
|
|
if "ps5" in config_data and config_data["ps5"] == True:
|
|
print("[*] Monitoring PS5")
|
|
MONITORED_CHANNELS.extend(PS5_CHANNELS)
|
|
|
|
# removing duplicate channels (if any) by making monitored channels a set
|
|
MONITORED_CHANNELS = set(MONITORED_CHANNELS)
|
|
|
|
# starting bot
|
|
print("Initialized.")
|
|
bot.gateway.run(auto_reconnect=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|