Skip to content

Commit

Permalink
Merge pull request #1 from vaparr/code-format
Browse files Browse the repository at this point in the history 
updated attribute names and code formatting
  • Loading branch information
@vaparr
vaparr authored Jan 4, 2021
2 parents 0eca407 + f8d8f6b commit 813a1d3
Show file tree
Hide file tree
Showing 2 changed files with 142 additions and 107 deletions.
247 changes: 141 additions & 106 deletions pyoverseerr/pyoverseerr.py
View file
Original file line number Diff line number Diff line change
@@ -1,7 +1,5 @@
"""A python module to retrieve information from Overseerr.
"""A python module to retrieve information from Overseerr. For license information, see the LICENSE.txt file."""

For license information, see the LICENSE.txt file.
"""
import json
import logging

Expand All @@ -11,31 +9,28 @@

def request(f):
r = f().json()
print ("return:", r)
print("return:", r)
return r


class Overseerr(object):
"""A class for handling connections with an Overseerr instance."""

def __init__(
self, ssl, username, host, port, urlbase="", api_key=None, password=None
):
def __init__(self, ssl, username, host, port, urlbase="", api_key=None, password=None):

self._base_url = _BASE_URL.format(
ssl="s" if ssl else "", host=host, port=port, urlbase=urlbase
)
self._base_url = _BASE_URL.format(ssl="s" if ssl else "", host=host, port=port, urlbase=urlbase)

self._api_key = api_key
self._username = username
self._password = password
self._auth = None

def test_connection(self):
print ("Testing connections to Overseerr @",self._base_url)
print("Testing connection to Overseerr @", self._base_url)
self._request_connection(path="Settings/Public")

def _request_connection(self, path, post_data=None, auth=True):

import requests

url = f"{self._base_url}{path}"
Expand Down Expand Up @@ -78,14 +73,13 @@ def authenticate(self):
self._auth = {"X-Api-Key": self._api_key}
return

# credentials = {"userName": self._username, "password": self._password}

# token = (
# self._request_connection(path="Token", post_data=credentials, auth=False)
# .json()
# .get("access_token")
# )
# self._auth = {"Authorization": f"Bearer {token}"}
# credentials = {"userName": self._username, "password": self._password}
# token = (
# self._request_connection(path="Token", post_data=credentials, auth=False)
# .json()
# .get("access_token")
# )
# self._auth = {"Authorization": f"Bearer {token}"}

def search_movie(self, query):
return self._request_connection(f"search?query={query}&page=1&language=en").json()
Expand All @@ -94,167 +88,208 @@ def search_tv(self, query):
return self._request_connection(f"search?query={query}&page=1&language=en").json()

def search_music_album(self, query):
# return self._request_connection(f"search?query={query}&page=1&language=en").json()
return
# return self._request_connection(f"search?query={query}&page=1&language=en").json()
return

def request_movie(self, movie_id):
data = {
"mediatType": "movie",
"mediaType": "movie",
"mediaId": movie_id,
}
print(data)
request(lambda: self._request_connection(path="request", post_data=data))

def request_tv(
self, tv_id, request_all=False, request_latest=False, request_first=False
):

tv_data = self._request_connection(f"tv/{tv_id}").json()
tvdb_id = tv_data['externalIds']['tvdbId']

if ((request_all == False) and (request_latest == False) and (request_first == False)):
request_all = True

if (request_all == True):
seasons_array = []
for season in tv_data['seasons']:
if (season['seasonNumber'] == 0):
continue
seasons_array.append(season['seasonNumber'])

if (request_latest == True):
for season in tv_data['seasons']:
latest_season = season['seasonNumber']
seasons_array = [ latest_season ]

if (request_first == True):
seasons_array = [ 1 ]

