TorrentUplouderBot/tubot/dirgetter/getter.py

97 lines
2.6 KiB
Python
Raw Normal View History

# -*- coding: utf-8 -*-
####################################
# Directory-Getter implementations #
####################################
# Imports
from tubot.dirgetter.types import GetterTypes
from tubot.dirgetter.abc import DirGetter
from aiofiles.os import listdir
from aiofiles.ospath import isdir
from aiohttp import ClientResponse, ClientSession
class OSGetter(DirGetter):
"""
Python.os module implementation of DirectoryGetter
"""
_gtype = GetterTypes.OS
base_dir: str
def __init__(self, base_dir: str) -> None:
"""
:param base_dir: Path to parent directory
"""
super().__init__()
self.base_dir = base_dir
@property
async def folders(self) -> dict:
dirs = {}
ls = await listdir(self.base_dir)
if len(ls) == 0:
raise ValueError("No dirs found")
for item in ls:
if await isdir(f"{self.base_dir}/{item}"):
dirs[item] = f"{self.base_dir}/{item}"
return dirs
async def __validate__(self) -> bool:
return await isdir(self.base_dir)
class Jellyfin(DirGetter):
"""
Jellyfin API implementation of DirectoryGetter
"""
_gtype = GetterTypes.Jellyfin
host: str
token: str
def __init__(self, host: str, api_token: str) -> None:
"""
:param host: Adress of Jellyfin server
:param api_token: Jellyfin API Token for auth
"""
super().__init__()
self.host = host
self.token = api_token
async def _get(self, api: str) -> ClientResponse:
async with ClientSession() as session:
resp = await session.get(f"{self.host}/{api}?api_key={self.token}")
status = resp.status
match status:
case 200:
return resp
case 401:
raise ConnectionError("401: Auth error")
case 403:
raise ConnectionError("403: Forbidden")
case 404:
raise ConnectionError("403: Not found")
raise ConnectionError()
@property
async def idx(self) -> str | None:
resp = await self._get("System/Info")
json = await resp.json()
return json["Id"]
@property
async def folders(self) -> dict:
resp = await self._get("Library/VirtualFolders")
json = await resp.json()
dirs = {}
for folder in json:
dirs[folder["Name"]] = folder["Locations"][0]
return dirs
async def __validate__(self) -> bool:
if await self.idx is not None:
return True
return False