Add PN_CACHE env value

- Add ENV abstract class to static
- Init static.env
- Add PN_CACHE env

setup.py

@@ -8,5 +8,5 @@ setup(
     author_email="root@orudo.ru",
     description="A simple Telegram bot that will allow you to upload torrent files / magnet links to a remote Torrent server (qBitTorrent, Transmission, etc.)",
     install_requires=["aiohttp>=3.10.0", "aiofiles>=24.1.0", "aiofiles>=24.1.0"],
-    packages=["tubot", "tubot.static", "tubot.torrent"],
+    packages=["tubot", "tubot.static", "tubot.torrent", "tubot.dirgetter", "db"],
 )

tubot/db/abc.py

@@ -0,0 +1,94 @@
+# -*- coding: utf-8 -*-
+
+####################################
+# DataBase module abstract classes #
+####################################
+
+# Imports
+from abc import ABC, abstractmethod
+from tubot.static.abc import IValidatable
+from tubot.db.types import CacheDBTypes, UserStates
+
+
+class CacheDB(IValidatable, ABC):
+    """
+    Abstract class for CacheDB
+    """
+
+    _ctype: CacheDBTypes
+
+    def __init__(self) -> None:
+        if self._ctype is None:
+            raise NotImplementedError("CacheDB type not implemented")
+
+    # Users
+
+    @abstractmethod
+    async def add_user(self, tg_id: int, name: str) -> None:
+        """
+        Add user to cache db
+
+        :param tg_id: User telegram id
+        :param name: User telegram name
+        """
+        raise NotImplementedError
+
+    @abstractmethod
+    async def user_state(self, tg_id: int) -> UserStates:
+        """
+        Get user state
+
+        :param tg_id: User telegram id
+        :return: Status of user
+        """
+        raise NotImplementedError
+
+    @abstractmethod
+    async def change_user_state(self, tg_id: int, status: UserStates) -> None:
+        """
+        Change user state
+
+        :param tg_id: User telegram id
+        :param status: New user status
+        """
+        raise NotImplementedError
+
+    @abstractmethod
+    async def auth_user(self, tg_id: int) -> None:
+        """
+        Auth user
+
+        :param tg_id: User telegram id
+        """
+        raise NotImplementedError
+
+    @abstractmethod
+    async def is_user_auth(self, tg_id: int) -> bool:
+        """
+        Check if user is already auth
+
+        :param tg_id: User telegram id
+        """
+        raise NotImplementedError
+
+    # Dirs
+
+    @abstractmethod
+    async def cache_dirs(self, dirs: dict, expire: int) -> None:
+        """
+        Cache dirs from DirectoryGetter
+
+        :param dirs: Dirs dict
+        :param expire: Expire time (in seconds)
+        """
+        raise NotImplementedError
+
+    @property
+    @abstractmethod
+    async def get_dirs(self) -> dict:
+        """
+        Returns precached dirs
+
+        :return: Dirs dict
+        """
+        raise NotImplementedError

tubot/db/cache.py

@@ -0,0 +1,102 @@
+# -*- coding: utf-8 -*-
+
+###########################
+# CacheDB implementations #
+###########################
+
+# Imports
+from tubot.db.abc import CacheDB
+from tubot.db.types import CacheDBTypes, UserStates, User
+from tubot.static.env import PN_CACHE
+from pickle import loads, dumps
+from aiofiles.ospath import isdir, isfile
+from aiofiles.os import mkdir
+from aiofiles import open
+from asyncio import sleep
+
+
+class PythonCache(CacheDB):
+    """
+    Native python implementation of Cache DataBase
+    """
+
+    CACHE_DIR = PN_CACHE()()
+    _ctype = CacheDBTypes.PythonPKL
+    users: dict
+    dirs: dict
+
+    def __init__(self) -> None:
+        super().__init__()
+
+    async def _init(self) -> bool:
+        self.users = {}
+        self.dirs = {}
+        if await isfile(f"{self.CACHE_DIR}/user_cache.pkl"):
+            try:
+                await self._load_pkl()
+            except Exception:
+                return False
+        return True
+
+    async def _load_pkl(self) -> None:
+        if not await isdir(self.CACHE_DIR):
+            await mkdir(self.CACHE_DIR)
+        async with open(f"{self.CACHE_DIR}/user_cache.pkl", "rb") as file:
+            buffer = await file.read()
+            pkl = loads(buffer)
+            self.users = pkl
+
+    async def _save_pkl(self) -> None:
+        if not await isdir(self.CACHE_DIR):
+            await mkdir(self.CACHE_DIR)
+        async with open(f"{self.CACHE_DIR}/user_cache.pkl", "wb") as file:
+            await file.write(dumps(self.users))
+
+    async def __validate__(self) -> bool:
+        return await self._init()
+
+    # Users
+
+    async def add_user(self, tg_id: int, name: str) -> None:
+        if tg_id in tuple(self.users.keys()):
+            raise ValueError("User already exists")
+        user = User(tg_id, name)
+        self.users[tg_id] = user.dict
+        await self._save_pkl()
+
+    async def user_state(self, tg_id: int) -> UserStates:
+        if tg_id not in tuple(self.users.keys()):
+            raise ValueError("User doesn't exists")
+        user = self.users[tg_id]
+        return user["state"]
+
+    async def change_user_state(self, tg_id: int, status: UserStates) -> None:
+        if tg_id not in tuple(self.users.keys()):
+            raise ValueError("User doesn't exists")
+        user = self.users[tg_id]
+        user["state"] = status
+        await self._save_pkl()
+
+    async def auth_user(self, tg_id: int) -> None:
+        if tg_id not in tuple(self.users.keys()):
+            raise ValueError("User doesn't exists")
+        user = self.users[tg_id]
+        user["auth"] = True
+        await self._save_pkl()
+
+    async def is_user_auth(self, tg_id: int) -> bool:
+        if tg_id not in tuple(self.users.keys()):
+            raise ValueError("User doesn't exists")
+        user = self.users[tg_id]
+        return user["auth"]
+
+    # Dirs
+
+    async def cache_dirs(self, dirs: dict, expire: int) -> None:
+        self.dirs = dirs
+        await sleep(expire)
+        self.dirs = {}
+
+    @property
+    async def get_dirs(self) -> dict:
+        return self.dirs

