4
1
Fork 1

Compare commits

..

6 Commits

Author SHA1 Message Date
trueold89 1e3ce3a55e
Add PN_CACHE env value
- Add ENV abstract class to static
- Init static.env
- Add PN_CACHE env
2024-08-05 17:43:58 +03:00
trueold89 015e01b6be
Add PythonCache implementation of CacheDB
- add User class to db.types
- init db.cache
- Add PythonCache class
2024-08-05 16:43:00 +03:00
trueold89 ba696f4c56
Add CacheDB abstract class
- init tubot.db
- init tubot.db.types
- init tubot.db.abc
- add CacheDBTypes enum
- add UserStates enum
- add CacheDB abstract class
2024-08-05 15:42:35 +03:00
trueold89 ebca2172f4
Add JellyfinAPI implementation of DirGetter 2024-08-05 15:22:29 +03:00
trueold89 b00c60f448
Add OSGetter implementation of DirGetter
- init dirgetter.getter
- add OSGetter class
2024-08-05 14:56:18 +03:00
trueold89 0b48cb152e
Add DirectoryGetter abstract class
- init dirgetter module
- init dirgetter.types
- init girgetter.abc
- Add DirGetter absctract class
2024-08-03 16:49:22 +03:00
10 changed files with 437 additions and 1 deletions

View File

@ -8,5 +8,5 @@ setup(
author_email="root@orudo.ru", 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.)", 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"], 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"],
) )

94
tubot/db/abc.py Normal file
View File

@ -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

102
tubot/db/cache.py Normal file
View File

@ -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

52
tubot/db/types.py Normal file
View File

@ -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,
}

View File

32
tubot/dirgetter/abc.py Normal file
View File

@ -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

96
tubot/dirgetter/getter.py Normal file
View File

@ -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

17
tubot/dirgetter/types.py Normal file
View File

@ -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"

View File

@ -6,6 +6,7 @@
# Imports # Imports
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from os import environ
class IValidatable(ABC): class IValidatable(ABC):
@ -22,3 +23,28 @@ class IValidatable(ABC):
:return: Object validity boolean :return: Object validity boolean
""" """
raise NotImplementedError 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

17
tubot/static/env.py Normal file
View File

@ -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"