Files
telegramm_bot_ollama/app.py
smolkik_adm 2d20dcc473 init
2026-01-10 15:37:36 +07:00

126 lines
5.0 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
import os, logging, io, random
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup, LabeledPrice
from telegram.ext import Application, CommandHandler, CallbackQueryHandler, MessageHandler, PreCheckoutQueryHandler, filters, ContextTypes
from PIL import Image, ImageDraw, ImageFont
import requests
from payments import send_invoice
from dotenv import load_dotenv
load_dotenv()
TOKEN = os.getenv('TELEGRAM_TOKEN')
OLLAMA_URL = os.getenv('OLLAMA_URL', 'http://localhost:11434')
BALANCES = {} # В проде → Redis
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
MEME_TEMPLATES = ['drake.jpg', 'this-is-fine.jpg', 'distracted-bf.jpg']
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
keyboard = [
[InlineKeyboardButton("🎨 Создать мем (25⭐)", callback_data='buy_meme')],
[InlineKeyboardButton("💰 Баланс", callback_data='balance')]
]
reply_markup = InlineKeyboardMarkup(keyboard)
await update.message.reply_text(
'😺 <b>IT Meme Pro Bot</b>\n\n'
'Генерирую IT-мемы за <b>25 Telegram Stars</b>!\n'
'Коты в проде, Kubernetes фейлы, DevOps драма\n\n'
'Нажмите кнопку!',
reply_markup=reply_markup, parse_mode='HTML')
async def generate_meme(prompt: str):
# Ollama текст
url = f"{OLLAMA_URL}/api/generate"
data = {
"model": "qwen2.5:3b",
"prompt": f"""Создай короткий смешной текст IT-мема (макс 12 слов)
на тему: "{prompt}". Стиль: коты в продакшене, devops-проблемы. Только текст!""",
"stream": False
}
resp = requests.post(url, json=data, timeout=30).json()
meme_text = resp['response'].strip()
# Pillow изображение
template = Image.open(f'templates/{random.choice(MEME_TEMPLATES)}')
draw = ImageDraw.Draw(template)
try:
font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", 45)
except:
font = ImageFont.load_default()
# Текст с обводкой
bbox = draw.textbbox((0, 0), meme_text, font=font)
width, height = bbox[2] - bbox[0], bbox[3] - bbox[1]
x, y = (template.width - width) // 2, template.height - height - 60
# Обводка
for dx, dy in [(-2,0),(2,0),(0,-2),(0,2)]:
draw.text((x+dx, y+dy), meme_text, font=font, fill='black')
draw.text((x, y), meme_text, font=font, fill='white')
bio = io.BytesIO()
template.save(bio, 'PNG')
bio.seek(0)
return bio, meme_text
async def button_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
query = update.callback_query
await query.answer()
if query.data == 'balance':
bal = BALANCES.get(query.from_user.id, 0)
await query.edit_message_text(f'💰 Баланс: <b>{bal} мемов</b>', parse_mode='HTML')
elif query.data == 'buy_meme':
await send_invoice(update.callback_query.message, context)
async def precheckout_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
query = update.pre_checkout_query
await query.answer(ok=True)
async def successful_payment_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
user_id = update.effective_user.id
BALANCES[user_id] = BALANCES.get(user_id, 0) + 1 # +1 мем
await update.message.reply_text(
'✅ <b>Оплата прошла! Баланс: 1 мем</b>\n\n'
'🎨 Отправьте тему мема:\n"kubernetes pod pending"\n"nfs mount failed"',
parse_mode='HTML'
)
async def handle_text(update: Update, context: ContextTypes.DEFAULT_TYPE):
user_id = update.effective_user.id
balance = BALANCES.get(user_id, 0)
if balance <= 0:
await update.message.reply_text('❌ Нет мемов! Купите: /start')
return
await update.message.reply_text('🎨 Генерирую...')
try:
meme_img, meme_text = await generate_meme(update.message.text)
BALANCES[user_id] -= 1
await update.message.reply_photo(
photo=meme_img,
caption=f'😹 <b>Ваш IT-мем готов!</b>\n\n"{meme_text}"\n\n💰 Осталось: {BALANCES[user_id]} мемов',
parse_mode='HTML'
)
except Exception as e:
logger.error(f"Meme error: {e}")
await update.message.reply_text('❌ Ошибка. Попробуйте позже.')
def main():
app = Application.builder().token(TOKEN).build()
app.add_handler(CommandHandler("start", start))
app.add_handler(CallbackQueryHandler(button_handler))
app.add_handler(PreCheckoutQueryHandler(precheckout_callback))
app.add_handler(MessageHandler(filters.SUCCESSFUL_PAYMENT, successful_payment_callback))
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_text))
app.run_polling()
if __name__ == '__main__':
main()