Telegram-боты с Claude API — от нуля до продакшена

7234567890:AAH_abcXYZ123...

#!/usr/bin/env bash
set -euo pipefail

source ~/.config/telegram-bot.env

API="https://api.telegram.org/bot${TELEGRAM_BOT_TOKEN}"
OFFSET=0

# Функция отправки (обрабатывает лимит 4096 символов)
send() {
  local chat="$1" text="$2"
  while [ -n "$text" ]; do
    local chunk="${text:0:4000}"
    text="${text:4000}"
    curl -s -X POST "${API}/sendMessage" \
      --data-urlencode "chat_id=${chat}" \
      --data-urlencode "text=${chunk}" > /dev/null
  done
}

handle_message() {
  local chat="$1" user_text="$2"

  # Только разрешённые пользователи
  if [ "$chat" != "$TELEGRAM_ALLOWED_CHAT_ID" ]; then
    return
  fi

  # ⚠️ Защита от shell-injection: stdin вместо аргумента
  # Если user_text содержит `$(rm -rf ~)` или backtick — без quoting это
  # выполнится как код. printf '%q' квотит безопасно, но надёжнее — передать
  # текст через stdin, чтобы он вообще не парсился shell'ом.
  local reply
  reply=$(printf '%s' "$user_text" | claude -p --model sonnet \
    --max-budget-usd 0.50 2>&1) || reply="Ошибка: $reply"

  send "$chat" "$reply"
}

# ⚠️ Polling loop — БЕЗ subshell pipe
# Старая версия `jq ... | while read` запускала while в subshell:
# переменная OFFSET там обновлялась, но снаружи оставалась прежней.
# Перенаправление через process substitution `< <(...)` оставляет while
# в текущем shell — OFFSET виден и в следующей итерации.
# curl --fail --max-time 30: рвём зависшие соединения, ловим HTTP 5xx.
while true; do
  UPDATES=$(curl -s --fail --max-time 35 \
    "${API}/getUpdates?offset=${OFFSET}&timeout=30") || {
    echo "Telegram API timeout/error, retrying in 5s..." >&2
    sleep 5
    continue
  }

  while read -r upd; do
    OFFSET=$(($(echo "$upd" | jq '.update_id') + 1))
    CHAT=$(echo "$upd" | jq '.message.chat.id')
    TEXT=$(echo "$upd" | jq -r '.message.text // empty')
    [ -n "$TEXT" ] && handle_message "$CHAT" "$TEXT"
  done < <(echo "$UPDATES" | jq -c '.result[]?')
done

TELEGRAM_BOT_TOKEN=7234567890:AAH_твой_токен
TELEGRAM_ALLOWED_CHAT_ID=123456789   # Твой chat_id (узнай у @userinfobot)
ANTHROPIC_API_KEY=sk-ant-...

chmod 600 ~/.config/telegram-bot.env
chmod +x bot.sh
./bot.sh

pip install python-telegram-bot anthropic

import asyncio
import anthropic
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes
import os

claude = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

# ⚠️ Память: in-memory словарь — ТОЛЬКО для разработки и одного процесса
#
# Проблема: defaultdict(list) копит запись на каждый новый user_id навсегда.
# 10 000 пользователей × 20 сообщений × ~500 байт = ~100 МБ RAM, и растёт.
# При рестарте процесса вся история теряется.
#
# Для production: Redis / Cloudflare KV / Postgres с TTL.
# Ниже — LRU-эвикция по размеру (защита от OOM на in-memory варианте).
from collections import OrderedDict

MAX_HISTORY = 20       # сообщений на пользователя
MAX_USERS = 1000       # активных пользователей в памяти (LRU eviction)

class LRUConversations(OrderedDict):
    """LRU-словарь: при превышении MAX_USERS удаляет самого старого."""
    def __getitem__(self, key):
        if key not in self:
            self[key] = []
        else:
            self.move_to_end(key)  # mark as recently used
        return super().__getitem__(key)

    def __setitem__(self, key, value):
        super().__setitem__(key, value)
        self.move_to_end(key)
        if len(self) > MAX_USERS:
            evicted_key, _ = self.popitem(last=False)  # drop oldest
            # В production: лог + alert "history evicted for user X"

