Implement endpoint for auto-approval based on existing watch providers

app.py

@@ -1,13 +1,95 @@
 import os
-from flask import Flask
+from flask import Flask, request
+import json
+import urllib.parse
+from yaml import load, dump
+
+from overseerrpy import Overseerr
+
+try:
+    from yaml import CLoader as Loader, CDumper as Dumper
+except ImportError:
+    from yaml import Loader, Dumper
+
+config = {}
+if os.path.exists("config.yaml"):
+    config = load(open('config.yaml'), Loader=Loader)
+
+def get_config(key, default=None, ctor: callable = str, data=None):
+
+    if data is None:
+        global config
+        data = config
+
+    if key in data:
+        return ctor(data[key])
+
+    subkeys = key.split('.')
+    subkey = subkeys[0]
+    if subkey in data:
+        return get_config(".".join(subkeys[1:]), default=default, ctor=ctor, data=data[subkey])
+
+    return default
+
 
 app = Flask(__name__)
+overseerr = Overseerr(ssl=get_config("overseerr.ssl", False, bool), host=get_config("overseerr.host"), port=get_config("overseerr.port", 80, int), api_key=get_config('overseerr.api_key'))
 
+try:
+    print(overseerr.test_connection())
+except Exception as e:
+    print(e)
+    exit(1)
 
 @app.route('/')
 def hello_world():  # put application's code here
     return 'Hello World!'
 
 
+def get_providers(provider_list):
+    distinct_providers = {}
+    for provider in provider_list:
+        if provider["iso_3166_1"] != "GB":
+            continue
+        for flatrate in provider["flatrate"]:
+            distinct_providers[flatrate["id"]] = flatrate["name"]
+
+    return distinct_providers
+
+@app.route("/overseerr", methods=['POST'])
+def overseerr_api():
+    data = request.json
+    print(data)
+
+    if data['notification_type'] != "MEDIA_PENDING":
+        return "Notification type not supported"
+
+    media = data['media']
+    media_type = media['media_type']
+
+    show_data = None
+    if media_type == "tv":
+        show_id = int(media["tmdbId"])
+        show_data = overseerr.get_tv_show(show_id)
+    elif media_type == "movie":
+        show_id = int(media["tmdbId"])
+        show_data = overseerr.get_movie(show_id)
+
+    if show_data:
+        providers = get_providers(show_data["watchProviders"])
+        media_request = data["request"]
+
+        if len(providers) > 0:
+            overseerr.update_request_status(int(media_request["request_id"]), "decline")
+            print(f"Request declined, \"{data['subject']}\" is available for streaming on the following providers:")
+            for provider in providers.values():
+                print(f"\t{provider}")
+        else:
+            print(f"Request for \"{data['subject']}\" approved.")
+            overseerr.update_request_status(int(media_request["request_id"]), "approve")
+
+    return "Hello Overseerr!"
+
+
 if __name__ == '__main__':
-    app.run(host=os.environ["SERVER_IP"], port=int(os.environ["SERVER_PORT"]))
+    app.run(host=os.environ.get("SERVER_IP", "0.0.0.0"), port=int(os.environ.get("SERVER_PORT", 5000)))

overseerrpy.py

@@ -0,0 +1,39 @@
+from typing import Union
+
+import requests
+
+
+class Overseerr(object):
+    def __init__(self, host: str, port: int, api_key: str, ssl: bool = True):
+        self.host = host
+        self.port = port
+        self.api_key = api_key
+        self.ssl = ssl
+
+    @property
+    def protocol(self):
+        return "https://" if self.ssl else "http://"
+
+    def _get(self, endpoint: str, params: dict = None):
+        url = self.protocol + self.host + ":" + str(self.port) + "/api/v1" + endpoint
+        return requests.get(url, headers={"X-Api-Key": self.api_key}, params=params)
+
+    def _post(self, endpoint: str, params: dict = None):
+        url = self.protocol + self.host + ":" + str(self.port) + "/api/v1" + endpoint
+        return requests.post(url, headers={"X-Api-Key": self.api_key}, json=params)
+
+    def test_connection(self):
+        print("Testing connection to Overseerr @", self.host)
+        settings = self._get("/settings/main").json()
+        if settings['applicationUrl'] == "":
+            return self.host
+        return settings['applicationUrl']
+
+    def get_tv_show(self, id):
+        return self._get(f"/tv/{id}").json()
+
+    def get_movie(self, id):
+        return self._get(f"/movie/{id}").json()
+
+    def update_request_status(self, req_id: int, status: Union["approve", "decline"]):
+        return self._post(f"/request/{req_id}/{status}").json()

requirements.txt

@@ -1 +1,3 @@
-Flask~=3.0.3
+Flask~=3.0.3
+PyYAML~=6.0.1
+requests~=2.32.3