diff --git a/.env.example b/.env.example index 29998e3..47fb032 100644 --- a/.env.example +++ b/.env.example @@ -3,4 +3,5 @@ TELEGRAM_BOT_TOKEN="123456789:XXXXXXXXXXXXXXXXXXXXXXXXXXXXX" TELEGRAM_ADMINS=[123456789,987654321] ### HEYZNER API KEY SETTINGS -HETZNER_API_KEY="XXXXXXXXXXXXXXXXXXXXXXXXXXXXX" +## if you one more api key, you can add like this [ ["name1","key1"] , ["name2","key2"] ] +HETZNER_API_KEYS=[ ["name","api key"] ] diff --git a/api/__init__.py b/api/__init__.py index 22b8073..be1536a 100644 --- a/api/__init__.py +++ b/api/__init__.py @@ -1,6 +1,3 @@ from .hetzner import HetznerManager -from config import EnvFile -HetznerAPI = HetznerManager(key=EnvFile.HETZNER_API_KEY) - -__all__ = ["HetznerAPI"] +__all__ = ["HetznerManager"] diff --git a/api/hetzner.py b/api/hetzner.py index 95a5ac5..01493da 100644 --- a/api/hetzner.py +++ b/api/hetzner.py @@ -33,74 +33,102 @@ async def wrapper(*args, **kwargs): class HetznerManager: - def __init__(self, key: str) -> None: - self.client = Client(token=key) - + @staticmethod @handle_hetzner_errors - async def get_servers(self) -> List[Server]: - return self.client.servers.get_all() + async def get_servers(key: str) -> List[Server]: + client = Client(token=key) + return client.servers.get_all() + @staticmethod @handle_hetzner_errors - async def get_server(self, server_id: int) -> Optional[Server]: - return self.client.servers.get_by_id(server_id) + async def get_server(key: str, server_id: int) -> Optional[Server]: + client = Client(token=key) + return client.servers.get_by_id(server_id) + @staticmethod @handle_hetzner_errors - async def get_server_type(self, server_id: int) -> Optional[ServerType]: - return self.client.server_types.get_by_id(server_id) + async def get_server_type(key: str, server_id: int) -> Optional[ServerType]: + client = Client(token=key) + return client.server_types.get_by_id(server_id) + @staticmethod @handle_hetzner_errors - async def power_on(self, server: Server) -> Optional[Action]: - return self.client.servers.power_on(server) + async def power_on(key: str, server: Server) -> Optional[Action]: + client = Client(token=key) + return client.servers.power_on(server) + @staticmethod @handle_hetzner_errors - async def power_off(self, server: Server) -> Optional[Action]: - return self.client.servers.power_off(server) + async def power_off(key: str, server: Server) -> Optional[Action]: + client = Client(token=key) + return client.servers.power_off(server) + @staticmethod @handle_hetzner_errors - async def reboot(self, server: Server) -> Optional[Action]: - return self.client.servers.reboot(server) + async def reboot(key: str, server: Server) -> Optional[Action]: + client = Client(token=key) + return client.servers.reboot(server) + @staticmethod @handle_hetzner_errors - async def reset_password(self, server: Server) -> Optional[str]: - return self.client.servers.reset_password(server).root_password + async def reset_password(key: str, server: Server) -> Optional[str]: + client = Client(token=key) + return client.servers.reset_password(server).root_password + @staticmethod @handle_hetzner_errors - async def delete(self, server: Server) -> Optional[Action]: - return self.client.servers.delete(server) + async def delete(key: str, server: Server) -> Optional[Action]: + client = Client(token=key) + return client.servers.delete(server) + @staticmethod @handle_hetzner_errors - async def reset(self, server: Server) -> Optional[Action]: - return self.client.servers.reset(server) + async def reset(key: str, server: Server) -> Optional[Action]: + client = Client(token=key) + return client.servers.reset(server) + @staticmethod @handle_hetzner_errors - async def get_images(self, arch: str = None) -> List[Image]: - return self.client.images.get_all(architecture=arch) + async def get_images(key: str, arch: str = None) -> List[Image]: + client = Client(token=key) + return client.images.get_all(architecture=arch) + @staticmethod @handle_hetzner_errors - async def rebuild_server(self, server: Server, image_id: int) -> Optional[Action]: - image = self.client.images.get_by_id(image_id) - return self.client.servers.rebuild(server, image) - + async def rebuild_server( + key: str, server: Server, image_id: int + ) -> Optional[Action]: + client = Client(token=key) + image = client.images.get_by_id(image_id) + return client.servers.rebuild(server, image) + + @staticmethod @handle_hetzner_errors async def create_server( - self, name: str, server_type: ServerType, image: Image + key: str, name: str, server_type: ServerType, image: Image ) -> Optional[Server]: - server = self.client.servers.create( - name=name, server_type=server_type, image=image - ) + client = Client(token=key) + server = client.servers.create(name=name, server_type=server_type, image=image) return server.server - async def get_image(self, id: int) -> Optional[Image]: - return self.client.images.get_by_id(id) + async def get_image(key: str, id: int) -> Optional[Image]: + client = Client(token=key) + return client.images.get_by_id(id) + @staticmethod @handle_hetzner_errors - async def get_server_types(self) -> List[ServerType]: - return self.client.server_types.get_all() + async def get_server_types(key: str) -> List[ServerType]: + client = Client(token=key) + return client.server_types.get_all() + @staticmethod @handle_hetzner_errors - async def get_datacenters(self) -> List[Datacenter]: - return self.client.datacenters.get_all() + async def get_datacenters(key: str) -> List[Datacenter]: + client = Client(token=key) + return client.datacenters.get_all() + @staticmethod @handle_hetzner_errors - async def get_datacenter(self, id: int) -> Optional[Datacenter]: - return self.client.datacenters.get_by_id(id) + async def get_datacenter(key: str, id: int) -> Optional[Datacenter]: + client = Client(token=key) + return client.datacenters.get_by_id(id) diff --git a/config/env.py b/config/env.py index c6de1ec..2785699 100644 --- a/config/env.py +++ b/config/env.py @@ -1,3 +1,4 @@ +import hashlib from pydantic_settings import BaseSettings, SettingsConfigDict @@ -10,8 +11,22 @@ class EnvFileReader(BaseSettings): TELEGRAM_BOT_TOKEN: str = "" TELEGRAM_ADMINS: list[int] = [] - HETZNER_API_KEY: str = "" + HETZNER_API_KEYS: list[tuple[str, str]] = [] def is_admin(self, userid: int) -> bool: """check userid is admin or not""" return userid in self.TELEGRAM_ADMINS + + def to_hash(self, key: str, length: int = 4) -> str: + """create a hash from key""" + hash_object = hashlib.md5(key.encode()) + full_hash = hash_object.hexdigest() + return full_hash[:length] + + def from_hash(self, hashkey: str, length: int = 4) -> str: + """create a key from hash""" + for key_pair in self.HETZNER_API_KEYS: + current_hash = self.to_hash(key_pair[1]) + if current_hash == hashkey: + return key_pair[1] + return None diff --git a/keys/__init__.py b/keys/__init__.py index 2c031a3..b6c427f 100644 --- a/keys/__init__.py +++ b/keys/__init__.py @@ -1,21 +1,15 @@ -from .action import Actions -from .callback import ( - ServerAction, - ServerList, - ServerTypeSelect, - LocationTypeSelect, - ImageTypeSelect, -) +from .enums import Actions, Pages, ServerCreate, ServerUpdate +from .callback import PageCB, SelectCB from .keyboard import KeyboardsCreater Keyboards = KeyboardsCreater() __all__ = [ "Actions", - "ServerAction", - "ServerList", + "PageCB", + "SelectCB", + "Pages", "Keyboards", - "ServerTypeSelect", - "LocationTypeSelect", - "ImageTypeSelect", + "ServerCreate", + "ServerUpdate", ] diff --git a/keys/callback.py b/keys/callback.py index 9d6890f..2d30bae 100644 --- a/keys/callback.py +++ b/keys/callback.py @@ -1,27 +1,19 @@ from aiogram.filters.callback_data import CallbackData +from .enums import Pages, Actions, ServerCreate, ServerUpdate -class ServerAction(CallbackData, prefix="server_action"): - action: str - server_id: int +class PageCB(CallbackData, prefix="pages"): + key: str = "KEY" + page: Pages | None = None + action: Actions | ServerUpdate | None = None + server_id: int | None = None + image_id: int | None = None confirm: bool = False - image_id: int = 0 -class ServerList(CallbackData, prefix="server_list"): - action: str - - -class ServerTypeSelect(CallbackData, prefix="server_type"): - server: int = 0 - is_select: bool = False - - -class LocationTypeSelect(CallbackData, prefix="location_type"): - location: int = 0 - is_select: bool = False - - -class ImageTypeSelect(CallbackData, prefix="image_type"): - image: int = 0 - is_select: bool = False +class SelectCB(CallbackData, prefix="select"): + key: str = "KEY" + page: Pages | None = None + action: Actions | None = None + datatype: ServerCreate | None = None + datavalue: str | int | None = None diff --git a/keys/enums/__init__.py b/keys/enums/__init__.py new file mode 100644 index 0000000..e4f4917 --- /dev/null +++ b/keys/enums/__init__.py @@ -0,0 +1,4 @@ +from .callbacks import Pages, Actions +from .servers import ServerCreate, ServerUpdate + +__all__ = ["Pages", "Actions", "ServerCreate", "ServerUpdate"] diff --git a/keys/enums/callbacks.py b/keys/enums/callbacks.py new file mode 100644 index 0000000..2355493 --- /dev/null +++ b/keys/enums/callbacks.py @@ -0,0 +1,16 @@ +from enum import Enum + + +class Actions(str, Enum): + HOME = "home" + LIST = "list" + INFO = "info" + CREATE = "create" + UPDATE = "update" + DELETE = "delete" + + +class Pages(str, Enum): + HOME = "home" + MENU = "menu" + SERVER = "server" diff --git a/keys/action.py b/keys/enums/servers.py similarity index 63% rename from keys/action.py rename to keys/enums/servers.py index a2270d5..093a66e 100644 --- a/keys/action.py +++ b/keys/enums/servers.py @@ -1,14 +1,18 @@ from enum import Enum -class Actions(str, Enum): +class ServerCreate(str, Enum): + LOCATION = "location" + SERVER = "server" + IMAGE = "image" + + +class ServerUpdate(str, Enum): POWER_ON = "power_on" POWER_OFF = "power_off" REBOOT = "reboot" RESET_PASSWORD = "reset_password" DELETE = "delete" - HOME = "home" - INFO = "info" REBUILD = "rebuild" - UPDATE = "update" RESET = "reset" + UPDATE = "update" diff --git a/keys/keyboard.py b/keys/keyboard.py index 2668877..1e3f36d 100644 --- a/keys/keyboard.py +++ b/keys/keyboard.py @@ -1,4 +1,4 @@ -from aiogram.types import InlineKeyboardButton +from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup from aiogram.utils.keyboard import InlineKeyboardBuilder from hcloud.servers.domain import Server @@ -6,101 +6,153 @@ from hcloud.server_types.domain import ServerType from hcloud.datacenters.domain import Datacenter -from .callback import ( - ServerAction, - ServerList, - ServerTypeSelect, - LocationTypeSelect, - ImageTypeSelect, -) -from .action import Actions +from .callback import PageCB, SelectCB +from .enums import Actions, Pages, ServerCreate, ServerUpdate from language import KeyboardText +from config import EnvFile class KeyboardsCreater: - def home(self): + def home(self, keys: list[list[str]]) -> InlineKeyboardMarkup: builder = InlineKeyboardBuilder() + for key in keys: + builder.button( + text=key[0], + callback_data=PageCB( + key=EnvFile.to_hash(key[1]), page=Pages.MENU + ).pack(), + ) + builder.button( - text=KeyboardText.BACK, callback_data=ServerList(action=Actions.HOME).pack() + text=KeyboardText.UPDATE, callback_data=PageCB(page=Pages.HOME).pack() ) - return builder.as_markup() + return builder.adjust(1).as_markup() - def menu(self, servers: list[Server]): + def menu(self, key: str) -> InlineKeyboardMarkup: builder = InlineKeyboardBuilder() + hashkey = EnvFile.to_hash(key) + builder.button( + text=KeyboardText.SERVERS, + callback_data=PageCB( + key=hashkey, page=Pages.SERVER, action=Actions.LIST + ).pack(), + ) + builder.row( + InlineKeyboardButton( + text=KeyboardText.HOMES, callback_data=PageCB(page=Pages.HOME).pack() + ) + ) + return builder.adjust(1).as_markup() + def servers(self, key: str, servers: list[Server] = None) -> InlineKeyboardMarkup: + builder = InlineKeyboardBuilder() + hashkey = EnvFile.to_hash(key) for server in servers: emoji = {"starting": "🟡", "stopping": "🔴", "running": "🟢", "off": "🔴"} status_emoji = emoji.get(server.status, "⚪") builder.button( text=f"{status_emoji} {server.name} ({server.public_net.ipv4.ip if server.public_net.ipv4 else 'No IPv4'})", - callback_data=ServerAction( - action=Actions.INFO, server_id=server.id + callback_data=PageCB( + key=hashkey, + page=Pages.SERVER, + action=Actions.INFO, + server_id=server.id, ).pack(), ) - builder.adjust(2) + builder.adjust(1) builder.row( InlineKeyboardButton( - text=KeyboardText.UPDATE, - callback_data=ServerList(action=Actions.HOME).pack(), + text=KeyboardText.UPDATE_SERVER, + callback_data=PageCB( + key=hashkey, page=Pages.SERVER, action=Actions.LIST + ).pack(), ), InlineKeyboardButton( - text=KeyboardText.CREATE, callback_data=LocationTypeSelect().pack() + text=KeyboardText.CREATE, + callback_data=PageCB( + key=hashkey, page=Pages.SERVER, action=Actions.CREATE + ).pack(), ), ) + builder.row( + InlineKeyboardButton( + text=KeyboardText.HOMES, callback_data=PageCB(page=Pages.HOME).pack() + ) + ) return builder.as_markup() - def edit_menu(self, serverid: int): + def edit_server(self, key: str, serverid: int): builder = InlineKeyboardBuilder() - + hashkey = EnvFile.to_hash(key) actions = { - Actions.POWER_ON: KeyboardText.POWER_ON, - Actions.POWER_OFF: KeyboardText.POWER_OFF, - Actions.REBOOT: KeyboardText.REBOOT, - Actions.RESET_PASSWORD: KeyboardText.RESET_PASSWORD, - Actions.DELETE: KeyboardText.DELETE, - Actions.REBUILD: KeyboardText.REBUILD, - Actions.UPDATE: KeyboardText.UPDATE_SERVER, - Actions.RESET: KeyboardText.RESET, + ServerUpdate.POWER_ON: KeyboardText.POWER_ON, + ServerUpdate.POWER_OFF: KeyboardText.POWER_OFF, + ServerUpdate.REBOOT: KeyboardText.REBOOT, + ServerUpdate.RESET_PASSWORD: KeyboardText.RESET_PASSWORD, + ServerUpdate.DELETE: KeyboardText.DELETE, + ServerUpdate.REBUILD: KeyboardText.REBUILD, + ServerUpdate.RESET: KeyboardText.RESET, + ServerUpdate.UPDATE: KeyboardText.UPDATE_SERVER, } for action, text in actions.items(): builder.button( text=text, - callback_data=ServerAction(action=action, server_id=serverid).pack(), + callback_data=PageCB( + key=hashkey, page=Pages.SERVER, action=action, server_id=serverid + ).pack(), ) builder.button( - text=KeyboardText.BACK, callback_data=ServerList(action=Actions.HOME).pack() + text=KeyboardText.SERVERS, + callback_data=PageCB( + key=hashkey, page=Pages.SERVER, action=Actions.LIST + ).pack(), ) builder.adjust(2) return builder.as_markup() - def confirm(self, action: str, serverid: int, imageid: int = 0): + def back_home(self) -> InlineKeyboardMarkup: builder = InlineKeyboardBuilder() + builder.button( + text=KeyboardText.UPDATE, callback_data=PageCB(page=Pages.HOME).pack() + ) + return builder.adjust(1).as_markup() + def confirm(self, key: str, action: str, serverid: int, imageid: int = 0): + builder = InlineKeyboardBuilder() + hashkey = EnvFile.to_hash(key) builder.button( text=KeyboardText.CONFIRM, - callback_data=ServerAction( + callback_data=PageCB( + key=hashkey, + page=Pages.SERVER, action=action, server_id=serverid, - confirm=True, image_id=imageid, + confirm=True, ).pack(), ) builder.button( text=KeyboardText.CANCEL, - callback_data=ServerAction(action=Actions.INFO, server_id=serverid).pack(), + callback_data=PageCB( + key=hashkey, + page=Pages.SERVER, + action=Actions.INFO, + server_id=serverid, + image_id=imageid, + ).pack(), ) builder.adjust(2) return builder.as_markup() - def rebuild(self, images: list[Image], serverid: int): + def rebuild(self, key: str, images: list[Image], serverid: int): builder = InlineKeyboardBuilder() - + hashkey = EnvFile.to_hash(key) latest_images: dict[str, Image] = {} for image in images: if not image.name: @@ -116,61 +168,83 @@ def rebuild(self, images: list[Image], serverid: int): for image in image_list: builder.button( text=image.name, - callback_data=ServerAction( - action=Actions.REBUILD, + callback_data=PageCB( + key=hashkey, + page=Pages.SERVER, + action=ServerUpdate.REBUILD, server_id=serverid, confirm=False, image_id=image.id, ).pack(), ) - builder.button( - text=KeyboardText.CANCEL, - callback_data=ServerAction(action=Actions.INFO, server_id=serverid).pack(), - ) - builder.adjust(2) + + builder.row( + InlineKeyboardButton( + text=KeyboardText.CANCEL, + callback_data=PageCB( + key=hashkey, + page=Pages.SERVER, + action=Actions.INFO, + server_id=serverid, + ).pack(), + ) + ) return builder.as_markup() - def location_types(self, location_types: list[Datacenter]): + def location_types(self, key: str, location_types: list[Datacenter]): builder = InlineKeyboardBuilder() - + hashkey = EnvFile.to_hash(key) for ser in location_types: builder.button( text=f"{ser.location.country}, {ser.location.city}", - callback_data=LocationTypeSelect( - location=ser.id, is_select=True + callback_data=SelectCB( + key=hashkey, + page=Pages.SERVER, + action=Actions.CREATE, + datavalue=ser.id, + datatype=ServerCreate.LOCATION, ).pack(), ) builder.button( - text=KeyboardText.CANCEL, - callback_data=ServerList(action=Actions.HOME).pack(), + text=KeyboardText.SERVERS, + callback_data=PageCB( + key=hashkey, page=Pages.SERVER, action=Actions.LIST + ).pack(), ) builder.adjust(1) return builder.as_markup() - def server_types(self, server_types: list[ServerType]): + def server_types(self, key: str, server_types: list[ServerType]): builder = InlineKeyboardBuilder() - + hashkey = EnvFile.to_hash(key) for ser in server_types: builder.button( text=f"[{ser.architecture}] {ser.name} C:{ser.cores} M:{ser.memory} P:{ser.prices[0]['price_monthly']['net'][:5]}", - callback_data=ServerTypeSelect(server=ser.id, is_select=True).pack(), + callback_data=SelectCB( + key=hashkey, + page=Pages.SERVER, + action=Actions.CREATE, + datavalue=ser.id, + datatype=ServerCreate.SERVER, + ).pack(), ) builder.button( - text=KeyboardText.CANCEL, - callback_data=ServerList(action=Actions.HOME).pack(), + text=KeyboardText.SERVERS, + callback_data=PageCB( + key=hashkey, page=Pages.SERVER, action=Actions.LIST + ).pack(), ) - builder.adjust(1) return builder.as_markup() - def image_types(self, images: list[Image]): + def image_types(self, key: str, images: list[Image]): builder = InlineKeyboardBuilder() - + hashkey = EnvFile.to_hash(key) latest_images: dict[str, Image] = {} for image in images: if not image.name: @@ -186,13 +260,24 @@ def image_types(self, images: list[Image]): for image in image_list: builder.button( text=image.name, - callback_data=ImageTypeSelect(image=image.id, is_select=True).pack(), + callback_data=SelectCB( + key=hashkey, + page=Pages.SERVER, + action=Actions.CREATE, + datavalue=image.id, + datatype=ServerCreate.IMAGE, + ).pack(), ) - builder.button( - text=KeyboardText.CANCEL, - callback_data=ServerList(action=Actions.HOME).pack(), + builder.adjust(2) + builder.row( + InlineKeyboardButton( + text=KeyboardText.HOMES, + callback_data=PageCB( + key=hashkey, page=Pages.SERVER, action=Actions.LIST + ).pack(), + ), + width=1, ) - builder.adjust(2) return builder.as_markup() diff --git a/language/keyboard.py b/language/keyboard.py index b948af8..a85b04a 100644 --- a/language/keyboard.py +++ b/language/keyboard.py @@ -8,7 +8,7 @@ class KeyboardTextsFile(BaseSettings): env_file=".env", extra="ignore" ) - UPDATE: str = "🔄 Update Servers" + UPDATE: str = "🔄 Update" POWER_ON: str = "🟢 Power On" POWER_OFF: str = "🔴 Power Off" REBOOT: str = "🔄 Reboot" @@ -22,3 +22,5 @@ class KeyboardTextsFile(BaseSettings): RESET: str = "🔄 Reset" HETZNER: str = "🟥 Hetzner" CREATE: str = "➕ Create" + SERVERS: str = "☁️ Servers" + HOMES: str = "🏛️ Home" diff --git a/language/message.py b/language/message.py index a023997..a2c54bc 100644 --- a/language/message.py +++ b/language/message.py @@ -9,6 +9,7 @@ class MessageTextsFile(BaseSettings): ) START: str = "👋 Welcome to ServerManagerBot\nDevelop and Design by @ErfJabs" + MENU: str = "🗃️ Your Account Menu:" SERVER_LIST: str = "🖥️ Here are your servers:" IS_UPDATED: str = "✅ is updated!" SERVER_INFO: str = ( @@ -24,7 +25,7 @@ class MessageTextsFile(BaseSettings): "📅 Created: {created} [{created_day} days ago]\n" "🔑 Password: {password}" ) - NOT_FOUND: str = "❗ not found!" + NOT_FOUND: str = "❗ Not Found! (check logs)" TRY_AGAIN: str = "⚠️ Oops! An error occurred, please try again..." CHECK_LOGS: str = "⚠️ Oops! An error occurred, please check the logs." CONFIRM_ACTION: str = "Are you sure you want to {action} this server?" diff --git a/middlewares/auth.py b/middlewares/auth.py index 085183a..38d6f06 100644 --- a/middlewares/auth.py +++ b/middlewares/auth.py @@ -17,6 +17,8 @@ async def __call__( user = event.message.from_user elif event.callback_query: user = event.callback_query.from_user + key = event.callback_query.data.split(sep=":", maxsplit=2)[1] + data["key"] = EnvFile.from_hash(hashkey=key) if key != "KEY" else None elif event.inline_query: user = event.inline_query.from_user diff --git a/routers/__init__.py b/routers/__init__.py index 616a97f..ec6c115 100644 --- a/routers/__init__.py +++ b/routers/__init__.py @@ -1,15 +1,14 @@ from aiogram import Router -from . import base, data, edit, create +from . import base +from .server import setup_server_routers -__all__ = ["setup_routers", "base", "data", "edit", "create"] +__all__ = ["setup_routers", "base"] def setup_routers() -> Router: router = Router() + router.include_router(setup_server_routers()) router.include_router(base.router) - router.include_router(data.router) - router.include_router(edit.router) - router.include_router(create.router) return router diff --git a/routers/base.py b/routers/base.py index f84dd44..7c1cc8d 100644 --- a/routers/base.py +++ b/routers/base.py @@ -3,33 +3,34 @@ from aiogram.types import Message, CallbackQuery from language import MessageText -from keys import Keyboards, ServerList, Actions -from api import HetznerAPI +from keys import Keyboards, PageCB, Pages +from config import EnvFile router = Router(name="start") @router.message(CommandStart(ignore_case=True)) async def start(message: Message): - servers = await HetznerAPI.get_servers() + await message.answer( + MessageText.START, reply_markup=Keyboards.home(EnvFile.HETZNER_API_KEYS) + ) - if not servers: - await message.answer(MessageText.NOT_FOUND) - return - await message.answer(MessageText.START, reply_markup=Keyboards.menu(servers)) - - -@router.callback_query(ServerList.filter(F.action == Actions.HOME)) -async def update_server_list(callback: CallbackQuery): - servers = await HetznerAPI.get_servers() +@router.callback_query(PageCB.filter(F.page == Pages.HOME)) +async def update(callback: CallbackQuery): + try: + await callback.message.edit_text( + MessageText.START, reply_markup=Keyboards.home(EnvFile.HETZNER_API_KEYS) + ) + except exceptions.TelegramAPIError: + await callback.answer(MessageText.IS_UPDATED) - if not servers: - return await callback.answer(MessageText.NOT_FOUND) +@router.callback_query(PageCB.filter(F.page == Pages.MENU)) +async def menu(callback: CallbackQuery, key: str | None): try: await callback.message.edit_text( - MessageText.START, reply_markup=Keyboards.menu(servers) + MessageText.SERVER_LIST, reply_markup=Keyboards.menu(key) ) except exceptions.TelegramAPIError: await callback.answer(MessageText.IS_UPDATED) diff --git a/routers/edit.py b/routers/edit.py deleted file mode 100644 index ae35e80..0000000 --- a/routers/edit.py +++ /dev/null @@ -1,123 +0,0 @@ -from aiogram import Router, F -from aiogram.types import CallbackQuery - -from hcloud.servers.client import BoundServer - -from .data import server_data -from keys import ServerAction, Keyboards, Actions -from language import MessageText -from api import HetznerAPI - -router = Router(name="edit") - - -@router.callback_query( - ServerAction.filter( - F.action.in_( - { - Actions.POWER_ON, - Actions.POWER_OFF, - Actions.REBOOT, - Actions.RESET_PASSWORD, - Actions.DELETE, - Actions.REBUILD, - Actions.RESET, - } - ) - ) -) -async def confirm_server_action(callback: CallbackQuery, callback_data: ServerAction): - if ( - not callback_data.confirm - and callback_data.action == Actions.REBUILD - and callback_data.image_id == 0 - ): - server = await HetznerAPI.get_server(callback_data.server_id) - images = await HetznerAPI.get_images(server.server_type.architecture) - if not images: - return await callback.answer(MessageText.CHECK_LOGS) - - await callback.message.edit_text( - text=MessageText.IMAGE_LIST, - reply_markup=Keyboards.rebuild( - images=images, serverid=callback_data.server_id - ), - ) - return - - # Confirmation for actions - if not callback_data.confirm: - await callback.message.edit_text( - text=MessageText.CONFIRM_ACTION.format(action=callback_data.action), - reply_markup=Keyboards.confirm( - callback_data.action, - callback_data.server_id, - imageid=callback_data.image_id, - ), - ) - return - - await callback.message.edit_text(MessageText.WAIT) - - # Fetch server - server = await HetznerAPI.get_server(callback_data.server_id) - if not server: - return await callback.answer(MessageText.CHECK_LOGS) - - # Execute action based on callback_data.action - action_result = await execute_server_action(callback_data, server, callback) - - if action_result is not None: - await callback.answer(action_result) - - if callback_data.action == Actions.RESET_PASSWORD: - return - - # Post-action handling - if callback_data.action == Actions.DELETE: - await update_server_list_ui(callback) - else: - await server_data( - callback, - ServerAction(action=Actions.INFO, server_id=callback_data.server_id), - ) - - -async def execute_server_action( - callback_data: ServerAction, server: BoundServer, callback: CallbackQuery -): - """Helper function to execute server actions based on the action type.""" - if callback_data.action == Actions.REBUILD: - if not callback_data.image_id: - return "Image ID not found" - result = await HetznerAPI.rebuild_server(server, callback_data.image_id) - if not result: - return MessageText.CHECK_LOGS - return "Rebuilding server initiated" - - action_method = getattr(HetznerAPI, callback_data.action) - result = await action_method(server) - - if not result: - return MessageText.CHECK_LOGS - - if callback_data.action == Actions.RESET_PASSWORD: - new_password = result - return await server_data( - callback, - ServerAction(action=Actions.INFO, server_id=callback_data.server_id), - server_password=new_password, - ) - - return f"{callback_data.action.capitalize()} action initiated" - - -async def update_server_list_ui(callback: CallbackQuery): - """Helper function to update the server list after an action like delete.""" - servers = await HetznerAPI.get_servers() - if not servers: - await callback.answer(MessageText.CHECK_LOGS) - else: - await callback.message.edit_text( - MessageText.START, reply_markup=Keyboards.menu(servers) - ) diff --git a/routers/server/__init__.py b/routers/server/__init__.py new file mode 100644 index 0000000..c4412bd --- /dev/null +++ b/routers/server/__init__.py @@ -0,0 +1,16 @@ +from aiogram import Router + +from . import menu, data, create, edit + +__all__ = ["setup_server_routers", "menu", "data", "create", "edit"] + + +def setup_server_routers() -> Router: + router = Router() + + router.include_router(data.router) + router.include_router(menu.router) + router.include_router(edit.router) + router.include_router(create.router) + + return router diff --git a/routers/server/create.py b/routers/server/create.py new file mode 100644 index 0000000..2693cd9 --- /dev/null +++ b/routers/server/create.py @@ -0,0 +1,106 @@ +import secrets + +from aiogram import Router, F +from aiogram.types import CallbackQuery +from aiogram.fsm.context import FSMContext + +from keys import Keyboards, SelectCB, Pages, Actions, PageCB, ServerCreate +from language import MessageText +from api import HetznerManager + +router = Router(name="server_create") + + +@router.callback_query( + PageCB.filter((F.page.is_(Pages.SERVER)) & (F.action.is_(Actions.CREATE))) +) +async def show_location_types(callback: CallbackQuery, key: str | None): + location_types = await HetznerManager.get_datacenters(key) + + if not location_types: + return await callback.answer(MessageText.NOT_FOUND) + + return await callback.message.edit_text( + text=MessageText.SELECT_LOCATION_TYPE, + reply_markup=Keyboards.location_types(key, location_types), + ) + + +@router.callback_query( + SelectCB.filter( + (F.page.is_(Pages.SERVER)) + & (F.action.is_(Actions.CREATE)) + & (F.datatype.is_(ServerCreate.LOCATION)) + ) +) +async def show_server_types( + callback: CallbackQuery, state: FSMContext, callback_data: SelectCB, key: str | None +): + await state.update_data(location=callback_data.datavalue) + + location_server_types = await HetznerManager.get_datacenter( + key, callback_data.datavalue + ) + + if not location_server_types: + return await callback.answer(MessageText.NOT_FOUND) + + return await callback.message.edit_text( + text=MessageText.SELECT_SERVER_TYPE, + reply_markup=Keyboards.server_types( + key, location_server_types.server_types.available + ), + ) + + +@router.callback_query( + SelectCB.filter( + (F.page.is_(Pages.SERVER)) + & (F.action.is_(Actions.CREATE)) + & (F.datatype.is_(ServerCreate.SERVER)) + ) +) +async def select_server_types( + callback: CallbackQuery, state: FSMContext, callback_data: SelectCB, key: str | None +): + await state.update_data(server=callback_data.datavalue) + + server_select = await HetznerManager.get_server_type( + key, int(callback_data.datavalue) + ) + image_types = await HetznerManager.get_images(key, arch=server_select.architecture) + + if not image_types: + return await callback.answer(MessageText.NOT_FOUND) + + return await callback.message.edit_text( + text=MessageText.SELECT_IMAGE_TYPE, + reply_markup=Keyboards.image_types(key, image_types), + ) + + +@router.callback_query( + SelectCB.filter( + (F.page.is_(Pages.SERVER)) + & (F.action.is_(Actions.CREATE)) + & (F.datatype.is_(ServerCreate.IMAGE)) + ) +) +async def select_image_type( + callback: CallbackQuery, state: FSMContext, callback_data: SelectCB, key: str | None +): + data = await state.get_data() + + server_create = await HetznerManager.create_server( + key=key, + name=secrets.token_hex(2), + server_type=await HetznerManager.get_server_type(key, int(data["server"])), + image=await HetznerManager.get_image(key, int(callback_data.datavalue)), + ) + + if not server_create: + return await callback.answer(MessageText.CHECK_LOGS) + + return await callback.message.edit_text( + text=MessageText.SERVER_CREATED, reply_markup=Keyboards.back_home() + ) diff --git a/routers/data.py b/routers/server/data.py similarity index 71% rename from routers/data.py rename to routers/server/data.py index dc98d26..ff88919 100644 --- a/routers/data.py +++ b/routers/server/data.py @@ -1,25 +1,29 @@ from datetime import datetime, timezone - from aiogram import Router, F, exceptions from aiogram.types import CallbackQuery -from keys import ServerAction, Keyboards, Actions -from api import HetznerAPI + +from api import HetznerManager +from keys import PageCB, Pages, Actions, Keyboards, ServerUpdate from language import MessageText -router = Router(name="data") +router = Router(name="server_menu") @router.callback_query( - ServerAction.filter(F.action.in_({Actions.INFO, Actions.UPDATE})) + PageCB.filter( + (F.page.is_(Pages.SERVER)) & (F.action.in_([Actions.INFO, ServerUpdate.UPDATE])) + ) ) async def server_data( - callback: CallbackQuery, callback_data: ServerAction, server_password: str = None + callback: CallbackQuery, + callback_data: PageCB, + key: str | None, + server_password: str | None = None, ): - server = await HetznerAPI.get_server(callback_data.server_id) - + server = await HetznerManager.get_server(key, callback_data.server_id) try: emoji = {"starting": "🟡", "stopping": "🔴", "running": "🟢", "off": "🔴"} - status_emoji = emoji.get(server.status, "⚪") + status_emoji = emoji.get(server.status, "⚪") if server.status else "⚪" await callback.message.edit_text( text=MessageText.SERVER_INFO.format( name=server.name, @@ -42,7 +46,7 @@ async def server_data( 3, ), ), - reply_markup=Keyboards.edit_menu(server.id), + reply_markup=Keyboards.edit_server(key, callback_data.server_id), ) except exceptions.TelegramAPIError: await callback.answer(MessageText.IS_UPDATED) diff --git a/routers/server/edit.py b/routers/server/edit.py new file mode 100644 index 0000000..b670045 --- /dev/null +++ b/routers/server/edit.py @@ -0,0 +1,140 @@ +from aiogram import Router, F +from aiogram.types import CallbackQuery + +from .data import server_data +from keys import Keyboards, Actions, PageCB, Pages, ServerUpdate +from language import MessageText +from api import HetznerManager +from config import EnvFile + +router = Router(name="server_edit") + + +@router.callback_query( + PageCB.filter( + ( + F.action.in_( + { + ServerUpdate.POWER_ON, + ServerUpdate.POWER_OFF, + ServerUpdate.REBOOT, + ServerUpdate.RESET_PASSWORD, + ServerUpdate.DELETE, + ServerUpdate.REBUILD, + ServerUpdate.RESET, + } + ) + ) + & (F.page.is_(Pages.SERVER)) + ) +) +async def confirm_server_action( + callback: CallbackQuery, callback_data: PageCB, key: str +): + if ( + not callback_data.confirm + and callback_data.action == ServerUpdate.REBUILD + and callback_data.image_id is None + ): + server = await HetznerManager.get_server(key, callback_data.server_id) + images = await HetznerManager.get_images(key, server.server_type.architecture) + if not images: + return await callback.answer(MessageText.CHECK_LOGS) + + await callback.message.edit_text( + text=MessageText.IMAGE_LIST, + reply_markup=Keyboards.rebuild( + key=key, images=images, serverid=callback_data.server_id + ), + ) + return + + # Confirmation for actions + if not callback_data.confirm: + await callback.message.edit_text( + text=MessageText.CONFIRM_ACTION.format( + action=(callback_data.action.value).replace("_", " ") + ), + reply_markup=Keyboards.confirm( + key, + callback_data.action, + callback_data.server_id, + imageid=callback_data.image_id, + ), + ) + return + + await callback.message.edit_text(MessageText.WAIT) + + # Fetch server + server = await HetznerManager.get_server(key, callback_data.server_id) + if not server: + return await callback.answer(MessageText.CHECK_LOGS) + + # Execute action based on callback_data.action + action_result = await execute_server_action(callback_data, server, callback, key) + + if action_result is not None: + await callback.answer(action_result) + + if callback_data.action == ServerUpdate.RESET_PASSWORD: + return + + # Post-action handling + if callback_data.action == ServerUpdate.DELETE: + await update_server_list_ui(callback, key) + else: + await server_data( + callback, + PageCB( + key=EnvFile.to_hash(key), + page=Pages.SERVER, + action=Actions.INFO, + server_id=callback_data.server_id, + ), + key, + ) + + +async def execute_server_action( + callback_data: PageCB, server, callback: CallbackQuery, key: str +): + """Helper function to execute server actions based on the action type.""" + if callback_data.action == ServerUpdate.REBUILD: + if not callback_data.image_id: + return "Image ID not found" + result = await HetznerManager.rebuild_server( + key, server, callback_data.image_id + ) + if not result: + return MessageText.CHECK_LOGS + return "Rebuilding server initiated" + + action_method = getattr(HetznerManager, callback_data.action) + result = await action_method(key, server) + + if not result: + return MessageText.CHECK_LOGS + + if callback_data.action == ServerUpdate.RESET_PASSWORD: + new_password = result + return await server_data( + callback, + PageCB( + key=EnvFile.to_hash(key), + page=Pages.SERVER, + action=Actions.INFO, + server_id=callback_data.server_id, + ), + key, + new_password, + ) + + return f"{callback_data.action.capitalize()} action initiated" + + +async def update_server_list_ui(callback: CallbackQuery, key: str): + """Helper function to update the server list after an action like delete.""" + await callback.message.edit_text( + MessageText.START, reply_markup=Keyboards.menu(key) + ) diff --git a/routers/server/menu.py b/routers/server/menu.py new file mode 100644 index 0000000..248e41d --- /dev/null +++ b/routers/server/menu.py @@ -0,0 +1,22 @@ +from aiogram import Router, F, exceptions +from aiogram.types import CallbackQuery + +from api import HetznerManager +from keys import PageCB, Pages, Actions, Keyboards +from language import MessageText + +router = Router(name="server_menu") + + +@router.callback_query( + PageCB.filter((F.page.is_(Pages.SERVER)) & (F.action.is_(Actions.LIST))) +) +async def server_actions(callback: CallbackQuery, key: str | None): + servers = await HetznerManager.get_servers(key) + try: + return await callback.message.edit_text( + text=MessageText.SERVER_LIST, + reply_markup=Keyboards.servers(key, servers or []), + ) + except exceptions.TelegramAPIError: + await callback.answer(MessageText.IS_UPDATED)