Initial commit

al_arr_sync/__main__.py

@@ -0,0 +1,37 @@
+import os
+import typing
+from dotenv import load_dotenv
+from al_arr_sync.anilist import AniListClient
+from al_arr_sync.sonarr import SonarrClient
+
+load_dotenv()
+
+
+def main() -> int:
+    al = AniListClient()
+    sonarr = SonarrClient.from_env()
+
+    username = os.environ["ANILIST_USERNAME"]
+    media = al.currently_watching(username)
+    
+    series: typing.List[typing.Any] = []
+    for entry in media:
+        media_format = entry["media"]["format"]
+        if media_format == "TV":
+            series.append(entry)
+
+    for show in series:
+        show_name = show["media"]["title"]["english"]
+        results = sonarr.lookup_series(show_name)
+        try:
+            sonarr.add_series(results[0])
+            print(f"Successfully added series {show_name}")
+        except Exception as e:
+            print(e)
+            continue
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

al_arr_sync/anilist.py

@@ -0,0 +1,46 @@
+import typing
+from requests import Session
+
+if typing.TYPE_CHECKING:
+    AnyDict = typing.Dict[typing.Any, typing.Any]
+
+
+USER_WATCHING_QUERY = """
+query ($userName: String) {
+  MediaListCollection(userName: $userName, type:ANIME, status:CURRENT) {
+    lists {
+      entries {
+        media {
+          title {
+            english
+          }
+          format
+        }
+      }
+    }
+  }
+}
+"""
+
+
+class AniListClient:
+    def __init__(self) -> None:
+        self.graphql_api_url = "https://graphql.anilist.co/"
+        self.http_session = Session()
+
+    def currently_watching(self, username: str) -> typing.List['AnyDict']:
+        resp = self.http_session.post(
+            self.graphql_api_url,
+            json={"query": USER_WATCHING_QUERY, "variables": {"userName": username}},
+        )
+        resp_data: 'AnyDict' = resp.json()
+        if errors := resp_data.get("errors"):
+            raise Exception(errors)
+
+        media = [
+            entry
+            for lst in resp_data["data"]["MediaListCollection"]["lists"]
+            for entry in lst["entries"]
+        ]
+
+        return media

al_arr_sync/sonarr.py

@@ -0,0 +1,61 @@
+import typing
+from requests import Session
+from requests import PreparedRequest
+import os
+from urllib.parse import urljoin
+
+if typing.TYPE_CHECKING:
+    AnyDict = typing.Dict[typing.Any, typing.Any]
+
+class SonarrClient:
+    def __init__(self, sonarr_url: str, api_key: str) -> None:
+        self.sonarr_url = sonarr_url
+        self.api_key = api_key
+        self.http_session = Session()
+
+    @staticmethod
+    def from_env() -> "SonarrClient":
+        return SonarrClient(
+            sonarr_url=os.environ["SONARR_API_URL"],
+            api_key=os.environ["SONARR_API_KEY"],
+        )
+
+    def _prepare_request(
+        self,
+        endpoint: str,
+        method: str = "GET",
+        params: 'AnyDict' = {},
+        json: typing.Optional['AnyDict'] = None,
+    ) -> PreparedRequest:
+        url = urljoin(self.sonarr_url, endpoint)
+        headers = {"X-Api-Key": self.api_key}
+
+        req = PreparedRequest()
+        req.prepare(method=method, url=url, headers=headers, params=params, json=json)
+        return req
+
+    def lookup_series(self, title: str) -> typing.List['AnyDict']:
+        req = self._prepare_request("/api/v3/series/lookup", params={"term": title})
+        resp = self.http_session.send(req)
+        return resp.json()
+
+    def add_series(self, *series: 'AnyDict'):
+        for show in series:
+            payload: 'AnyDict' = show.copy()
+            payload.update(
+                {
+                    "addOptions": {
+                        "monitor": "future",
+                        "searchForCutoffUnmetEpisodes": False,
+                        "searchForMissingEpisodes": False,
+                    },
+                    "rootFolderPath": os.environ["SONARR_FOLDER_PATH"],
+                    "qualityProfileId": int(os.environ["SONARR_QUALITY_PROFILE"]),
+                    "languageProfileId": int(os.environ["SONARR_LANGUAGE_PROFILE"]),
+                }
+            )
+            req = self._prepare_request("/api/v3/series", method="POST", json=payload)
+            resp = self.http_session.send(req)
+
+            if resp.status_code != 201:
+                raise Exception(f"Failed to add series {show['title']}:\n{resp.json()}")