Firepup650/bitbot-3.11-fork
2
0
Fork 0
Code Issues Pull requests Activity Actions

'utils.http.get_url' -> 'utils.http.request', return a Response object from

Browse source 
utils.http.request
This commit is contained in:
jesopo 2018-12-11 22:26:38 +00:00
parent a41cf7b471
commit 793d234a0b
29 changed files with 169 additions and 165 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
modules/bitcoin.py
View file

@ -10,16 +10,16 @@ class Module(ModuleManager.BaseModule):
:usage: [currency] :usage: [currency]
""" """
currency = (event["args"] or "USD").upper() currency = (event["args"] or "USD").upper()
page = utils.http.get_url("https://blockchain.info/ticker", page = utils.http.request("https://blockchain.info/ticker",
json=True) json=True)
if page: if page:
if currency in page: if currency in page.data:
conversion = page[currency] conversion = page.data[currency]
buy, sell = conversion["buy"], conversion["sell"] buy, sell = conversion["buy"], conversion["sell"]
event["stdout"].write("1 BTC = %.2f %s (buy) %.2f %s " event["stdout"].write("1 BTC = %.2f %s (buy) %.2f %s "
"(sell)" % (buy, currency, sell, currency)) "(sell)" % (buy, currency, sell, currency))
else: else:
event["stderr"].write("Unknown currency, available " event["stderr"].write("Unknown currency, available "
"currencies: %s" % ", ".join(page.keys())) "currencies: %s" % ", ".join(page.data.keys()))
else: else:
raise utils.EventsResultsError() raise utils.EventsResultsError()

6
modules/books.py
View file

@ -9,11 +9,11 @@ class Module(ModuleManager.BaseModule):
_name = "ISBN" _name = "ISBN"
def get_book(self, query, event): def get_book(self, query, event):
page = utils.http.get_url(URL_GOOGLEBOOKS, get_params={ page = utils.http.request(URL_GOOGLEBOOKS, get_params={
"q": query, "country": "us"}, json=True) "q": query, "country": "us"}, json=True)
if page: if page:
if page["totalItems"] > 0: if page.data["totalItems"] > 0:
book = page["items"][0]["volumeInfo"] book = page.data["items"][0]["volumeInfo"]
title = book["title"] title = book["title"]
sub_title = (", %s" % book.get("subtitle") sub_title = (", %s" % book.get("subtitle")
) if book.get("subtitle") else "" ) if book.get("subtitle") else ""

4
modules/check_urls.py
View file

@ -22,11 +22,11 @@ class Module(ModuleManager.BaseModule):
if match and event["channel"].get_setting("check-urls", if match and event["channel"].get_setting("check-urls",
event["server"].get_setting("check-urls", False)): event["server"].get_setting("check-urls", False)):
url = match.group(0) url = match.group(0)
page = utils.http.get_url(URL_VIRUSTOTAL, get_params={ page = utils.http.request(URL_VIRUSTOTAL, get_params={
"apikey": self.bot.config["virustotal-api-key"], "apikey": self.bot.config["virustotal-api-key"],
"resource": url}, json=True) "resource": url}, json=True)
if page and page.get("positives", 0) > 1: if page and page.data.get("positives", 0) > 1:
if event["channel"].get_setting("check-urls-kick", False): if event["channel"].get_setting("check-urls-kick", False):
event["channel"].send_kick(event["user"].nickname, event["channel"].send_kick(event["user"].nickname,
"Don't send malicious URLs!") "Don't send malicious URLs!")

18
modules/define.py
View file

@ -12,7 +12,7 @@ class Module(ModuleManager.BaseModule):
_last_called = 0 _last_called = 0
def _get_definition(self, word): def _get_definition(self, word):
page = utils.http.get_url(URL_WORDNIK % word, get_params={ page = utils.http.request(URL_WORDNIK % word, get_params={
"useCanonical": "true", "limit": 1, "useCanonical": "true", "limit": 1,
"sourceDictionaries": "wiktionary", "api_key": self.bot.config[ "sourceDictionaries": "wiktionary", "api_key": self.bot.config[
"wordnik-api-key"]}, json=True) "wordnik-api-key"]}, json=True)
@ -33,9 +33,9 @@ class Module(ModuleManager.BaseModule):
page = self._get_definition(word) page = self._get_definition(word)
if page: if page:
if len(page): if len(page.data):
event["stdout"].write("%s: %s" % (page[0]["word"], event["stdout"].write("%s: %s" % (page.data[0]["word"],
page[0]["text"])) page.data[0]["text"]))
else: else:
event["stderr"].write("No definitions found") event["stderr"].write("No definitions found")
else: else:
@ -50,13 +50,13 @@ class Module(ModuleManager.BaseModule):
RANDOM_DELAY_SECONDS): RANDOM_DELAY_SECONDS):
self._last_called = time.time() self._last_called = time.time()
page = utils.http.get_url(URL_WORDNIK_RANDOM, get_params={ page = utils.http.request(URL_WORDNIK_RANDOM, get_params={
"api_key":self.bot.config["wordnik-api-key"], "api_key":self.bot.config["wordnik-api-key"],
"min_dictionary_count":1},json=True) "min_dictionary_count":1},json=True)
if page and len(page): if page and len(page.data):
definition = self._get_definition(page["word"]) definition = self._get_definition(page.data["word"])
if len(definition): if definition and len(definition.data):
definition = definition[0] definition = definition.data[0]
else: else:
raise utils.EventError("Try again in a couple of seconds") raise utils.EventError("Try again in a couple of seconds")

4
modules/eval_lua.py
View file

