From f5003e3476c770ff2c2ca2484220512f02704e41 Mon Sep 17 00:00:00 2001 From: Ruslan-Isaev Date: Tue, 25 Feb 2025 07:21:43 +0300 Subject: [PATCH] Add files via upload --- S3.py | 187 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 187 insertions(+) create mode 100644 S3.py diff --git a/S3.py b/S3.py new file mode 100644 index 0000000..968a5e1 --- /dev/null +++ b/S3.py @@ -0,0 +1,187 @@ +version = (1, 0, 0) + +# meta developer: @RUIS_VlP + +import aioboto3 +import hashlib +import aiofiles +import os +from .. import loader, utils +import mimetypes + +async def s3_upload(url, bucket, filename, filepath, access_key, secret_key): + session = aioboto3.Session() + mime_type, _ = mimetypes.guess_type(filename) #вычисление метаданных, нужно для того что бы браузер корректно определял загруженный в хранилище файл + if mime_type is None: + mime_type = 'binary/octet-stream' + async with aiofiles.open(filename, 'rb') as f: + file_content = await f.read() + sha256_hash = hashlib.sha256(file_content).hexdigest() #вычисление хэша. обычно при загрузке он не нужен, но без него у меня бывают ошибки + + async with session.client("s3", endpoint_url=url, aws_access_key_id=access_key, aws_secret_access_key=secret_key) as s3: + await s3.upload_file(filename, bucket, f"{filepath}/{filename}", ExtraArgs={'ChecksumSHA256': sha256_hash, 'ContentType': mime_type}) + +async def s3_download(url, bucket, filename, filepath, access_key, secret_key): + session = aioboto3.Session() + async with session.client("s3", endpoint_url=url, aws_access_key_id=access_key, aws_secret_access_key=secret_key) as s3: + await s3.download_file(bucket, filename, f"{filepath}/{filename.split('/')[-1]}") + return f"{filepath}/{filename.split('/')[-1]}" + +async def s3_delete(url, bucket, filename, access_key, secret_key): + session = aioboto3.Session() + async with session.client("s3", endpoint_url=url, aws_access_key_id=access_key, aws_secret_access_key=secret_key) as s3: + await s3.delete_object(Bucket=bucket, Key=filename) + +async def s3_ls(url, bucket, filepath, access_key, secret_key): + session = aioboto3.Session() + async with session.client("s3", endpoint_url=url, aws_access_key_id=access_key, aws_secret_access_key=secret_key) as s3: + response = await s3.list_objects_v2(Bucket=bucket, Prefix=filepath) + return [obj["Key"] for obj in response.get("Contents", [])] #я сам не ебу что это, мне это ChatGPT сделал + + +@loader.tds +class S3Mod(loader.Module): + """Модуль для работы с S3 хранилищами""" + + strings = { + "name": "S3", + } + + def __init__(self): + self.config = loader.ModuleConfig( + loader.ConfigValue( + "url", + "None", + lambda: "Ссылка на ваше S3 хранилище", + validator=loader.validators.String(), + ), + loader.ConfigValue( + "bucketname", + "None", + lambda: "Имя bucket'а", + validator=loader.validators.String(), + ), + loader.ConfigValue( + "access_key", + "None", + lambda: "Ключ авторизации", + validator=loader.validators.Hidden(), + ), + loader.ConfigValue( + "secret_key", + "None", + lambda: "Секретный ключ", + validator=loader.validators.Hidden(), + ), + ) + + @loader.command() + async def S3upload(self, message): + """ - сохраняет файл в S3 хранилище""" + url = self.config["url"] or "None" + bucket = self.config["bucketname"] or "None" + access = self.config["access_key"] or "None" + secret = self.config["secret_key"] or "None" + if url == "None" or bucket == "None" or secret == "None" or access == "None": + await utils.answer(message, f"❌ Вы не настроили модуль! Укажите необходимые данные в config. Команда: {self.get_prefix()}config S3") + return + args = utils.get_args_raw(message) + if args: + filepath = args.split(" ")[0] + filepath = filepath[1:] if filepath.startswith('/') else filepath #удаление / из переменной, если она идет первым символом + else: + filepath = "FromHikka" + reply = await message.get_reply_message() + if reply and reply.media: + await utils.answer(message, "💿 Сохраняю файл на сервер...") + try: + filename = await message.client.download_media(reply.media) + await utils.answer(message, "☁️ Сохраняю файл в S3 хранилище...") + await s3_upload(url, bucket, filename, filepath, access, secret) + await utils.answer(message, "💿 Удаляю файл с сервера...") + os.remove(filename) + await utils.answer(message, f"✅ Успешно! Файл сохранен в директорию /{filepath} на вашем S3 хранилище") + except Exception as e: + await utils.answer(message, f"❌ Ошибка!\n\n{e}") + os.remove(filename) + else: + await utils.answer(message, "❌ Ошибка! Не найден ответ на сообщение или в ответном сообщении отсутствуют файлы.") + + + @loader.command() + async def S3LS(self, message): + """ - список файлов в S3 хранилище""" + url = self.config["url"] or "None" + bucket = self.config["bucketname"] or "None" + access = self.config["access_key"] or "None" + secret = self.config["secret_key"] or "None" + if url == "None" or bucket == "None" or secret == "None" or access == "None": + await utils.answer(message, f"❌ Вы не настроили модуль! Укажите необходимые данные в config. Команда: {self.get_prefix()}config S3") + return + args = utils.get_args_raw(message) + if args: + filepath = args + filepath = filepath[1:] if filepath.startswith('/') else filepath + else: + filepath = "" + try: + ls = await s3_ls(url, bucket, filepath, access, secret) + output = '\n'.join(['▪️' + item + '' for item in ls]) #превращение списка в человекочитаемый текст + await utils.answer(message, f"🗂 Список файлов и директорий в /{filepath}:\n\n{output}") + except Exception as e: + await utils.answer(message, f"❌ Ошибка!\n\n{e}") + + @loader.command() + async def S3delete(self, message): + """ - удаляет файл из S3 хрпнилища""" + url = self.config["url"] or "None" + bucket = self.config["bucketname"] or "None" + access = self.config["access_key"] or "None" + secret = self.config["secret_key"] or "None" + if url == "None" or bucket == "None" or secret == "None" or access == "None": + await utils.answer(message, f"❌ Вы не настроили модуль! Укажите необходимые данные в config. Команда: {self.get_prefix()}config S3") + return + args = utils.get_args_raw(message) + if args: + filepath = args + filepath = filepath[1:] if filepath.startswith('/') else filepath #удаление / из переменной, если она идет первым символом + else: + await utils.answer(message, "❌ Вы не указали файл для удаления!") + return + try: + await s3_delete(url, bucket, filepath, access, secret) + await utils.answer(message, f"✅ Файл {filepath} успешно удален!") + except Exception as e: + await utils.answer(message, f"❌ Ошибка!\n\n{e}") + + @loader.command() + async def S3download(self, message): + """ - скачивает файл из S3 хрпнилища и отправляет в Telegram""" + url = self.config["url"] or "None" + bucket = self.config["bucketname"] or "None" + access = self.config["access_key"] or "None" + secret = self.config["secret_key"] or "None" + if url == "None" or bucket == "None" or secret == "None" or access == "None": + await utils.answer(message, f"❌ Вы не настроили модуль! Укажите необходимые данные в config. Команда: {self.get_prefix()}config S3") + return + args = utils.get_args_raw(message) + if args: + filename = args + filename = filename[1:] if filename.startswith('/') else filename #удаление / из переменной, если она идет первым символом + else: + await utils.answer(message, "❌ Вы не указали файл для сохранения!") + return + try: + dl = await s3_download(url, bucket, filename, utils.get_base_dir(), access, secret) + await utils.answer_file(message, dl, caption=f"✅ Вот ваш файл /{filename}!") + os.remove(dl) + except Exception as e: + await utils.answer(message, f"❌ Ошибка!\n\n{e}") + + @loader.command() + async def s3config(self, message): + """- открыть конфигурацию модуля""" + name = "S3" + await self.allmodules.commands["config"]( + await utils.answer(message, f"{self.get_prefix()}config {name}") + ) \ No newline at end of file