data = {
def request_tv(self, tv_id, request_all=False, request_latest=False, request_first=False):

tv_data = self._request_connection(f"tv/{tv_id}").json()
tvdb_id = tv_data["externalIds"]["tvdbId"]

if (request_all == False) and (request_latest == False) and (request_first == False):
request_all = True

if request_all == True:
seasons_array = []
for season in tv_data["seasons"]:
if season["seasonNumber"] == 0:
continue
seasons_array.append(season["seasonNumber"])

if request_latest == True:
for season in tv_data["seasons"]:
latest_season = season["seasonNumber"]
seasons_array = [latest_season]

if request_first == True:
seasons_array = [1]

data = {
"mediaType": "tv",
"mediaId": tv_id,
"tvdbId": tvdb_id,
"seasons": seasons_array }
"seasons": seasons_array,
}

print (data)
request(lambda: self._request_connection(path="request", post_data=data))
print(data)
request(lambda: self._request_connection(path="request", post_data=data))

def request_music(self, album_id):
#data = {"foreignAlbumId": album_id}
#request(lambda: self._request_connection(path="Request/music", post_data=data))
# data = {"foreignAlbumId": album_id}
# request(lambda: self._request_connection(path="Request/music", post_data=data))
return

@property
def movie_requests(self):
requests = self._request_connection("Request").json()['results']
requests = self._request_connection("Request").json()["results"]
i = 0
for request in requests:
if request['type'] == "movie":
i += 1
if request["type"] == "movie":
i += 1
return i

@property
def last_movie_request(self):
requests = self._request_connection("Request").json()['results']
requests = self._request_connection("Request").json()["results"]
tmdb_id = None
for request in requests:
if request['type'] == "movie":
tmdb_id = request['media']['tmdbId']
break

if (tmdb_id is not None):
movie_data = self._request_connection(f"movie/{tmdb_id}").json()
return {"Name" :movie_data['title'], "Type" : "movie", "RequestedBy" : request['requestedBy']['username'] }
return {"Name" :None, "Type" : None, "RequestedBy" : None }
if request["type"] == "movie":
tmdb_id = request["media"]["tmdbId"]
break

if tmdb_id is not None:
movie_data = self._request_connection(f"movie/{tmdb_id}").json()
return {
"last_request_id": request["id"],
"last_request_status": request["status"],
"last_request_created": request["createdAt"],
"last_request_title": movie_data["title"],
"last_request_type": request["type"],
"last_request_username": request["requestedBy"]["username"],
}
return None

@property
def last_tv_request(self):
requests = self._request_connection("Request").json()['results']
requests = self._request_connection("Request").json()["results"]
tmdb_id = None
for request in requests:
if request['type'] == "tv":
tmdb_id = request['media']['tmdbId']
break
if request["type"] == "tv":
tmdb_id = request["media"]["tmdbId"]
break

if (tmdb_id is not None):
if tmdb_id is not None:
tv_data = self._request_connection(f"tv/{tmdb_id}").json()
return {"Name" :tv_data['name'], "Type" : "tv", "RequestedBy" : request['requestedBy']['username'] }
return {
"last_request_id": request["id"],
"last_request_status": request["status"],
"last_request_created": request["createdAt"],
"last_request_title": tv_data["name"],
"last_request_type": request["type"],
"last_request_username": request["requestedBy"]["username"],
}

return {"Name" :None, "Type" : None, "RequestedBy" : None }
return None

@property
def tv_requests(self):
requests = self._request_connection("Request").json()['results']
requests = self._request_connection("Request").json()["results"]
i = 0
for request in requests:
if request['type'] == "tv":
i += 1
if request["type"] == "tv":
i += 1
return i

@property
def music_requests(self):
# requests = self._request_connection("Request/music/total").text
# return 0 if requests is None else requests
# requests = self._request_connection("Request/music/total").text
# return 0 if requests is None else requests
return 0

@property
def total_requests(self):
return len(self._request_connection("Request").json()['results'])
return len(self._request_connection("Request").json()["results"])

@property
def available_requests(self):
return len(self._request_connection("Request?filter=available").json()['results'])
return len(self._request_connection("Request?filter=available").json()["results"])

