Kyoo/scanner/providers/kyoo_client.py

import os
import logging
import jsons
from aiohttp import ClientSession
from typing import List, Literal, Any
from urllib.parse import quote


class KyooClient:
	def __init__(self) -> None:
		self._api_key = os.environ.get("KYOO_APIKEY")
		if not self._api_key:
			self._api_key = os.environ.get("KYOO_APIKEYS")
			if not self._api_key:
				print("Missing environment variable 'KYOO_APIKEY'.")
				exit(2)
			self._api_key = self._api_key.split(",")[0]

		self._url = os.environ.get("KYOO_URL", "http://back:5000")

	async def __aenter__(self):
		jsons.set_serializer(lambda x, **_: format_date(x), Optional[date | int])  # type: ignore
		self.client = ClientSession(
			json_serialize=lambda *args, **kwargs: jsons.dumps(
				*args, key_transformer=jsons.KEY_TRANSFORMER_CAMELCASE, **kwargs
			),
		)
		return self

	async def __aexit__(self, exc_type, exc_value, exc_tb):
		await self.client.close()

	async def get_registered_paths(self) -> List[str]:
		paths = None
		async with self.client.get(
			f"{self._url}/episodes",
			params={"limit": 0},
			headers={"X-API-Key": self._api_key},
		) as r:
			r.raise_for_status()
			ret = await r.json()
			paths = list(x["path"] for x in ret["items"])

		async with self.client.get(
			f"{self._url}/movies",
			params={"limit": 0},
			headers={"X-API-Key": self._api_key},
		) as r:
			r.raise_for_status()
			ret = await r.json()
			paths += list(x["path"] for x in ret["items"])
		return paths

	async def create_issue(self, path: str, issue: str, extra: dict | None = None):
		await self.client.post(
			f"{self._url}/issues",
			json={"domain": "scanner", "cause": path, "reason": issue, "extra": extra},
			headers={"X-API-Key": self._api_key},
		)

	async def delete_issue(self, path: str):
		await self.client.delete(
			f'{self._url}/issues?filter=domain eq scanner and cause eq "{path}"',
			headers={"X-API-Key": self._api_key},
		)

	async def link_collection(
		self, collection: str, type: Literal["movie"] | Literal["show"], id: str
	):
		async with self.client.put(
			f"{self._url}/collections/{collection}/{type}/{id}",
			headers={"X-API-Key": self._api_key},
		) as r:
			# Allow 409 and continue as if it worked.
			if not r.ok and r.status != 409:
				logging.error(f"Request error: {await r.text()}")
				r.raise_for_status()

	async def post(self, path: str, *, data: dict[str, Any]) -> str:
		logging.debug(
			"Sending %s: %s",
			path,
			jsons.dumps(
				data,
				key_transformer=jsons.KEY_TRANSFORMER_CAMELCASE,
				jdkwargs={"indent": 4},
			),
		)
		async with self.client.post(
			f"{self._url}/{path}",
			json=data,
			headers={"X-API-Key": self._api_key},
		) as r:
			# Allow 409 and continue as if it worked.
			if not r.ok and r.status != 409:
				logging.error(f"Request error: {await r.text()}")
				r.raise_for_status()
			ret = await r.json()

			if r.status == 409 and (
				(path == "shows" and ret["startAir"][:4] != str(data["start_air"].year))
				or (
					path == "movies"
					and ret["airDate"][:4] != str(data["air_date"].year)
				)
			):
				logging.info(
					f"Found a {path} with the same slug ({ret['slug']}) and a different date, using the date as part of the slug"
				)
				year = (data["start_air"] if path == "movie" else data["air_date"]).year
				data["slug"] = f"{ret['slug']}-{year}"
				return await self.post(path, data=data)
			return ret["id"]

	async def delete(
		self,
		path: str,
		type: Literal["episode", "movie"] | None = None,
	):
		logging.info("Deleting %s", path)

		if type is None or type == "movie":
			async with self.client.delete(
				f'{self._url}/movies?filter=path eq "{quote(path)}"',
				headers={"X-API-Key": self._api_key},
			) as r:
				if not r.ok:
					logging.error(f"Request error: {await r.text()}")
					r.raise_for_status()

		if type is None or type == "episode":
			async with self.client.delete(
				f'{self._url}/episodes?filter=path eq "{quote(path)}"',
				headers={"X-API-Key": self._api_key},
			) as r:
				if not r.ok:
					logging.error(f"Request error: {await r.text()}")
					r.raise_for_status()

		await self.delete_issue(path)