conversation_history = LRUConversations()

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user = update.effective_user
    await update.message.reply_text(
        f"Привет, {user.first_name}! Я AI-ассистент. Задай любой вопрос."
    )

async def clear_history(update: Update, context: ContextTypes.DEFAULT_TYPE):
    conversation_history[update.effective_user.id].clear()
    await update.message.reply_text("История очищена.")

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_id = update.effective_user.id
    user_text = update.message.text
    
    # Показываем "печатает..."
    await context.bot.send_chat_action(
        chat_id=update.effective_chat.id,
        action="typing"
    )
    
    # Добавляем в историю
    conversation_history[user_id].append({
        "role": "user",
        "content": user_text
    })
    
    # Обрезаем если слишком длинная
    if len(conversation_history[user_id]) > MAX_HISTORY:
        conversation_history[user_id] = conversation_history[user_id][-MAX_HISTORY:]
    
    # Запрос к Claude
    response = claude.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=2000,
        system="Ты полезный AI-ассистент. Отвечай кратко и по делу.",
        messages=conversation_history[user_id]
    )
    
    reply = "".join(b.text for b in response.content if b.type == "text")
    
    # Сохраняем ответ
    conversation_history[user_id].append({
        "role": "assistant",
        "content": reply
    })
    
    # Telegram лимит 4096 символов
    if len(reply) > 4096:
        for i in range(0, len(reply), 4096):
            await update.message.reply_text(reply[i:i+4096])
    else:
        await update.message.reply_text(reply)

def main():
    app = Application.builder().token(os.environ["TELEGRAM_BOT_TOKEN"]).build()
    app.add_handler(CommandHandler("start", start))
    app.add_handler(CommandHandler("clear", clear_history))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
    print("Бот запущен...")
    app.run_polling()

if __name__ == "__main__":
    main()

import "dotenv/config";
import { Bot } from "grammy";
import Anthropic from "@anthropic-ai/sdk";

const bot = new Bot(process.env.TELEGRAM_BOT_TOKEN!);
const claude = new Anthropic();

// ⚠️ История диалогов — in-memory Map, ТОЛЬКО для разработки
//
// Проблема: Map<number, ...> копит запись на каждый user_id навсегда.
// При 10k пользователей процесс съест всю RAM и упадёт (OOM).
// При рестарте — вся история теряется (не persistent).
//
// Для production используй Redis или Cloudflare KV с TTL:
//   await env.KV.put(`chat:${userId}`, JSON.stringify(history),
//                    { expirationTtl: 60 * 60 * 24 * 7 }); // 7 дней
//
// Ниже — LRU-эвикция по размеру (защита от OOM на single-process сценарии).
const MAX_USERS = 1000;
const conversations = new Map<number, Array<{role: string, content: string}>>();

function getHistory(userId: number): Array<{role: string, content: string}> {
  if (conversations.has(userId)) {
    // LRU touch: удалить и вернуть в конец (Map сохраняет insertion order)
    const h = conversations.get(userId)!;
    conversations.delete(userId);
    conversations.set(userId, h);
    return h;
  }
  // Эвикция самого старого если переполнение
  if (conversations.size >= MAX_USERS) {
    const oldestKey = conversations.keys().next().value;
    if (oldestKey !== undefined) conversations.delete(oldestKey);
  }
  const fresh: Array<{role: string, content: string}> = [];
  conversations.set(userId, fresh);
  return fresh;
}

