telegram-translation-bot/bot.py at main · Manuel0516/telegram-translation-bot · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
"""Telegram interface: a thin layer around the local translation model."""

import logging
import re
from collections import defaultdict, deque

from telegram import Update
from telegram.ext import (
    Application,
    CommandHandler,
    ContextTypes,
    MessageHandler,
    filters,
)

from config import MAX_CONTEXT_MESSAGES, TELEGRAM_BOT_TOKEN

_EMOJI_RE = re.compile(
    "[\U00002600-\U000027BF"
    "\U0001F300-\U0001F9FF"
    "\U0001FA00-\U0001FAFF"
    "\U0001F1E0-\U0001F1FF"
    "︀-️"
    "‍⃣]+"
)


def _is_emoji_only(text: str) -> bool:
    return not _EMOJI_RE.sub("", text).strip()
from translator import translate_message

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
logger = logging.getLogger(__name__)

# Per-chat state, kept in memory only (no database in version 1).
enabled: dict[int, bool] = defaultdict(bool)
history: dict[int, deque[str]] = defaultdict(lambda: deque(maxlen=MAX_CONTEXT_MESSAGES))

HELP_TEXT = (
    "I translate between Spanish, English and Swedish.\n"
    "Send a message and I reply with the other two languages.\n\n"
    "/on — enable automatic translation\n"
    "/off — disable automatic translation\n"
    "/status — show whether translation is on or off\n"
    "/help — show this help"
)


async def start(update: Update, _: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text(HELP_TEXT)


async def help_command(update: Update, _: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text(HELP_TEXT)


async def on(update: Update, _: ContextTypes.DEFAULT_TYPE) -> None:
    enabled[update.effective_chat.id] = True
    await update.message.reply_text("Translation is ON.")


async def off(update: Update, _: ContextTypes.DEFAULT_TYPE) -> None:
    enabled[update.effective_chat.id] = False
    await update.message.reply_text("Translation is OFF.")


async def status(update: Update, _: ContextTypes.DEFAULT_TYPE) -> None:
    state = "ON" if enabled[update.effective_chat.id] else "OFF"
    await update.message.reply_text(f"Translation is {state}.")


async def translate(update: Update, _: ContextTypes.DEFAULT_TYPE) -> None:
    chat_id = update.effective_chat.id
    text = update.message.text
    if not enabled[chat_id] or not text or not text.strip() or _is_emoji_only(text):
        return

    try:
        context = [m for m in history[chat_id] if not _is_emoji_only(m)]
        result = await translate_message(text, context)
    except Exception:
        logger.exception("Translation failed")
        await update.message.reply_text(
            "Translation failed. Check that the local model server is running."
        )
        return

    await update.message.reply_text(result)
    history[chat_id].append(text)


def main() -> None:
    app = Application.builder().token(TELEGRAM_BOT_TOKEN).build()
    app.add_handler(CommandHandler("start", start))
    app.add_handler(CommandHandler("help", help_command))
    app.add_handler(CommandHandler("on", on))
    app.add_handler(CommandHandler("off", off))
    app.add_handler(CommandHandler("status", status))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, translate))
    logger.info("Bot started.")
    app.run_polling()


if __name__ == "__main__":
    main()