259 lines
9.7 KiB
Python
259 lines
9.7 KiB
Python
import os
|
|
import json
|
|
import asyncio
|
|
from datetime import datetime, timedelta
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
import nextcord
|
|
from nextcord.ext import tasks, commands
|
|
|
|
|
|
def load_config(config_name):
|
|
if not os.path.exists(config_name):
|
|
return None
|
|
with open(config_name, "rt", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def save_config(config_name, config):
|
|
with open(config_name, 'wt', encoding="utf-8") as f:
|
|
json.dump(config, f, indent=1, ensure_ascii=False)
|
|
|
|
|
|
# dictionary of urls and time they were added
|
|
processed_urls = {}
|
|
|
|
config = load_config("config.json")
|
|
if config is None:
|
|
config = {"deepl_key": "", "discord_bot_token": "", "ping_terms": {}}
|
|
save_config("config.json", config)
|
|
|
|
if not config.get("deepl_key") or not config.get("discord_bot_token"):
|
|
raise Exception("Config file is missing some values")
|
|
|
|
intents = nextcord.Intents.default()
|
|
bot = commands.Bot(command_prefix=".", intents=intents)
|
|
headers = {
|
|
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/50.0.2661.102 Safari/537.36'}
|
|
deepl_headers = {'Authorization': f'DeepL-Auth-Key {config["deepl_key"]}', 'User-Agent': 'YahooScraper/1.0.0'}
|
|
search_term = 'けものフレンズ'
|
|
ping_terms: dict = config["ping_terms"]
|
|
|
|
|
|
def add_ping(term, user_id):
|
|
if term not in ping_terms:
|
|
ping_terms[term] = [user_id]
|
|
elif user_id not in ping_terms[term]:
|
|
ping_terms[term].append(user_id)
|
|
else:
|
|
print(f"{user_id} is already being pinged for {term}")
|
|
return False
|
|
|
|
config["ping_terms"] = ping_terms
|
|
save_config("config.json", config)
|
|
return True
|
|
|
|
|
|
def remove_ping(term, user_id):
|
|
if term not in ping_terms:
|
|
return False
|
|
elif user_id not in ping_terms[term]:
|
|
return False
|
|
else:
|
|
#if there is only one remaining person in the ping term list, delete the whole term
|
|
if len(ping_terms[term]) == 1 and ping_terms[term][0] == user_id:
|
|
del ping_terms[term]
|
|
#if there is more than on person in the ping term list, delete only that user_id
|
|
else:
|
|
ping_terms[term].remove(user_id)
|
|
|
|
config["ping_terms"] = ping_terms
|
|
save_config("config.json", config)
|
|
return True
|
|
|
|
|
|
def get_user_pings(user_id):
|
|
return [term for term in ping_terms.keys() if user_id in ping_terms[term]]
|
|
|
|
|
|
async def ping_users(auction_name_jp, auction_name_en):
|
|
"""If the name of the listing is particularly interesting, ping some people
|
|
- HAV0X"""
|
|
users_to_ping = []
|
|
interested_terms = []
|
|
for term in ping_terms.keys():
|
|
if term.lower() in auction_name_jp.lower() or term.lower() in auction_name_en.lower():
|
|
interested_terms.append(term)
|
|
for user in ping_terms[term]:
|
|
if user not in users_to_ping:
|
|
users_to_ping.append(user)
|
|
|
|
if len(users_to_ping) > 0:
|
|
ping_message = "Interesting listing, pinging: " + ", ".join([f"<@{user}>" for user in users_to_ping]) + "\nTerms of interest: " + ', '.join([f'"{s}"' for s in interested_terms])
|
|
await send_message_text("Japari Modding", "yahoo-auctions", ping_message)
|
|
|
|
|
|
def translate(text: str):
|
|
text = requests.post("https://api-free.deepl.com/v2/translate",
|
|
json={"text": [text], "target_lang": "EN", 'source_lang': 'JA'}, headers=deepl_headers).json()[
|
|
'translations'][0]['text']
|
|
return text.replace("Beast Friends", "Kemono Friends")
|
|
|
|
|
|
def embed_auction(service: str, url: str, name: str, name_en: str, thumbnail: str, price=None):
|
|
desc = name
|
|
if price is not None:
|
|
desc += '\nPrice: ' + price
|
|
embed = nextcord.Embed(title=name_en, url=url, description=desc)
|
|
embed.set_author(name=service)
|
|
embed.set_image(url=thumbnail)
|
|
return embed
|
|
|
|
|
|
async def send_message(server_name: str, channel_name: str, embed: nextcord.Embed):
|
|
for guild in bot.guilds:
|
|
if guild.name == server_name:
|
|
for channel in guild.channels:
|
|
if channel.name == channel_name and isinstance(channel, nextcord.TextChannel):
|
|
await channel.send(embed=embed)
|
|
break
|
|
|
|
|
|
async def send_message_text(server_name: str, channel_name: str, text: str):
|
|
for guild in bot.guilds:
|
|
if guild.name == server_name:
|
|
for channel in guild.channels:
|
|
if channel.name == channel_name and isinstance(channel, nextcord.TextChannel):
|
|
await channel.send(content=text)
|
|
break
|
|
|
|
|
|
async def check_yahoo_fleamarket(search_term: str, page: int, notify: bool):
|
|
url = f'https://buyee.jp/paypayfleamarket/search?keyword={search_term}&order-sort=created_time&page={page}'
|
|
|
|
response = requests.get(url, headers=headers)
|
|
soup = BeautifulSoup(response.text, 'html.parser')
|
|
|
|
# Find all item listings on the page
|
|
item_listings = soup.find('ul', {'class': 'item-lists'}).find_all('li', {'class': 'list'})
|
|
date = datetime.now().date()
|
|
|
|
for item in item_listings:
|
|
url = 'https://buyee.jp' + item.find('a').get('href')
|
|
|
|
if url in processed_urls:
|
|
continue
|
|
|
|
try:
|
|
name = item.find('h2', {'class': 'name'}).text
|
|
price = item.find('p', {'class': 'price'}).text
|
|
thumbnail_data = item.find('img', {'class': "thumbnail"}).get("data-bind")
|
|
thumbnail_start = thumbnail_data.find('imagePath: \'') + len('imagePath: \'')
|
|
thumbnail_end = thumbnail_data.find('\'', thumbnail_start)
|
|
thumbnail = "https:" + thumbnail_data[thumbnail_start:thumbnail_end]
|
|
|
|
if notify:
|
|
name_en = translate(name)
|
|
embed = embed_auction("Yahoo! Flea market", url, name, name_en, thumbnail, price)
|
|
await send_message("Japari Modding", "yahoo-auctions", embed)
|
|
await ping_users(name, name_en)
|
|
|
|
print('New item added:', url)
|
|
processed_urls[url] = date
|
|
except Exception as e:
|
|
print(url, e)
|
|
|
|
|
|
async def check_yahoo_auction(search_term: str, page: int, notify: bool):
|
|
url = f'https://buyee.jp/item/search/query/{search_term}?sort=end&order=d&page={page}'
|
|
|
|
response = requests.get(url, headers=headers)
|
|
soup = BeautifulSoup(response.text, 'html.parser')
|
|
|
|
# Find all item listings on the page
|
|
item_listings = soup.find('ul', {'class': 'auctionSearchResult'}).find_all('li', {'class': 'itemCard'})
|
|
date = datetime.now().date()
|
|
|
|
for item in item_listings:
|
|
url = 'https://buyee.jp' + item.find('a').get('href')
|
|
|
|
if url in processed_urls:
|
|
continue
|
|
|
|
try:
|
|
name = item.find('div', {'class': 'itemCard__itemName'}).find('a').text
|
|
price = item.find('div', {'class': 'g-price__outer'}).find('span').text
|
|
thumbnail = item.find('img', {'class': 'g-thumbnail__image'}).get('data-src').split(';')[0]
|
|
|
|
if notify:
|
|
name_en = translate(name)
|
|
embed = embed_auction("Yahoo! JAPAN Auction", url, name, name_en, thumbnail, price)
|
|
await send_message("Japari Modding", "yahoo-auctions", embed)
|
|
await ping_users(name, name_en)
|
|
|
|
print('New item added:', url)
|
|
processed_urls[url] = date
|
|
except Exception as e:
|
|
print(url, e)
|
|
|
|
|
|
@bot.event
|
|
async def on_ready():
|
|
for i in range(1, 4):
|
|
await check_yahoo_auction(search_term, i, False)
|
|
await check_yahoo_fleamarket(search_term, i, False)
|
|
await asyncio.sleep(10)
|
|
myLoop.start()
|
|
|
|
|
|
@tasks.loop(minutes=30)
|
|
async def myLoop():
|
|
fulldate = datetime.now()
|
|
|
|
print(f'Last check: {fulldate.strftime("%d/%m/%Y %H:%M:%S")}\nCache count: {len(processed_urls)}')
|
|
|
|
try:
|
|
await check_yahoo_auction(search_term, 1, True)
|
|
except Exception as e:
|
|
print("Yahoo auction check failed:", e)
|
|
|
|
try:
|
|
await check_yahoo_fleamarket(search_term, 1, True)
|
|
except Exception as e:
|
|
print("Yahoo flea market check failed:", e)
|
|
|
|
onlydate = fulldate.date()
|
|
for key in list(processed_urls.keys()):
|
|
if onlydate - processed_urls[key] > timedelta(weeks=3):
|
|
del processed_urls[key]
|
|
|
|
|
|
@bot.slash_command(description="Pings you when auction name contains the given term")
|
|
async def ping_me(interaction: nextcord.Interaction, term: str):
|
|
if add_ping(term, interaction.user.id):
|
|
await interaction.response.send_message(f"Success! You will be pinged for `{term}`", ephemeral=False)
|
|
else:
|
|
pings = get_user_pings(interaction.user.id)
|
|
if len(pings) > 0:
|
|
await interaction.response.send_message(f"Failed to add `{term}`. You are pinged for `{', '.join(pings)}`",
|
|
ephemeral=True)
|
|
else:
|
|
await interaction.response.send_message(f"Failed to add `{term}`. You are not pinged for anything",
|
|
ephemeral=True)
|
|
|
|
|
|
@bot.slash_command(description="No longer pings you when auction name contains the given term")
|
|
async def unping_me(interaction: nextcord.Interaction, term: str):
|
|
if remove_ping(term, interaction.user.id):
|
|
await interaction.response.send_message(f"Success! You will no longer be pinged for `{term}`", ephemeral=False)
|
|
else:
|
|
pings = get_user_pings(interaction.user.id)
|
|
if len(pings) > 0:
|
|
await interaction.response.send_message(f"Failed to remove `{term}`. You are pinged for `{', '.join(pings)}`",
|
|
ephemeral=True)
|
|
else:
|
|
await interaction.response.send_message(f"Failed to remove `{term}`. You are not pinged for anything",
|
|
ephemeral=True)
|
|
|
|
|
|
bot.run(config["discord_bot_token"], reconnect=True) |