bot.on("message:text", async (ctx) => {
  const userId = ctx.from.id;
  const userText = ctx.message.text;
  
  const history = getHistory(userId);  // LRU-touch + create if absent
  history.push({ role: "user", content: userText });
  
  // Placeholder сообщение
  const placeholder = await ctx.reply("...");
  
  let buffer = "";
  let lastEdit = Date.now();
  
  // Стриминг от Claude
  const stream = claude.messages.stream({
    model: "claude-sonnet-5-5",
    max_tokens: 2000,
    system: "Ты полезный AI-ассистент. Отвечай на русском языке.",
    messages: history as any,
  });
  
  for await (const chunk of stream) {
    if (chunk.type === "content_block_delta" && 
        chunk.delta.type === "text_delta") {
      buffer += chunk.delta.text;
      
      // Обновляем каждые 800мс (лимит Telegram API)
      if (Date.now() - lastEdit > 800 && buffer.length > 20) {
        try {
          await ctx.api.editMessageText(
            ctx.chat.id, placeholder.message_id, buffer.slice(0, 4000)
          );
          lastEdit = Date.now();
        } catch (e) { /* сообщение не изменилось */ }
      }
    }
  }
  
  // Финальное обновление
  await ctx.api.editMessageText(ctx.chat.id, placeholder.message_id, buffer.slice(0, 4000));
  
  history.push({ role: "assistant", content: buffer });
  if (history.length > 20) conversations.set(userId, history.slice(-20));
});

bot.command("clear", async (ctx) => {
  conversations.delete(ctx.from.id);
  await ctx.reply("История очищена.");
});

bot.start();

import io
import openai
import aiohttp
from telegram.ext import MessageHandler, filters

openai_client = openai.AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])

async def handle_voice(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await context.bot.send_chat_action(update.effective_chat.id, "typing")
    
    # Шаг 1: Скачиваем OGG файл
    file = await context.bot.get_file(update.message.voice.file_id)
    async with aiohttp.ClientSession() as session:
        async with session.get(file.file_path) as resp:
            audio_bytes = await resp.read()
    
    # Шаг 2: Транскрибируем через Whisper
    audio_file = io.BytesIO(audio_bytes)
    audio_file.name = "voice.ogg"
    
    # имя модели распознавания — пример: актуальные модели смотри в документации OpenAI
    transcription = await openai_client.audio.transcriptions.create(
        model="whisper-1",
        file=audio_file,
        language="ru"
    )
    
    transcript = transcription.text
    
    # Шаг 3: Показываем что распознали
    await update.message.reply_text(f"🎙️ Распознал: _{transcript}_", parse_mode="Markdown")
    
    # Шаг 4: Передаём в Claude
    response = claude.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=1000,
        messages=[{"role": "user", "content": transcript}]
    )
    await update.message.reply_text("".join(b.text for b in response.content if b.type == "text"))

app.add_handler(MessageHandler(filters.VOICE, handle_voice))

from telegram import InlineKeyboardButton, InlineKeyboardMarkup

async def show_menu(update: Update, context: ContextTypes.DEFAULT_TYPE):
    keyboard = [
        [
            InlineKeyboardButton("📝 Написать пост", callback_data="mode_content"),
            InlineKeyboardButton("📊 Аналитика", callback_data="mode_analytics"),
        ],
        [InlineKeyboardButton("🗑️ Очистить историю", callback_data="clear_history")],
    ]
    await update.message.reply_text(
        "Выбери режим:", reply_markup=InlineKeyboardMarkup(keyboard)
    )

async def handle_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
    query = update.callback_query
    await query.answer()  # ОБЯЗАТЕЛЬНО — убирает "часики" у кнопки
    
    if query.data == "mode_content":
        context.user_data["mode"] = "content"
        await query.edit_message_text("Режим: Контент-помощник. Пиши тему 👇")
    elif query.data == "clear_history":
        conversation_history[query.from_user.id].clear()
        await query.answer("История очищена!", show_alert=True)

from telegram.ext import CallbackQueryHandler
app.add_handler(CallbackQueryHandler(handle_callback))

# Установка
npm install -g wrangler
wrangler login

# Новый проект
mkdir my-telegram-bot && cd my-telegram-bot
npm install grammy @anthropic-ai/sdk

