# open source телеграм-бота на Python (аве Штольмане!)
# vibecoded by Claude
import asyncio
import logging
import os
import random
import re
from aiogram import Bot, Dispatcher, F
from aiogram.enums import ChatAction
from aiogram.types import Message
from dotenv import load_dotenv
from openai import AsyncOpenAI
load_dotenv("/opt/bot/.env")
BOT_USERNAME = "" # заполнится при старте
MAX_LENGTH = 4096
BOT_TOKEN = os.environ["BOT_TOKEN"]
YANDEX_API_KEY = os.environ["YANDEX_API_KEY"]
YANDEX_FOLDER_ID = os.environ["YANDEX_FOLDER_ID"]
# Полный URI модели — обязательно с folder_id
default_model = f"gpt://{YANDEX_FOLDER_ID}/deepseek-v4.1-flash"
YANDEX_MODEL = os.getenv("YANDEX_MODEL", default_model)
# ID группы обсуждений. Если задан, то бот отвечает только в ней.
GROUP_ID = os.getenv("GROUP_ID")
SYSTEM_PROMPT = (
"Будь как проработанный монах-технарь: шаришь за науку, мемы и нижний интернет, но без токсичности. Напиши тёплый, небанальный коммент к тексту ниже (эмодзи и разметка Telegram приветствуются, интернет — по необходимости). Если текст непонятен или обрезан — кинь шутку или коан. "
)
logging.basicConfig(level=logging.INFO)
bot = Bot(BOT_TOKEN)
dp = Dispatcher()
llm = AsyncOpenAI(
api_key=YANDEX_API_KEY,
base_url="https://ai.api.cloud.yandex.net/v1",
project=YANDEX_FOLDER_ID,
)
async def ask_llm(post_text: str) -> str:
prompt = (
SYSTEM_PROMPT
+ post_text
)
response = await llm.responses.create(
model=YANDEX_MODEL,
input=prompt,
tools=[{"type": "web_search"}],
temperature=0.72,
max_output_tokens=7000,
)
return (response.output_text or "").strip()
# Если написать боту команду, то он ответит через ИИ.
@dp.message(F.chat.type == "private", F.text.strip().lower().startswith("/deepseek"))
async def on_deepseek(message: Message):
query = message.text.strip()[len("/deepseek"):].strip()
if not query:
return
try:
answer = await ask_llm(query)
except Exception:
logging.exception("Ошибка при обращении к модели.")
answer = "Ошибка при обращении к модели."
if answer:
for i in range(0, len(answer), MAX_LENGTH):
chunk = answer[i:i + MAX_LENGTH]
await message.reply(chunk)
# Если написать боту любое сообщение, то активизируется гача-игра:
def roll_fish() -> str:
r = random.random()
if r < 0.5:
return "⭐ " + random.choice(JUNK)
if r < 0.75:
return "⭐⭐ " + random.choice(NORMAL)
if r < 0.89:
return "⭐⭐⭐ " + random.choice(RARE)
if r < 0.97:
return "⭐⭐⭐⭐ " + random.choice(LEGEND)
return "⭐⭐⭐⭐⭐ " + random.choice(EPIC)
@dp.message(F.chat.type == "private", F.text.strip().lower().in_({"Порыбачить", "Рыба", "/fish"}))
async def on_fish(message: Message):
await message.answer(f"{roll_fish()}\n\nВетра Тейвата продолжают обдувать вас. Где-то вдали играет флейта хиличурла.. (/fish)")
@dp.message(F.chat.type == "private")
async def on_private_any(message: Message):
await message.answer("Вы стоите у реки. 'Может, порыбачить?..', думаете вы.. Ветра Тейвата обдувают вас. (/fish)")
# Реакция на посты канала, автоматически пересланные в группу:
@dp.message(F.chat.type.in_({"group", "supergroup"}), F.is_automatic_forward)
async def on_channel_post(message: Message):
if not message.is_automatic_forward:
return
if GROUP_ID and str(message.chat.id) != GROUP_ID:
return
post_text = message.text or message.caption
if not post_text:
return # пост без текста (например, только картинка)
try:
answer = await ask_llm(post_text)
except Exception:
logging.exception("Ошибка при обращении к модели")
return
if answer:
for i in range(0, len(answer), MAX_LENGTH):
chunk = answer[i:i + MAX_LENGTH]
await message.reply(chunk)
async def main():
global BOT_USERNAME
BOT_USERNAME = (await bot.get_me()).username
await dp.start_polling(bot)
if __name__ == "__main__":
asyncio.run(main())