@property
def pending_requests(self):
return len(self._request_connection("Request?filter=pending").json()['results'])
return len(self._request_connection("Request?filter=pending").json()["results"])

@property
def last_pending_request(self):
requests = self._request_connection("Request?filter=pending").json()['results']
requests = self._request_connection("Request?filter=pending").json()["results"]
tmdb_id = None
for request in requests:
tmdb_id = request['media']['tmdbId']
break

if (tmdb_id is not None):
if request['type'] == "tv":
tv_data = self._request_connection(f"tv/{tmdb_id}").json()
return {"Name" :tv_data['name'], "Type" : "tv", "RequestedBy" : request['requestedBy']['username'] }
tmdb_id = request["media"]["tmdbId"]
break

if request['type'] == "movie":
movie_data = self._request_connection(f"movie/{tmdb_id}").json()
return {"Name" :movie_data['title'], "Type" : "movie", "RequestedBy" : request['requestedBy']['username'] }
if tmdb_id is not None:
if request["type"] == "tv":
tv_data = self._request_connection(f"tv/{tmdb_id}").json()
return {
"last_request_id": request["id"],
"last_request_status": request["status"],
"last_request_created": request["createdAt"],
"last_request_title": tv_data["name"],
"last_request_type": request["type"],
"last_request_username": request["requestedBy"]["username"],
}

if request["type"] == "movie":
movie_data = self._request_connection(f"movie/{tmdb_id}").json()
return {
"last_request_id": request["id"],
"last_request_status": request["status"],
"last_request_created": request["createdAt"],
"last_request_title": movie_data["title"],
"last_request_type": request["type"],
"last_request_username": request["requestedBy"]["username"],
}

return {"Name" :None, "Type" : None, "RequestedBy" : None }
return None

@property
def last_total_request(self):
requests = self._request_connection("Request").json()['results']
requests = self._request_connection("Request").json()["results"]
tmdb_id = None
for request in requests:
tmdb_id = request['media']['tmdbId']
break
tmdb_id = request["media"]["tmdbId"]
break

if (tmdb_id is not None):
if request['type'] == "tv":
tv_data = self._request_connection(f"tv/{tmdb_id}").json()
return {"Name" :tv_data['name'], "Type" : "tv", "RequestedBy" : request['requestedBy']['username'] }

if request['type'] == "movie":
movie_data = self._request_connection(f"movie/{tmdb_id}").json()
return {"Name" :movie_data['title'], "Type" : "movie", "RequestedBy" : request['requestedBy']['username'] }
if tmdb_id is not None:
if request["type"] == "tv":
tv_data = self._request_connection(f"tv/{tmdb_id}").json()
return {
"last_request_id": request["id"],
"last_request_status": request["status"],
"last_request_created": request["createdAt"],
"last_request_title": tv_data["name"],
"last_request_type": request["type"],
"last_request_username": request["requestedBy"]["username"],
}

if request["type"] == "movie":
movie_data = self._request_connection(f"movie/{tmdb_id}").json()
return {
"last_request_id": request["id"],
"last_request_status": request["status"],
"last_request_created": request["createdAt"],
"last_request_title": movie_data["title"],
"last_request_type": request["type"],
"last_request_username": request["requestedBy"]["username"],
}

return None


@property
def approved_requests(self):
return len(self._request_connection("Request?filter=approved").json()['results'])
return len(self._request_connection("Request?filter=approved").json()["results"])

@property
def unavailable_requests(self):
return len(self._request_connection("Request?filter=unavailable").json()['results'])
return len(self._request_connection("Request?filter=unavailable").json()["results"])


class OverseerrError(Exception):
pass
2 changes: 1 addition & 1 deletion setup.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@

setup(
name="pyoverseerr",
version="0.1.13",
version="0.1.14",
url="https://github.com/vaparr/pyoverseerr",
author="Vaparr",
author_email="Vaparr@vaparr.org",
Expand Down

0 comments on commit 813a1d3

Please sign in to comment.