tubot/db/types.py

@@ -0,0 +1,52 @@
+# -*- coding: utf-8 -*-
+
+#############################
+# Types for DataBase module #
+#############################
+
+# Imports
+from enum import Enum
+
+
+class CacheDBTypes(Enum):
+    """
+    Types of CacheDB
+    """
+
+    PythonPKL = "python"
+
+
+class UserStates(Enum):
+    """
+    Types of User status
+    """
+
+    ...
+
+
+class User(object):
+    """
+    User class
+    """
+
+    tg_id: int
+    name: str
+    state: UserStates | None = None
+    auth: bool = False
+
+    def __init__(
+        self, tg_id: int, name: str, state: UserStates | None = None, auth: bool = False
+    ) -> None:
+        self.tg_id = tg_id
+        self.name = name
+        self.state = state
+        self.auth = auth
+
+    @property
+    def dict(self):
+        return {
+            "tg_id": self.tg_id,
+            "name": self.name,
+            "state": self.state,
+            "auth": self.auth,
+        }

tubot/dirgetter/abc.py

@@ -0,0 +1,32 @@
+# -*- coding: utf-8 -*-
+
+############################################
+# Directory-Getter module abstract classes #
+############################################
+
+# Imports
+from abc import ABC, abstractmethod
+from tubot.static.abc import IValidatable
+from tubot.dirgetter.types import GetterTypes
+
+
+class DirGetter(IValidatable, ABC):
+    """
+    DirectoryGetter Abstract class
+    """
+
+    _gtype: GetterTypes
+
+    def __init__(self) -> None:
+        if self._gtype is None:
+            raise NotImplementedError("DirGetter type is not implemented")
+
+    @property
+    @abstractmethod
+    async def folders(self) -> dict:
+        """
+        Returns a dictionary of media folders {name: path}
+
+        :return: Dict of media folders
+        """
+        raise NotImplementedError

tubot/dirgetter/getter.py

@@ -0,0 +1,96 @@
+# -*- 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

tubot/dirgetter/types.py

@@ -0,0 +1,17 @@
+# -*- coding: utf-8 -*-
+
+#####################################
+# Types for Directory-Getter module #
+#####################################
+
+# Imports
+from enum import Enum
+
+
+class GetterTypes(Enum):
+    """
+    Types of getters
+    """
+
+    OS = "Python os module"
+    Jellyfin = "Jelyfin API"

tubot/static/abc.py

@@ -6,6 +6,7 @@
 
 # Imports
 from abc import ABC, abstractmethod
+from os import environ
 
 
 class IValidatable(ABC):
@@ -22,3 +23,28 @@ class IValidatable(ABC):
         :return: Object validity boolean
         """
         raise NotImplementedError
+
+
+class ENV(object):
+
+    _name: str | None = None
+    DEFAULT: str
+
+    def __init__(self) -> None:
+        if self._name is None or self.DEFAULT is None:
+            raise NotImplementedError
+
+    @property
+    def from_os(self) -> str | None:
+        if self._name is not None:
+            return environ.get(self._name)
+
+    @property
+    def value(self) -> str:
+        val = self.from_os
+        if val is not None:
+            return val
+        return self.DEFAULT
+
+    def __call__(self) -> str:
+        return self.value

tubot/static/env.py

@@ -0,0 +1,17 @@
+# -*- coding: utf-8 -*-
+
+############
+# ENV Vars #
+############
+
+# Imports
+from tubot.static.abc import ENV
+
+
+class PN_CACHE(ENV):
+    """
+    Python Native Cache dir
+    """
+
+    _name = "PN_CACHE"
+    DEFAULT = "/etc/tubot"