@ -8,7 +8,7 @@ class Module(ModuleManager.BaseModule):
@utils.hook("received.command.lua", min_args=1) @utils.hook("received.command.lua", min_args=1)
def eval(self, event): def eval(self, event):
try: try:
page = utils.http.get_url(EVAL_URL, page = utils.http.request(EVAL_URL,
post_data={"input": event["args"]}, post_data={"input": event["args"]},
method="POST", method="POST",
soup=True) soup=True)
@ -17,7 +17,7 @@ class Module(ModuleManager.BaseModule):
event["user"].nickname) event["user"].nickname)
if page: if page:
textareas = page.find_all("textarea") textareas = page.data.find_all("textarea")
if len(textareas) > 1: if len(textareas) > 1:
out = textareas[1].text.strip("\n") out = textareas[1].text.strip("\n")
event["stdout"].write("%s: %s" % (event["user"].nickname, out)) event["stdout"].write("%s: %s" % (event["user"].nickname, out))

6
modules/eval_python.py
View file

@ -35,7 +35,7 @@ class Module(ModuleManager.BaseModule):
def _eval(self, lang, event): def _eval(self, lang, event):
page = None page = None
try: try:
page = utils.http.get_url(EVAL_URL, page = utils.http.request(EVAL_URL,
post_data={ post_data={
"lang": lang, "lang": lang,
"code": EVAL_TEMPLATE, "code": EVAL_TEMPLATE,
@ -47,8 +47,8 @@ class Module(ModuleManager.BaseModule):
except: except:
pass pass
if page: if page.data:
out = page.split("</b></span><br>", 1)[1] out = page.data.split("</b></span><br>", 1)[1]
out = html.unescape(out) out = html.unescape(out)
out = json.loads(out) out = json.loads(out)

6
modules/eval_rust.py
View file

@ -30,12 +30,12 @@ class Module(ModuleManager.BaseModule):
args = API_ARGS.copy() args = API_ARGS.copy()
args["code"] = FN_TEMPLATE % event["args"] args["code"] = FN_TEMPLATE % event["args"]
try: try:
page = utils.http.get_url(EVAL_URL, json_data=args, page = utils.http.request(EVAL_URL, json_data=args,
method="POST", json=True) method="POST", json=True)
except socket.timeout: except socket.timeout:
raise utils.EventError("%s: eval timed out" % raise utils.EventError("%s: eval timed out" %
event["user"].nickname) event["user"].nickname)
err_or_out = "stdout" if page["success"] else "stderr" err_or_out = "stdout" if page.data["success"] else "stderr"
event[err_or_out].write("%s: %s" % (event["user"].nickname, event[err_or_out].write("%s: %s" % (event["user"].nickname,
page[err_or_out].strip("\n"))) page.data[err_or_out].strip("\n")))

10
modules/google.py
View file

@ -17,15 +17,15 @@ class Module(ModuleManager.BaseModule):
""" """
phrase = event["args"] or event["target"].buffer.get() phrase = event["args"] or event["target"].buffer.get()
if phrase: if phrase:
page = utils.http.get_url(URL_GOOGLESEARCH, get_params={ page = utils.http.request(URL_GOOGLESEARCH, get_params={
"q": phrase, "key": self.bot.config[ "q": phrase, "key": self.bot.config[
"google-api-key"], "cx": self.bot.config[ "google-api-key"], "cx": self.bot.config[
"google-search-id"], "prettyPrint": "true", "google-search-id"], "prettyPrint": "true",
"num": 1, "gl": "gb"}, json=True) "num": 1, "gl": "gb"}, json=True)
if page: if page:
if "items" in page and len(page["items"]): if "items" in page.data and len(page.data["items"]):
event["stdout"].write( event["stdout"].write(
"(%s) %s" % (phrase, page["items"][0]["link"])) "(%s) %s" % (phrase, page.data["items"][0]["link"]))
else: else:
event["stderr"].write("No results found") event["stderr"].write("No results found")
else: else:
@ -41,11 +41,11 @@ class Module(ModuleManager.BaseModule):
""" """
phrase = event["args"] or event["target"].buffer.get() phrase = event["args"] or event["target"].buffer.get()
if phrase: if phrase:
page = utils.http.get_url(URL_GOOGLESUGGEST, get_params={ page = utils.http.request(URL_GOOGLESUGGEST, get_params={
"output": "json", "client": "hp", "gl": "gb", "q": phrase}) "output": "json", "client": "hp", "gl": "gb", "q": phrase})
if page: if page:
# google gives us jsonp, so we need to unwrap it. # google gives us jsonp, so we need to unwrap it.
page = page.split("(", 1)[1][:-1] page = page.data.split("(", 1)[1][:-1]
page = json.loads(page) page = json.loads(page)
suggestions = page[1] suggestions = page[1]
suggestions = [utils.http.strip_html(s[0]) for s in suggestions] suggestions = [utils.http.strip_html(s[0]) for s in suggestions]

5
modules/haveibeenpwned.py
View file

@ -11,11 +11,10 @@ class Module(ModuleManager.BaseModule):
hacked databases hacked databases
:usage: <username/email> :usage: <username/email>
""" """
page = utils.http.get_url(URL_HAVEIBEENPWNEDAPI % event["args"], page = utils.http.request(URL_HAVEIBEENPWNEDAPI % event["args"],
json=True, code=True) json=True, code=True)
if page: if page:
code, page = page if page.code == 200:
if code == 200:
event["stdout"].write( event["stdout"].write(
"It seems '%s' has been pwned. check on %s." % (event["args"], "It seems '%s' has been pwned. check on %s." % (event["args"],
URL_HAVEIBEENPWNED)) URL_HAVEIBEENPWNED))

10
modules/imdb.py
View file

@ -15,16 +15,16 @@ class Module(ModuleManager.BaseModule):
:help: Search for a given title on IMDb :help: Search for a given title on IMDb
:usage: <movie/tv title> :usage: <movie/tv title>
""" """
page = utils.http.get_url(URL_OMDB, get_params={ page = utils.http.request(URL_OMDB, get_params={
"t": event["args"], "t": event["args"],
"apikey": self.bot.config["omdbapi-api-key"]}, "apikey": self.bot.config["omdbapi-api-key"]},
json=True) json=True)
if page: if page:
if "Title" in page: if "Title" in page.data:
event["stdout"].write("%s, %s (%s) %s (%s/10.0) %s" % ( event["stdout"].write("%s, %s (%s) %s (%s/10.0) %s" % (
page["Title"], page["Year"], page["Runtime"], page.data["Title"], page.data["Year"], page.data["Runtime"],
page["Plot"], page["imdbRating"], page.data["Plot"], page.data["imdbRating"],
URL_IMDBTITLE % page["imdbID"])) URL_IMDBTITLE % page.data["imdbID"]))
else: else:
event["stderr"].write("Title not found") event["stderr"].write("Title not found")
else: else:

12
modules/imgur.py
View file

@ -20,12 +20,12 @@ class Module(ModuleManager.BaseModule):
def _image_info(self, hash): def _image_info(self, hash):
api_key = self.bot.config["imgur-api-key"] api_key = self.bot.config["imgur-api-key"]
result = utils.http.get_url(URL_IMAGE % hash, result = utils.http.request(URL_IMAGE % hash,
headers={"Authorization": "Client-ID %s" % api_key}, headers={"Authorization": "Client-ID %s" % api_key},
json=True) json=True)
if result and result["success"]: if result and result.data["success"]:
data = result["data"] data = result.data["data"]
text = self._prefix(data) text = self._prefix(data)
text += "(%s %dx%d, %d views)" % (data["type"], data["width"], text += "(%s %dx%d, %d views)" % (data["type"], data["width"],
@ -38,12 +38,12 @@ class Module(ModuleManager.BaseModule):
def _gallery_info(self, hash): def _gallery_info(self, hash):
api_key = self.bot.config["imgur-api-key"] api_key = self.bot.config["imgur-api-key"]
result = utils.http.get_url(URL_GALLERY % hash, result = utils.http.request(URL_GALLERY % hash,
headers={"Authorization": "Client-ID %s" % api_key}, headers={"Authorization": "Client-ID %s" % api_key},
json=True) json=True)
if result and result["success"]: if result and result.data["success"]:
data = result["data"] data = result.data["data"]
text = self._prefix(data) text = self._prefix(data)
text += "(%d views, %d▲▼%d)" % (data["views"], text += "(%d views, %d▲▼%d)" % (data["views"],
data["ups"], data["downs"]) data["ups"], data["downs"])

22
modules/ip_addresses.py
View file

@ -34,19 +34,19 @@ class Module(ModuleManager.BaseModule):
:usage: <IP> :usage: <IP>
:prefix: GeoIP :prefix: GeoIP
""" """
page = utils.http.get_url(URL_GEOIP % event["args_split"][0], page = utils.http.request(URL_GEOIP % event["args_split"][0],
json=True) json=True)
if page: if page:
if page["status"] == "success": if page.data["status"] == "success":
data = page["query"] data = page.data["query"]
data += " | Organisation: %s" % page["org"] data += " | Organisation: %s" % page.data["org"]
data += " | City: %s" % page["city"] data += " | City: %s" % page.data["city"]
data += " | Region: %s (%s)" % (page["regionName"], data += " | Region: %s (%s)" % (page.data["regionName"],
page["countryCode"]) page.data["countryCode"])
data += " | ISP: %s" % page["isp"] data += " | ISP: %s" % page.data["isp"]
data += " | Lon/Lat: %s/%s" % (page["lon"], data += " | Lon/Lat: %s/%s" % (page.data["lon"],
page["lat"]) page.data["lat"])
data += " | Timezone: %s" % page["timezone"] data += " | Timezone: %s" % page.data["timezone"]
event["stdout"].write(data) event["stdout"].write(data)
else: else:
event["stderr"].write("No geoip data found") event["stderr"].write("No geoip data found")

6
modules/isgd.py
View file

@ -11,11 +11,11 @@ class Module(ModuleManager.BaseModule):
if not re.match(REGEX_URL, url): if not re.match(REGEX_URL, url):
url = "http://%s" % url url = "http://%s" % url
data = utils.http.get_url(ISGD_API_URL, get_params= page = utils.http.request(ISGD_API_URL, get_params=
{"format": "json", "url": url}, json=True) {"format": "json", "url": url}, json=True)
if data and data["shorturl"]: if page and page.data["shorturl"]:
return data["shorturl"] return page.data["shorturl"]
@utils.hook("received.command.shorten", min_args=1) @utils.hook("received.command.shorten", min_args=1)
def shorten(self, event): def shorten(self, event):

16
modules/lastfm.py
View file

@ -24,14 +24,14 @@ class Module(ModuleManager.BaseModule):
lastfm_username = event["user"].get_setting("lastfm", lastfm_username = event["user"].get_setting("lastfm",
event["user"].nickname) event["user"].nickname)
shown_username = event["user"].nickname shown_username = event["user"].nickname
page = utils.http.get_url(URL_SCROBBLER, get_params={ page = utils.http.request(URL_SCROBBLER, get_params={
"method": "user.getrecenttracks", "user": lastfm_username, "method": "user.getrecenttracks", "user": lastfm_username,
"api_key": self.bot.config["lastfm-api-key"], "api_key": self.bot.config["lastfm-api-key"],
"format": "json", "limit": "1"}, json=True) "format": "json", "limit": "1"}, json=True)
if page: if page:
if "recenttracks" in page and len(page["recenttracks" if "recenttracks" in page.data and len(page.data["recenttracks"
]["track"]): ]["track"]):
now_playing = page["recenttracks"]["track"][0] now_playing = page.data["recenttracks"]["track"][0]
track_name = now_playing["name"] track_name = now_playing["name"]
artist = now_playing["artist"]["#text"] artist = now_playing["artist"]["#text"]
@ -53,14 +53,14 @@ class Module(ModuleManager.BaseModule):
short_url = " -- " + short_url if short_url else "" short_url = " -- " + short_url if short_url else ""
info_page = utils.http.get_url(URL_SCROBBLER, get_params={ info_page = utils.http.request(URL_SCROBBLER, get_params={
"method": "track.getInfo", "artist": artist, "method": "track.getInfo", "artist": artist,
"track": track_name, "autocorrect": "1", "track": track_name, "autocorrect": "1",
"api_key": self.bot.config["lastfm-api-key"], "api_key": self.bot.config["lastfm-api-key"],
"user": lastfm_username, "format": "json"}, json=True) "user": lastfm_username, "format": "json"}, json=True)
tags = [] tags = []
if "toptags" in info_page.get("track", []): if "toptags" in info_page.data.get("track", []):
for tag in info_page["track"]["toptags"]["tag"]: for tag in info_page.data["track"]["toptags"]["tag"]:
tags.append(tag["name"]) tags.append(tag["name"])
if tags: if tags:
tags = " (%s)" % ", ".join(tags) tags = " (%s)" % ", ".join(tags)
@ -68,8 +68,8 @@ class Module(ModuleManager.BaseModule):
tags = "" tags = ""
play_count = "" play_count = ""
if "userplaycount" in info_page.get("track", []): if "userplaycount" in info_page.data.get("track", []):
play_count = int(info_page["track"]["userplaycount"]) play_count = int(info_page.data["track"]["userplaycount"])
play_count = " (%d play%s)" % (play_count, play_count = " (%d play%s)" % (play_count,
"s" if play_count > 1 else "") "s" if play_count > 1 else "")

8
modules/nr.py
View file

@ -235,10 +235,10 @@ class Module(ModuleManager.BaseModule):
trains.append(parsed) trains.append(parsed)
if eagle_url: if eagle_url:
summary_query = utils.http.get_url("%s/json/summaries/%s?uids=%s" % (eagle_url, now.date().isoformat(), "%20".join([a["uid"] for a in trains])), json=True, headers={"x-eagle-key": self.bot.config["eagle-api-key"]}) summary_query = utils.http.request("%s/json/summaries/%s?uids=%s" % (eagle_url, now.date().isoformat(), "%20".join([a["uid"] for a in trains])), json=True, headers={"x-eagle-key": self.bot.config["eagle-api-key"]})
if summary_query: if summary_query:
for t in trains: for t in trains:
summary = summary_query[t["uid"]] summary = summary_query.data[t["uid"]]
t.update(summary) t.update(summary)
summary_plat = summary.get("platforms", {}).get(query["crs"]) summary_plat = summary.get("platforms", {}).get(query["crs"])
if summary_plat and t["platform"]=="?": if summary_plat and t["platform"]=="?":
@ -333,9 +333,9 @@ class Module(ModuleManager.BaseModule):
query = client.service.QueryServices(service_id, datetime.utcnow().date().isoformat(), query = client.service.QueryServices(service_id, datetime.utcnow().date().isoformat(),
datetime.utcnow().time().strftime("%H:%M:%S+0000")) datetime.utcnow().time().strftime("%H:%M:%S+0000"))
if eagle_url: if eagle_url:
schedule_query = utils.http.get_url("%s/json/schedule/%s/%s" % (eagle_url, service_id, datetime.now().date().isoformat()), json=True, headers={"x-eagle-key": eagle_key}) schedule_query = utils.http.request("%s/json/schedule/%s/%s" % (eagle_url, service_id, datetime.now().date().isoformat()), json=True, headers={"x-eagle-key": eagle_key})
if schedule_query: if schedule_query:
schedule = schedule_query["current"] schedule = schedule_query.data["current"]
if not query and not schedule: if not query and not schedule:
return event["stdout"].write("No service information is available for this identifier.") return event["stdout"].write("No service information is available for this identifier.")

4
modules/soundcloud.py
View file

@ -43,12 +43,12 @@ class Module(ModuleManager.BaseModule):
else: else:
get_params["url"] = url get_params["url"] = url
page = utils.http.get_url( page = utils.http.request(
URL_SOUNDCLOUD_TRACK if has_query else URL_SOUNDCLOUD_RESOLVE, URL_SOUNDCLOUD_TRACK if has_query else URL_SOUNDCLOUD_RESOLVE,
get_params=get_params, json=True) get_params=get_params, json=True)
if page: if page:
page = page[0] if has_query else page page = page.data[0] if has_query else page
title = page["title"] title = page["title"]
user = page["user"]["username"] user = page["user"]["username"]
duration = time.strftime("%H:%M:%S", time.gmtime(page[ duration = time.strftime("%H:%M:%S", time.gmtime(page[

6
modules/spotify.py
View file

@ -10,12 +10,12 @@ class Module(ModuleManager.BaseModule):
:help: Search for a track on spotify :help: Search for a track on spotify
:usage: <term> :usage: <term>
""" """
page = utils.http.get_url(URL_SPOTIFY, get_params= page = utils.http.request(URL_SPOTIFY, get_params=
{"type": "track", "limit": 1, "q": event["args"]}, {"type": "track", "limit": 1, "q": event["args"]},
json=True) json=True)
if page: if page:
if len(page["tracks"]["items"]): if len(page.data["tracks"]["items"]):
item = page["tracks"]["items"][0] item = page.data["tracks"]["items"][0]
title = item["name"] title = item["name"]
artist_name = item["artists"][0]["name"] artist_name = item["artists"][0]["name"]
url = item["external_urls"]["spotify"] url = item["external_urls"]["spotify"]

46
modules/tfl.py
View file

@ -63,26 +63,26 @@ class Module(ModuleManager.BaseModule):
real_stop_id = "" real_stop_id = ""
stop_name = "" stop_name = ""
if stop_id.isdigit(): if stop_id.isdigit():
bus_search = utils.http.get_url(URL_BUS_SEARCH % stop_id, bus_search = utils.http.request(URL_BUS_SEARCH % stop_id,
get_params={"app_id": app_id, "app_key": app_key}, get_params={"app_id": app_id, "app_key": app_key},
json=True) json=True)
bus_stop = bus_search["matches"][0] bus_stop = bus_search.data["matches"][0]
real_stop_id = bus_stop["id"] real_stop_id = bus_stop.data["id"]
stop_name = bus_stop["name"] stop_name = bus_stop.data["name"]
else: else:
bus_stop = utils.http.get_url(URL_STOP % stop_id, bus_stop = utils.http.request(URL_STOP % stop_id,
get_params={"app_id": app_id, "app_key": app_key}, get_params={"app_id": app_id, "app_key": app_key},
json=True) json=True)
if bus_stop: if bus_stop:
real_stop_id = stop_id real_stop_id = stop_id
stop_name = bus_stop["commonName"] stop_name = bus_stop.data["commonName"]
if real_stop_id: if real_stop_id:
bus_stop = utils.http.get_url(URL_BUS % real_stop_id, bus_stop = utils.http.request(URL_BUS % real_stop_id,
get_params={"app_id": app_id, "app_key": app_key}, get_params={"app_id": app_id, "app_key": app_key},
json=True) json=True)
busses = [] busses = []
for bus in bus_stop: for bus in bus_stop.data:
bus_number = bus["lineName"] bus_number = bus["lineName"]
human_time = self.vehicle_span(bus["expectedArrival"]) human_time = self.vehicle_span(bus["expectedArrival"])
time_until = self.vehicle_span(bus["expectedArrival"], human=False) time_until = self.vehicle_span(bus["expectedArrival"], human=False)
@ -135,10 +135,10 @@ class Module(ModuleManager.BaseModule):
app_id = self.bot.config["tfl-api-id"] app_id = self.bot.config["tfl-api-id"]
app_key = self.bot.config["tfl-api-key"] app_key = self.bot.config["tfl-api-key"]
lines = utils.http.get_url(URL_LINE, get_params={ lines = utils.http.request(URL_LINE, get_params={
"app_id": app_id, "app_key": app_key}, json=True) "app_id": app_id, "app_key": app_key}, json=True)
statuses = [] statuses = []
for line in lines: for line in lines.data:
for status in line["lineStatuses"]: for status in line["lineStatuses"]:
entry = { entry = {
"id": line["id"], "id": line["id"],
@ -182,13 +182,13 @@ class Module(ModuleManager.BaseModule):
#As awful as this is, it also makes it ~work~. #As awful as this is, it also makes it ~work~.
stop_name = event["args"].replace(" ", "%20") stop_name = event["args"].replace(" ", "%20")
stop_search = utils.http.get_url(URL_STOP_SEARCH % stop_name, get_params={ stop_search = utils.http.request(URL_STOP_SEARCH % stop_name, get_params={
"app_id": app_id, "app_key": app_key, "maxResults": "6", "faresOnly": "False"}, json=True) "app_id": app_id, "app_key": app_key, "maxResults": "6", "faresOnly": "False"}, json=True)
if stop_search: if stop_search:
for stop in stop_search["matches"]: for stop in stop_search.data["matches"]:
pass pass
results = ["%s (%s): %s" % (stop["name"], ", ".join(stop["modes"]), stop["id"]) for stop in stop_search["matches"]] results = ["%s (%s): %s" % (stop["name"], ", ".join(stop["modes"]), stop["id"]) for stop in stop_search.data["matches"]]
event["stdout"].write("[%s results] %s" % (stop_search["total"], "; ".join(results))) event["stdout"].write("[%s results] %s" % (stop_search.data["total"], "; ".join(results)))
else: else:
event["stderr"].write("No results") event["stderr"].write("No results")
@ -203,15 +203,15 @@ class Module(ModuleManager.BaseModule):
vehicle_id = event["args_split"][0] vehicle_id = event["args_split"][0]
vehicle = utils.http.get_url(URL_VEHICLE % vehicle_id, get_params={ vehicle = utils.http.request(URL_VEHICLE % vehicle_id, get_params={
"app_id": app_id, "app_key": app_key}, json=True)[0] "app_id": app_id, "app_key": app_key}, json=True)[0]
arrival_time = self.vehicle_span(vehicle["expectedArrival"], human=False) arrival_time = self.vehicle_span(vehicle.data["expectedArrival"], human=False)
platform = self.platform(vehicle["platformName"]) platform = self.platform(vehicle.data["platformName"])
event["stdout"].write("%s (%s) to %s. %s. Arrival at %s (%s) in %s minutes on %s" % ( event["stdout"].write("%s (%s) to %s. %s. Arrival at %s (%s) in %s minutes on %s" % (
vehicle["vehicleId"], vehicle["lineName"], vehicle["destinationName"], vehicle["currentLocation"], vehicle.data["vehicleId"], vehicle.data["lineName"], vehicle.data["destinationName"], vehicle.data["currentLocation"],
vehicle["stationName"], vehicle["naptanId"], arrival_time, platform)) vehicle.data["stationName"], vehicle.data["naptanId"], arrival_time, platform))
@utils.hook("received.command.tflservice", min_args=1) @utils.hook("received.command.tflservice", min_args=1)
def service(self, event): def service(self, event):
@ -233,10 +233,10 @@ class Module(ModuleManager.BaseModule):
event["stdout"].write("%s is too high. Remember that the first arrival is 0" % service_id) event["stdout"].write("%s is too high. Remember that the first arrival is 0" % service_id)
return return
service = results[int(service_id)] service = results[int(service_id)]
arrivals = utils.http.get_url(URL_LINE_ARRIVALS % service["route"], arrivals = utils.http.request(URL_LINE_ARRIVALS % service["route"],
get_params={"app_id": app_id, "app_key": app_key}, json=True) get_params={"app_id": app_id, "app_key": app_key}, json=True)
arrivals = [a for a in arrivals if a["vehicleId"] == service["id"]] arrivals = [a for a in arrivals.data if a["vehicleId"] == service["id"]]
arrivals = sorted(arrivals, key=lambda b: b["timeToStation"]) arrivals = sorted(arrivals, key=lambda b: b["timeToStation"])
event["stdout"].write( event["stdout"].write(
@ -257,7 +257,7 @@ class Module(ModuleManager.BaseModule):
stop_id = event["args_split"][0] stop_id = event["args_split"][0]
stop = utils.http.get_url(URL_STOP % stop_id, get_params={ stop = utils.http.requesst(URL_STOP % stop_id, get_params={
"app_id": app_id, "app_key": app_key}, json=True) "app_id": app_id, "app_key": app_key}, json=True)
def route(self, event): def route(self, event):
@ -266,7 +266,7 @@ class Module(ModuleManager.BaseModule):
route_id = event["args_split"][0] route_id = event["args_split"][0]
route = utils.http.get_url(URL_ROUTE % route_id, get_params={ route = utils.http.request(URL_ROUTE % route_id, get_params={
"app_id": app_id, "app_key": app_key}, json=True) "app_id": app_id, "app_key": app_key}, json=True)
event["stdout"].write("") event["stdout"].write("")

12
modules/thesaurus.py
View file

@ -12,14 +12,14 @@ class Module(ModuleManager.BaseModule):
:usage: <word> [type] :usage: <word> [type]
""" """
phrase = event["args_split"][0] phrase = event["args_split"][0]
page = utils.http.get_url(URL_THESAURUS % (self.bot.config[ page = utils.http.request(URL_THESAURUS % (self.bot.config[
"bighugethesaurus-api-key"], phrase), json=True) "bighugethesaurus-api-key"], phrase), json=True)
syn_ant = event["command"][:3] syn_ant = event["command"][:3]
if page: if page:
if not len(event["args_split"]) > 1: if not len(event["args_split"]) > 1:
word_types = [] word_types = []
for word_type in page.keys(): for word_type in page.data.keys():
if syn_ant in page[word_type]: if syn_ant in page.data[word_type]:
word_types.append(word_type) word_types.append(word_type)
if word_types: if word_types:
word_types = sorted(word_types) word_types = sorted(word_types)
@ -30,11 +30,11 @@ class Module(ModuleManager.BaseModule):
event["stderr"].write("No categories available") event["stderr"].write("No categories available")
else: else:
category = event["args_split"][1].lower() category = event["args_split"][1].lower()
if category in page: if category in page.data:
if syn_ant in page[category]: if syn_ant in page.data[category]:
event["stdout"].write("%ss for %s: %s" % ( event["stdout"].write("%ss for %s: %s" % (
event["command"].title(), phrase, ", ".join( event["command"].title(), phrase, ", ".join(
page[category][syn_ant]))) page.data[category][syn_ant])))
else: else:
event["stderr"].write("No %ss for %s" % ( event["stderr"].write("No %ss for %s" % (
event["command"], phrase)) event["command"], phrase))

7
modules/title.py
View file

@ -21,16 +21,15 @@ class Module(ModuleManager.BaseModule):
if not url: if not url:
raise utils.EventError("No URL provided/found.") raise utils.EventError("No URL provided/found.")
soup = None
try: try:
soup = utils.http.get_url(url, soup=True) page = utils.http.request(url, soup=True)
except: except:
pass pass
if not soup: if not page:
raise utils.EventError("Failed to get URL.") raise utils.EventError("Failed to get URL.")
title = soup.title title = page.data.title
if title: if title:
title = title.text.replace("\n", " ").replace("\r", "" title = title.text.replace("\n", " ").replace("\r", ""
).replace(" ", " ").strip() ).replace(" ", " ").strip()

27
modules/trakt.py
View file

@ -19,30 +19,29 @@ class Module(ModuleManager.BaseModule):
else: else:
username = event["user"].get_setting("trakt", username = event["user"].get_setting("trakt",
event["user"].nickname) event["user"].nickname)
page = utils.http.get_url(URL_TRAKT % username, headers={ page = utils.http.request(URL_TRAKT % username, headers={
"Content-Type": "application/json", "Content-Type": "application/json",
"trakt-api-version": "2", "trakt-api-key": "trakt-api-version": "2", "trakt-api-key":
self.bot.config["trakt-api-key"]}, json=True, self.bot.config["trakt-api-key"]}, json=True,
code=True) code=True)
if page[0]: if page
code, page = page if page.code == 200:
if code == 200: type = page.data["type"]
type = page["type"]
if type == "movie": if type == "movie":
title = page["movie"]["title"] title = page.data["movie"]["title"]
year = page["movie"]["year"] year = page.data["movie"]["year"]
slug = page["movie"]["ids"]["slug"] slug = page.data["movie"]["ids"]["slug"]
event["stdout"].write( event["stdout"].write(
"%s is now watching %s (%s) %s" % ( "%s is now watching %s (%s) %s" % (
username, title, year, username, title, year,
URL_TRAKTSLUG % ("movie", slug))) URL_TRAKTSLUG % ("movie", slug)))
elif type == "episode": elif type == "episode":
season = page["episode"]["season"] season = page.data["episode"]["season"]
episode_number = page["episode"]["number"] episode_number = page.data["episode"]["number"]
episode_title = page["episode"]["title"] episode_title = page.data["episode"]["title"]
show_title = page["show"]["title"] show_title = page.data["show"]["title"]
show_year = page["show"]["year"] show_year = page.data["show"]["year"]
slug = page["show"]["ids"]["slug"] slug = page.data["show"]["ids"]["slug"]
event["stdout"].write( event["stdout"].write(
"%s is now watching %s s%se%s - %s %s" % ( "%s is now watching %s s%se%s - %s %s" % (
username, show_title, str(season).zfill(2), username, show_title, str(season).zfill(2),

5
modules/translate.py
View file

@ -32,11 +32,12 @@ class Module(ModuleManager.BaseModule):
target_language = language_match.group(2) target_language = language_match.group(2)
phrase = phrase.split(" ", 1)[1] phrase = phrase.split(" ", 1)[1]
data = utils.http.get_url(URL_TRANSLATE, get_params={ page = utils.http.request(URL_TRANSLATE, get_params={
"client": "gtx", "sl": source_language, "client": "gtx", "sl": source_language,
"tl": target_language, "dt": "t", "q": phrase}) "tl": target_language, "dt": "t", "q": phrase})
if data and not data == "[null,null,\"\"]": if page and not page.data == "[null,null,\"\"]":
data = page.data
while ",," in data: while ",," in data:
data = data.replace(",,", ",null,") data = data.replace(",,", ",null,")
data = data.replace("[,", "[null,") data = data.replace("[,", "[null,")

6
modules/upc.py
View file

@ -15,13 +15,13 @@ class Module(ModuleManager.BaseModule):
if not arg_len == 12 and not arg_len == 13: if not arg_len == 12 and not arg_len == 13:
raise utils.EventError("Invalid UPC/EAN/GTIN provided") raise utils.EventError("Invalid UPC/EAN/GTIN provided")
page = utils.http.get_url(UPCITEMDB_URL, page = utils.http.request(UPCITEMDB_URL,
get_params={"upc": event["args_split"][0]}, get_params={"upc": event["args_split"][0]},
json=True) json=True)
if page: if page:
if not len(page["items"]): if not len(page.data["items"]):
raise utils.EventError("UPC/EAN not found") raise utils.EventError("UPC/EAN not found")
item = page["items"][0] item = page.data["items"][0]
brand = item.get("brand", None) brand = item.get("brand", None)
brand = "" if not brand else "%s - " % brand brand = "" if not brand else "%s - " % brand

8
modules/urbandictionary.py
View file

@ -18,12 +18,12 @@ class Module(ModuleManager.BaseModule):
if match: if match:
number = int(match.group(1)) number = int(match.group(1))
term = term.split(" ", 1)[1] term = term.split(" ", 1)[1]
page = utils.http.get_url(URL_URBANDICTIONARY, page = utils.http.request(URL_URBANDICTIONARY,
get_params={"term": term}, json=True) get_params={"term": term}, json=True)
if page: if page:
if len(page["list"]): if len(page.data["list"]):
if number > 0 and len(page["list"]) > number-1: if number > 0 and len(page.data["list"]) > number-1:
definition = page["list"][number-1] definition = page.data["list"][number-1]
event["stdout"].write("%s: %s" % (definition["word"], event["stdout"].write("%s: %s" % (definition["word"],
definition["definition"].replace("\n", " ").replace( definition["definition"].replace("\n", " ").replace(
"\r", "").replace(" ", " "))) "\r", "").replace(" ", " ")))

16
modules/weather.py
View file

@ -12,19 +12,19 @@ class Module(ModuleManager.BaseModule):
:usage: <location> :usage: <location>
""" """
api_key = self.bot.config["openweathermap-api-key"] api_key = self.bot.config["openweathermap-api-key"]
page = utils.http.get_url(URL_WEATHER, get_params={ page = utils.http.request(URL_WEATHER, get_params={
"q": event["args"], "units": "metric", "q": event["args"], "units": "metric",
"APPID": api_key}, "APPID": api_key},
json=True) json=True)
if page: if page:
if "weather" in page: if "weather" in page.data:
location = "%s, %s" % (page["name"], page["sys"][ location = "%s, %s" % (page.data["name"], page.data["sys"][
"country"]) "country"])
celsius = "%dC" % page["main"]["temp"] celsius = "%dC" % page.data["main"]["temp"]
fahrenheit = "%dF" % ((page["main"]["temp"]*(9/5))+32) fahrenheit = "%dF" % ((page.data["main"]["temp"]*(9/5))+32)
description = page["weather"][0]["description"].title() description = page.data["weather"][0]["description"].title()
humidity = "%s%%" % page["main"]["humidity"] humidity = "%s%%" % page.data["main"]["humidity"]
wind_speed = "%sKM/h" % page["wind"]["speed"] wind_speed = "%sKM/h" % page.data["wind"]["speed"]
event["stdout"].write( event["stdout"].write(
"(%s) %s/%s | %s | Humidity: %s | Wind: %s" % ( "(%s) %s/%s | %s | Humidity: %s | Wind: %s" % (

4
modules/wikipedia.py
View file

@ -10,13 +10,13 @@ class Module(ModuleManager.BaseModule):
:help: Get information from wikipedia :help: Get information from wikipedia
:usage: <term> :usage: <term>
""" """
page = utils.http.get_url(URL_WIKIPEDIA, get_params={ page = utils.http.request(URL_WIKIPEDIA, get_params={
"action": "query", "prop": "extracts", "action": "query", "prop": "extracts",
"titles": event["args"], "exintro": "", "titles": event["args"], "exintro": "",
"explaintext": "", "exchars": "500", "explaintext": "", "exchars": "500",
"redirects": "", "format": "json"}, json=True) "redirects": "", "format": "json"}, json=True)
if page: if page:
pages = page["query"]["pages"] pages = page.data["query"]["pages"]
article = list(pages.items())[0][1] article = list(pages.items())[0][1]
if not "missing" in article: if not "missing" in article:
title, info = article["title"], article["extract"] title, info = article["title"], article["extract"]

8
modules/wolframalpha.py
View file

@ -14,14 +14,14 @@ class Module(ModuleManager.BaseModule):
:help: Evauate a given string on Wolfram|Alpha :help: Evauate a given string on Wolfram|Alpha
:usage: <query> :usage: <query>
""" """
code, result = utils.http.get_url(URL_WA, page = utils.http.requestl(URL_WA,
get_params={"i": event["args"], get_params={"i": event["args"],
"appid": self.bot.config["wolframalpha-api-key"], "appid": self.bot.config["wolframalpha-api-key"],
"reinterpret": "true", "units": "metric"}, code=True) "reinterpret": "true", "units": "metric"}, code=True)
if not result == None: if page:
if code == 200: if page.code == 200:
event["stdout"].write("%s: %s" % (event["args"], result)) event["stdout"].write("%s: %s" % (event["args"], page.data))
else: else:
event["stdout"].write("No results") event["stdout"].write("No results")
else: else:

18
modules/youtube.py
View file

@ -21,13 +21,13 @@ ARROW_DOWN = "↓"
"validate": utils.bool_or_none}) "validate": utils.bool_or_none})
class Module(ModuleManager.BaseModule): class Module(ModuleManager.BaseModule):
def get_video_page(self, video_id, part): def get_video_page(self, video_id, part):
return utils.http.get_url(URL_YOUTUBEVIDEO, get_params={"part": part, return utils.http.request(URL_YOUTUBEVIDEO, get_params={"part": part,
"id": video_id, "key": self.bot.config["google-api-key"]}, "id": video_id, "key": self.bot.config["google-api-key"]},
json=True) json=True)
def video_details(self, video_id): def video_details(self, video_id):
snippet = self.get_video_page(video_id, "snippet") snippet = self.get_video_page(video_id, "snippet")
if snippet["items"]: if snippet.data["items"]:
snippet = snippet["items"][0]["snippet"] snippet = snippet.data["items"][0]["snippet"]
statistics = self.get_video_page(video_id, "statistics")[ statistics = self.get_video_page(video_id, "statistics")[
"items"][0]["statistics"] "items"][0]["statistics"]
content = self.get_video_page(video_id, "contentDetails")[ content = self.get_video_page(video_id, "contentDetails")[
@ -60,15 +60,15 @@ class Module(ModuleManager.BaseModule):
search = event["query"] search = event["query"]
video_id = "" video_id = ""
search_page = utils.http.get_url(URL_YOUTUBESEARCH, search_page = utils.http.request(URL_YOUTUBESEARCH,
get_params={"q": search, "part": "snippet", get_params={"q": search, "part": "snippet",
"maxResults": "1", "type": "video", "maxResults": "1", "type": "video",
"key": self.bot.config["google-api-key"]}, "key": self.bot.config["google-api-key"]},
json=True) json=True)
if search_page: if search_page:
if search_page["pageInfo"]["totalResults"] > 0: if search_page.data["pageInfo"]["totalResults"] > 0:
video_id = search_page["items"][0]["id"]["videoId"] video_id = search_page.data["items"][0]["id"]["videoId"]
return "https://youtu.be/%s" % video_id return "https://youtu.be/%s" % video_id
@utils.hook("received.command.yt", alias_of="youtube") @utils.hook("received.command.yt", alias_of="youtube")
@ -93,14 +93,14 @@ class Module(ModuleManager.BaseModule):
if search or video_id: if search or video_id:
if not video_id: if not video_id:
search_page = utils.http.get_url(URL_YOUTUBESEARCH, search_page = utils.http.request(URL_YOUTUBESEARCH,
get_params={"q": search, "part": "snippet", get_params={"q": search, "part": "snippet",
"maxResults": "1", "type": "video", "maxResults": "1", "type": "video",
"key": self.bot.config["google-api-key"]}, "key": self.bot.config["google-api-key"]},
json=True) json=True)
if search_page: if search_page:
if search_page["pageInfo"]["totalResults"] > 0: if search_page.data["pageInfo"]["totalResults"] > 0:
video_id = search_page["items"][0]["id"]["videoId"] video_id = search_page.data["items"][0]["id"]["videoId"]
else: else:
event["stderr"].write("No videos found") event["stderr"].write("No videos found")
else: else:

26
src/utils/http.py
View file

@ -18,10 +18,18 @@ class HTTPParsingException(HTTPException):
def throw_timeout(): def throw_timeout():
raise HTTPTimeoutException() raise HTTPTimeoutException()
def get_url(url: str, method: str="GET", get_params: dict={}, class Response(object):
def __init__(self, code: int, data: typing.Any,
headers: typing.Dict[str, str]):
self.code = code
self.data = data
self.headers = headers
def request(url: str, method: str="GET", get_params: dict={},
post_data: typing.Any=None, headers: dict={}, post_data: typing.Any=None, headers: dict={},
json_data: typing.Any=None, code: bool=False, json: bool=False, json_data: typing.Any=None, code: bool=False, json: bool=False,
soup: bool=False, parser: str="lxml", fallback_encoding: str="utf8"): soup: bool=False, parser: str="lxml", fallback_encoding: str="utf8",
) -> Response:
if not urllib.parse.urlparse(url).scheme: if not urllib.parse.urlparse(url).scheme:
url = "http://%s" % url url = "http://%s" % url
@ -49,23 +57,21 @@ def get_url(url: str, method: str="GET", get_params: dict={},
finally: finally:
signal.signal(signal.SIGALRM, signal.SIG_IGN) signal.signal(signal.SIGALRM, signal.SIG_IGN)
response_headers = utils.CaseInsensitiveDict(response.headers)
if soup: if soup:
soup = bs4.BeautifulSoup(response_content, parser) soup = bs4.BeautifulSoup(response_content, parser)
if code: return Response(response.status_code, soup, response_heders)
return response.status_code, soup
return soup
data = response_content.decode(response.encoding or fallback_encoding) data = response_content.decode(response.encoding or fallback_encoding)
if json and data: if json and data:
try: try:
data = _json.loads(data) return Response(response.status_code, _json.loads(data),
response_headers)
except _json.decoder.JSONDecodeError as e: except _json.decoder.JSONDecodeError as e:
raise HTTPParsingException(str(e)) raise HTTPParsingException(str(e))
if code: return Response(response.status_code, data, response_headers)
return response.status_code, data
else:
return data
def strip_html(s: str) -> str: def strip_html(s: str) -> str:
return bs4.BeautifulSoup(s, "lxml").get_text() return bs4.BeautifulSoup(s, "lxml").get_text()