# Секреты (не в код!)
wrangler secret put BOT_TOKEN
wrangler secret put ANTHROPIC_API_KEY
wrangler secret put WEBHOOK_SECRET

name = "my-telegram-bot"
main = "src/index.ts"
compatibility_date = "2026-10-01"  # поставь актуальную дату на момент создания проекта

import { Bot, webhookCallback } from "grammy";
import Anthropic from "@anthropic-ai/sdk";

export interface Env {
  BOT_TOKEN: string;
  ANTHROPIC_API_KEY: string;
  WEBHOOK_SECRET: string;
}

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    // Проверка secret_token
    const secret = request.headers.get("X-Telegram-Bot-Api-Secret-Token");
    if (secret !== env.WEBHOOK_SECRET) {
      return new Response("Unauthorized", { status: 401 });
    }
    
    const bot = new Bot(env.BOT_TOKEN);
    const claude = new Anthropic({ apiKey: env.ANTHROPIC_API_KEY });
    
    bot.on("message:text", async (ctx) => {
      const response = await claude.messages.create({
        model: "claude-haiku-4-5",  // Haiku — экономия на Workers
        max_tokens: 1024,
        messages: [{ role: "user", content: ctx.message.text }],
      });
      await ctx.reply(response.content.map((b) => (b.type === "text" ? b.text : "")).join("") as string);
    });
    
    return webhookCallback(bot, "cloudflare-mod")(request);
  },
};

# Деплой
wrangler deploy
# Вывод: https://my-telegram-bot.YOUR-USERNAME.workers.dev

# Регистрация webhook (один раз)
curl "https://api.telegram.org/bot<BOT_TOKEN>/setWebhook" \
  -d "url=https://my-telegram-bot.YOUR-USERNAME.workers.dev" \
  -d "secret_token=<WEBHOOK_SECRET>"

# Структура проекта
my-bot/
├── bot.py
├── requirements.txt   # python-telegram-bot и anthropic (версии бери актуальные)
└── Procfile           # web: python bot.py

# Деплой
git init && git add . && git commit -m "initial"
# → Railway.com → New Project → Deploy from GitHub
# Добавить переменные: TELEGRAM_BOT_TOKEN, ANTHROPIC_API_KEY

# ❌ ЗАПРЕЩЕНО
bot = Bot(token="7234567890:AAH...")

# ✅ ПРАВИЛЬНО
bot = Bot(token=os.environ["TELEGRAM_BOT_TOKEN"])

ALLOWED_USERS = {123456789}  # Твои chat_id

async def check_access(update: Update) -> bool:
    if update.effective_user.id not in ALLOWED_USERS:
        await update.message.reply_text("Нет доступа.")
        return False
    return True

from collections import defaultdict
from datetime import datetime, timedelta

requests = defaultdict(list)

def check_rate_limit(user_id: int, max_req=5, window_sec=60) -> bool:
    now = datetime.now()
    cutoff = now - timedelta(seconds=window_sec)
    requests[user_id] = [ts for ts in requests[user_id] if ts > cutoff]
    if len(requests[user_id]) >= max_req:
        return False
    requests[user_id].append(now)
    return True

SUPPORT_PROMPT = """Ты агент поддержки компании Acme Realty.
Помогаешь русскоязычным найти недвижимость в Эквадоре.
При вопросах о ценах — уточни бюджет и предпочтения.
При жалобах — принеси извинения и попроси контакт.
Если не знаешь — честно скажи и предложи связаться с менеджером."""

Нажал кнопку → выбрал интерес (купить/арендовать) 
→ ввёл бюджет → ввёл контакт 
→ менеджер получил лид в отдельный чат

MODES = {
    "telegram": "Пост для Telegram, макс 800 символов, 3-5 эмодзи",
    "instagram": "Instagram с 10-15 хэштегами в конце",
    "blog": "Статья минимум 800 слов с подзаголовками",
}