remoll/tbot-with-extras
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

t

Browse Source
This commit is contained in:
Remy Moll
2022-01-01 14:51:13 +01:00
parent bf4b02902b
commit f468332843
52 changed files with 3272 additions and 3419 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
bot/api/__init__.py
View File

@@ -1,6 +1,6 @@
from . import keys
from . import reddit
from . import weather
from . import reddit
from . import search
from . import keys
from . import reddit
from . import weather
from . import reddit
from . import search
from . import metmuseum

78
bot/api/metmuseum.py
View File

@@ -1,39 +1,39 @@
import requests
import random
from PIL import Image
import io
class ArtFetch:
def __init__(self):
self.base_url = "https://collectionapi.metmuseum.org/"
self.objects = self.fetch_objects() # chosen set of images to select randomly
def fetch_objects(self):
"""We restrict ourselves to a few domains."""
# fetch all departements
t = requests.get(self.base_url + "public/collection/v1/departments").json()
deps = t["departments"]
keep_id = []
for d in deps:
name = d["displayName"]
if name == "American Decorative Arts" or name == "Arts of Africa, Oceania, and the Americas" or name == "Asian Art" or name == "European Paintings":
keep_id.append(str(d["departmentId"]))
# fetch artworks listed under these departments
data = {"departmentIds" : "|".join(keep_id)}
t = requests.get(self.base_url + "public/collection/v1/objects",params=data).json()
# num = t["total"]
ids = t["objectIDs"]
return ids
def get_random_art(self):
"""Returns an image object of a randomly selected artwork"""
# fetch the artwork's url
r_id = self.objects[random.randint(0,len(self.objects))]
t = requests.get(self.base_url + "public/collection/v1/objects/" + str(r_id)).json()
im_url = t["primaryImageSmall"]
# download the image
resp = requests.get(im_url)
img = Image.open(io.BytesIO(resp.content))
return img
import requests
import random
from PIL import Image
import io
class ArtFetch:
def __init__(self):
self.base_url = "https://collectionapi.metmuseum.org/"
self.objects = self.fetch_objects() # chosen set of images to select randomly
def fetch_objects(self):
"""We restrict ourselves to a few domains."""
# fetch all departements
t = requests.get(self.base_url + "public/collection/v1/departments").json()
deps = t["departments"]
keep_id = []
for d in deps:
name = d["displayName"]
if name == "American Decorative Arts" or name == "Arts of Africa, Oceania, and the Americas" or name == "Asian Art" or name == "European Paintings":
keep_id.append(str(d["departmentId"]))
# fetch artworks listed under these departments
data = {"departmentIds" : "|".join(keep_id)}
t = requests.get(self.base_url + "public/collection/v1/objects",params=data).json()
# num = t["total"]
ids = t["objectIDs"]
return ids
def get_random_art(self):
"""Returns an image object of a randomly selected artwork"""
# fetch the artwork's url
r_id = self.objects[random.randint(0,len(self.objects))]
t = requests.get(self.base_url + "public/collection/v1/objects/" + str(r_id)).json()
im_url = t["primaryImageSmall"]
# download the image
resp = requests.get(im_url)
img = Image.open(io.BytesIO(resp.content))
return img

112
bot/api/reddit.py
View File

@@ -1,56 +1,56 @@
import praw
class RedditFetch():
def __init__(self, key):
self.stream = praw.Reddit(client_id = key["id"], client_secret = key["secret"], user_agent=key["user_agent"])
def get_top(self, subreddit, number, return_type="text"):
if return_type == "text":
posts = []
try:
for submission in self.stream.subreddit(subreddit).top(limit=number):
p = {}
if not submission.stickied:
p["title"] = submission.title
p["content"] = submission.selftext
posts.append(p)
return posts
except:
return []
else:
images = []
try:
for submission in self.stream.subreddit(subreddit).top(limit=number):
if not submission.stickied:
t = {"image": submission.url, "caption": submission.title}
images.append(t)
return images
except:
return []
def get_random_rising(self, subreddit, number, return_type="text"):
if return_type == "text":
posts = []
try:
for submission in self.stream.subreddit(subreddit).random_rising(limit=number):
p = {}
if not submission.stickied:
p["title"] = submission.title
p["content"] = submission.selftext
posts.append(p)
return posts
except:
return []
else:
images = []
try:
for submission in self.stream.subreddit(subreddit).random_rising(limit=number):
if not submission.stickied:
t = {"image": submission.url, "caption": submission.title}
images.append(t)
return images
except:
return []
import praw
class RedditFetch():
def __init__(self, key):
self.stream = praw.Reddit(client_id = key["id"], client_secret = key["secret"], user_agent=key["user_agent"])
def get_top(self, subreddit, number, return_type="text"):
if return_type == "text":
posts = []
try:
for submission in self.stream.subreddit(subreddit).top(limit=number):
p = {}
if not submission.stickied:
p["title"] = submission.title
p["content"] = submission.selftext
posts.append(p)
return posts
except:
return []
else:
images = []
try:
for submission in self.stream.subreddit(subreddit).top(limit=number):
if not submission.stickied:
t = {"image": submission.url, "caption": submission.title}
images.append(t)
return images
except:
return []
def get_random_rising(self, subreddit, number, return_type="text"):
if return_type == "text":
posts = []
try:
for submission in self.stream.subreddit(subreddit).random_rising(limit=number):
p = {}
if not submission.stickied:
p["title"] = submission.title
p["content"] = submission.selftext
posts.append(p)
return posts
except:
return []
else:
images = []
try:
for submission in self.stream.subreddit(subreddit).random_rising(limit=number):
if not submission.stickied:
t = {"image": submission.url, "caption": submission.title}
images.append(t)
return images
except:
return []

