from .. import loader, utils
import requests
import io
import aiohttp
import logging
logger = logging.getLogger(__name__)
# meta developer: @kmodules
# changelog: Фиксы, обновитесь обязательно
__version__ = (1, 5, 4)
version = __version__
@loader.tds
class KsenonGPTMod(loader.Module):
"""KsenonGPT module for text and image generation using KsenonAPI"""
strings = {
"name": "KsenonGPT",
"no_api_key": "⚠️ You have not set up the API key!\n\n🌐 The key is very easy to get, it's free.\n\n🔓 Bot: @ksenonapi_gettoken_bot",
"generating_text": "💬 Responding to your message...",
"text_generated": "💬 Request: {}\n\n💬 {}",
"generating_image": "🖼 Generating image...\n\n🤖 Model: {}\n🔗 Request: {}",
"image_generated": "🖼 Image generated!\n\n🤖 Model: {}\n🔗 Request: {}\n\n📥 Link: {}",
"error_blocked": "🔒 You have been blocked!\n\n🗑NSFW | politics etc. generation is prohibited.",
"error_occurred": "⚠️ An error occurred!\n\n💻 Model: {}\n📶 Server status, code: {}\n➡️ Error: {}",
"text_models": "🌐 Text models:\n\no1-preview\ngpt-4o\nclaude-3-5-sonnet\nsearchgpt (GPT + Internet)\nblackboxai-pro\nclaude-3-5-sonnet-20240620\nclaude-3-haiku-ddg\ngemini-1.5-pro-latest\nllama-3.1-405b\ngpt-3.5-turbo-202201\ngpt-4o-mini-ddg\ngpt-4o-2024-05-13\nmicrosoft/Phi-3.5-mini-instruct\nQwen/Qwen2.5-Coder-32B-Instruct\nQwen/QwQ-32B-Preview
\n\n✅️ We have 150+ models!\n🛡 https://t.me/ksenonapi_models",
"image_models": "🌐 Image models:\n\nflux-pro-mg\nflux-dev\nsd3-ultra\npixart-alpha
",
"no_args": "⚠️ No arguments provided!",
"update_available": "⚠️ KsenonGPT update available!\n\n🔼 New version: {}\n🔽 Current version: {}\n\n🌐 Changelog:\n🔔 {}\n\n✔️ Command to update:\n.dlmod https://raw.githubusercontent.com/TheKsenon/MyHikkaModules/refs/heads/main/ksenongpt.py",
"latest_version": "🥳 You have the latest version of KsenonGPT!\n\n⚠️Developers are making updates and fixes almost every day, check frequently!",
"select_model": "🤖 Select AI model:\n\n🔑 You can also specify a model directly by using .setmodel model_name",
"model_set": "🎯 Model has been set to: {}",
"invalid_model": "❌ Invalid model specified!",
"need_set_model": "🔥 Please set the model using .setmodel command!"
}
strings_ru = {
"name": "KsenonGPT",
"no_api_key": "⚠️ Вы не настроили API ключ!\n\n🌐 Ключ очень легко достать, он бесплатный.\n\n🔓 Бот: @ksenonapi_gettoken_bot",
"generating_text": "💬 Отвечаю на ваше сообщение...",
"text_generated": "💬 Запрос: {}\n\n💬 {}",
"generating_image": "🖼 Генерирую изображение...\n\n🤖 Модель: {}\n🔗 Запрос: {}",
"image_generated": "🖼 Изображение сгенерировано!\n\n🤖 Модель: {}\n🔗 Запрос: {}\n\n📥 Ссылка: {}",
"error_blocked": "🔒 Вы были заблокированы!\n\n🗑Запрещено генерировать NSFW | политика и т.д.",
"error_occurred": "⚠️ Произошла ошибка!\n\n💻 Модель: {}\n📶 Статус сервера, код: {}\n➡️ Ошибка: {}",
"text_models": "🌐 Текстовые модели:\n\no1-preview\ngpt-4o\nclaude-3-5-sonnet\nsearchgpt (GPT + Internet)\nblackboxai-pro\nclaude-3-5-sonnet-20240620\nclaude-3-haiku-ddg\ngemini-1.5-pro-latest\nllama-3.1-405b\ngpt-3.5-turbo-202201\ngpt-4o-mini-ddg\ngpt-4o-2024-05-13\nmicrosoft/Phi-3.5-mini-instruct\nQwen/Qwen2.5-Coder-32B-Instruct\nQwen/QwQ-32B-Preview
\n\n✅️ У нас 150+ моделей!\n🛡 https://t.me/ksenonapi_models",
"image_models": "🌐 Модели для изображений:\n\nflux-pro-mg\nflux-dev\nsd3-ultra\npixart-alpha
",
"no_args": "⚠️ Не указаны аргументы!",
"update_available": "⚠️ Доступно обновление KsenonGPT!\n\n🔼 Новая версия: {}\n🔽 Текущая версия: {}\n\n🌐 Список изменений:\n🔔 {}\n\n✔️ Команда для обновления:\n.dlmod https://raw.githubusercontent.com/TheKsenon/MyHikkaModules/refs/heads/main/ksenongpt.py",
"latest_version": "🥳 У вас последняя версия KsenonGPT!\n\n⚠️Разработчики делают обновления и исправления почти каждый день, проверяйте чаще!",
"select_model": "🤖 Выберите ИИ модель:\n\n🔑 Также вы можете указать конкретную модель, напишите .setmodel название_модели",
"model_set": "🎯 Модель установлена: {}",
"invalid_model": "❌ Указана неверная модель!",
"need_set_model": "🔥 Поставьте модель в .setmodel!"
}
def __init__(self):
self.config = loader.ModuleConfig(
loader.ConfigValue(
"api_key",
"",
"API key from @ksenonapi_gettoken_bot",
validator=loader.validators.Hidden(),
),
loader.ConfigValue(
"default_model",
"",
"Default AI model to use",
)
)
def _create_model_buttons(self):
buttons = []
models = [
("DeepSeek R1", "deepseek-r1"),
("GPT-4o", "gpt-4o"),
("Claude 3.5 Sonnet", "claude-3-5-sonnet"),
("SearchGPT", "searchgpt"),
("P1", "p1"),
("Gemini 2.0 Flash", "gemini-flash-2.0")
]
row = []
for i, (name, model_id) in enumerate(models):
row.append({"text": name, "callback": self._set_model, "args": (model_id,)})
if len(row) == 2:
buttons.append(row)
row = []
if row:
buttons.append(row)
return buttons
async def setmodelcmd(self, message):
"""Set default AI model for text generation"""
args = utils.get_args_raw(message)
if args:
self.config["default_model"] = args
await utils.answer(message, self.strings["model_set"].format(args))
return
await self.inline.form(
text=self.strings["select_model"],
message=message,
reply_markup=self._create_model_buttons())
async def _set_model(self, call, model):
self.config["default_model"] = model
await call.edit(
self.strings["model_set"].format(model),
reply_markup=None)
async def gentextcmd(self, message):
"""Generate text - .gentext """
if not self.config["api_key"]:
await utils.answer(message, self.strings["no_api_key"])
return
if not self.config["default_model"]:
await utils.answer(message, self.strings["need_set_model"])
return
args = utils.get_args_raw(message)
if not args:
await utils.answer(message, self.strings["no_args"])
return
prompt = args
model = self.config["default_model"]
msg = await utils.answer(message, self.strings["generating_text"])
headers = {
"Authorization": self.config["api_key"],
"Content-Type": "application/json"
}
data = {
"model": model,
"prompt": prompt
}
async with aiohttp.ClientSession() as session:
async with session.post(
"http://aeza.theksenon.pro/v1/api/text/generate",
headers=headers,
json=data
) as response:
result = await response.json()
if response.status != 200:
await utils.answer(
msg,
self.strings["error_occurred"].format(
model,
response.status,
result.get("error", "Unknown error")
)
)
return
if "error" in result:
if result["error"] == "Your token has been blocked":
await utils.answer(msg, self.strings["error_blocked"])
return
await utils.answer(
msg,
self.strings["error_occurred"].format(
model,
"N/A",
result["error"]
)
)
return
await utils.answer(
msg,
self.strings["text_generated"].format(
prompt,
result["response"]
)
)
async def genimgcmd(self, message):
"""Generate image - .genimg """
if not self.config["api_key"]:
await utils.answer(message, self.strings["no_api_key"])
return
args = utils.get_args_raw(message)
if not args:
await utils.answer(message, self.strings["no_args"])
return
try:
prompt, model = args.rsplit(maxsplit=1)
except ValueError:
prompt = args
model = "flux-pro-mg"
msg = await utils.answer(
message,
self.strings["generating_image"].format(model, prompt)
)
headers = {
"Authorization": self.config["api_key"],
"Content-Type": "application/json"
}
data = {
"model": model,
"prompt": prompt
}
async with aiohttp.ClientSession() as session:
async with session.post(
"http://aeza.theksenon.pro/v1/api/image/generate",
headers=headers,
json=data
) as response:
result = await response.json()
if response.status != 200:
await utils.answer(
msg,
self.strings["error_occurred"].format(
model,
response.status,
result.get("error", "Unknown error")
)
)
return
if "error" in result:
if result["error"] == "Your token has been blocked":
await utils.answer(msg, self.strings["error_blocked"])
return
await utils.answer(
msg,
self.strings["error_occurred"].format(
model,
"N/A",
result["error"]
)
)
return
image_url = result["url"]
async with session.get(image_url) as response:
if response.status != 200:
await utils.answer(
msg,
self.strings["error_occurred"].format(
model,
response.status,
"Failed to download image"
)
)
return
image_data = io.BytesIO(await response.read())
image_data.name = "image.png"
await self._client.send_file(
message.peer_id,
image_data,
caption=self.strings["image_generated"].format(
model,
prompt,
image_url
),
reply_to=message.reply_to_msg_id
)
if message.out:
await message.delete()
async def txtmodelscmd(self, message):
"""List of text models"""
await utils.answer(message, self.strings["text_models"])
async def imgmodelscmd(self, message):
"""List of image models"""
await utils.answer(message, self.strings["image_models"])
async def kupdatecmd(self, message):
"""Check for updates"""
async with aiohttp.ClientSession() as session:
async with session.get("https://raw.githubusercontent.com/TheKsenon/MyHikkaModules/refs/heads/main/ksenongpt.py") as response:
if response.status != 200:
return
content = await response.text()
try:
version_line = [line for line in content.split("\n") if "__version__" in line][0]
latest_version = tuple(map(int, version_line.split("(")[1].split(")")[0].split(",")))
if latest_version > version:
changelog = "New version available!"
await utils.answer(
message,
self.strings["update_available"].format(
".".join(map(str, latest_version)),
".".join(map(str, version)),
changelog
)
)
else:
await utils.answer(message, self.strings["latest_version"])
except:
logger.error("Failed to parse version from GitHub")