40
bot/api/search.py
View File

@@ -1,21 +1,21 @@
import duckduckpy
class WebSearch():
def __init__(self):
self.search = duckduckpy.query
def get_result(self, query):
try:
res = []
response = self.search(query, container = "dict")["related_topics"]
for r in response:
if "text" in r:
res.append({
"text" : r["text"],
"url": r["first_url"]
})
except:
res = ["Connection error"]
return res
import duckduckpy
class WebSearch():
def __init__(self):
self.search = duckduckpy.query
def get_result(self, query):
try:
res = []
response = self.search(query, container = "dict")["related_topics"]
for r in response:
if "text" in r:
res.append({
"text" : r["text"],
"url": r["first_url"]
})
except:
res = ["Connection error"]
return res
# TODO: this api has more potential. Extract images or quick facts!

174
bot/api/weather.py
View File

@@ -1,87 +1,87 @@
import requests
import datetime
import logging
logger = logging.getLogger(__name__)
class WeatherFetch():
def __init__(self, key):
self.last_fetch = datetime.datetime.fromtimestamp(0)
self.last_weather = ""
self.calls = 0
self.url = "https://api.openweathermap.org/data/2.5/onecall?"
self.key = key
def show_weather(self, location):
delta = datetime.datetime.now() - self.last_fetch
if delta.total_seconds()/60 > 60 or "\n" not in self.last_weather: # 1 hour passed:
data = {"lat" : location[0], "lon" : location[1], "exclude" : "minutely,hourly", "appid" : self.key, "units" : "metric"}
self.calls += 1
logger.info("Just fetched weather. ({}th time)".format(self.calls))
# today = datetime.datetime.today().weekday()
# days = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
try:
weather = requests.get(self.url,params=data).json()
# categories = {"Clouds": ":cloud:", "Rain": ":cloud_with_rain:", "Thunderstorm": "thunder_cloud_rain", "Drizzle": ":droplet:", "Snow": ":cloud_snow:", "Clear": ":sun:", "Mist": "Mist", "Smoke": "Smoke", "Haze": "Haze", "Dust": "Dust", "Fog": "Fog", "Sand": "Sand", "Dust": "Dust", "Ash": "Ash", "Squall": "Squall", "Tornado": "Tornado",}
now = weather["current"]
ret_weather = []
ret_weather.append({
"short" : now["weather"][0]["main"],
"temps" : [int(now["temp"])]
})
weather_days = weather["daily"]
for i, day in enumerate(weather_days):
ret_weather.append({
"short" : day["weather"][0]["main"],
"temps" : [int(day["temp"]["min"]),int(day["temp"]["max"])]
})
except:
ret_weather = []
# now = weather["current"]
# message = "<b>Now:</b> " + categories[now["weather"][0]["main"]] + "\n"
# message += ":thermometer: " + str(int(now["temp"])) + "°\n\n"
# weather_days = weather["daily"]
# for i, day in enumerate(weather_days):
# if i == 0:
# message += "<b>" + "Today" + ":</b> " + categories[day["weather"][0]["main"]] + "\n"
# else:
# message += "<b>" + days[(today + i + 1) % 7] + ":</b> " + categories[day["weather"][0]["main"]] + "\n"
# message += ":thermometer: :fast_down_button: " + str(int(day["temp"]["min"])) + "° , :thermometer: :fast_up_button: " + str(int(day["temp"]["max"])) + "°\n\n"
# except:
# message = "Query failed, it's my fault, I'm sorry :sad:"
self.last_weather = ret_weather
self.last_fetch = datetime.datetime.now()
else:
ret_weather = self.last_weather
return ret_weather
# def get_weather_by_city(self, city):
# loc = get_coords_from_city(self, city)
# weather = self.show_weather(loc)
# return weather
# def get_coords_from_city(self, city):
# url = "https://devru-latitude-longitude-find-v1.p.rapidapi.com/latlon.php"
# data = {"location": city}
# headers = {
# "x-rapidapi-key" : "d4e0ab7ab3mshd5dde5a282649e0p11fd98jsnc93afd98e3aa",
# "x-rapidapi-host" : "devru-latitude-longitude-find-v1.p.rapidapi.com",
# }
# #try:
# resp = requests.request("GET", url, headers=headers, params=data)
# result = resp.text
# #except:
# # result = "???"
# return result
import requests
import datetime
import logging
logger = logging.getLogger(__name__)
class WeatherFetch():
def __init__(self, key):
self.last_fetch = datetime.datetime.fromtimestamp(0)
self.last_weather = ""
self.calls = 0
self.url = "https://api.openweathermap.org/data/2.5/onecall?"
self.key = key
def show_weather(self, location):
delta = datetime.datetime.now() - self.last_fetch
if delta.total_seconds()/60 > 60 or "\n" not in self.last_weather: # 1 hour passed:
data = {"lat" : location[0], "lon" : location[1], "exclude" : "minutely,hourly", "appid" : self.key, "units" : "metric"}
self.calls += 1
logger.info("Just fetched weather. ({}th time)".format(self.calls))
# today = datetime.datetime.today().weekday()
# days = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
try:
weather = requests.get(self.url,params=data).json()
# categories = {"Clouds": ":cloud:", "Rain": ":cloud_with_rain:", "Thunderstorm": "thunder_cloud_rain", "Drizzle": ":droplet:", "Snow": ":cloud_snow:", "Clear": ":sun:", "Mist": "Mist", "Smoke": "Smoke", "Haze": "Haze", "Dust": "Dust", "Fog": "Fog", "Sand": "Sand", "Dust": "Dust", "Ash": "Ash", "Squall": "Squall", "Tornado": "Tornado",}
now = weather["current"]
ret_weather = []
ret_weather.append({
"short" : now["weather"][0]["main"],
"temps" : [int(now["temp"])]
})
weather_days = weather["daily"]
for i, day in enumerate(weather_days):
ret_weather.append({
"short" : day["weather"][0]["main"],
"temps" : [int(day["temp"]["min"]),int(day["temp"]["max"])]
})
except:
ret_weather = []
# now = weather["current"]
# message = "<b>Now:</b> " + categories[now["weather"][0]["main"]] + "\n"
# message += ":thermometer: " + str(int(now["temp"])) + "°\n\n"
# weather_days = weather["daily"]
# for i, day in enumerate(weather_days):
# if i == 0:
# message += "<b>" + "Today" + ":</b> " + categories[day["weather"][0]["main"]] + "\n"
# else:
# message += "<b>" + days[(today + i + 1) % 7] + ":</b> " + categories[day["weather"][0]["main"]] + "\n"
# message += ":thermometer: :fast_down_button: " + str(int(day["temp"]["min"])) + "° , :thermometer: :fast_up_button: " + str(int(day["temp"]["max"])) + "°\n\n"
# except:
# message = "Query failed, it's my fault, I'm sorry :sad:"
self.last_weather = ret_weather
self.last_fetch = datetime.datetime.now()
else:
ret_weather = self.last_weather
return ret_weather
# def get_weather_by_city(self, city):
# loc = get_coords_from_city(self, city)
# weather = self.show_weather(loc)
# return weather
# def get_coords_from_city(self, city):
# url = "https://devru-latitude-longitude-find-v1.p.rapidapi.com/latlon.php"
# data = {"location": city}
# headers = {
# "x-rapidapi-key" : "d4e0ab7ab3mshd5dde5a282649e0p11fd98jsnc93afd98e3aa",
# "x-rapidapi-host" : "devru-latitude-longitude-find-v1.p.rapidapi.com",
# }
# #try:
# resp = requests.request("GET", url, headers=headers, params=data)
# result = resp.text
# #except:
# # result = "???"
# return result