mirror of
https://github.com/eternnoir/pyTelegramBotAPI.git
synced 2023-08-10 21:12:57 +03:00
BaseMiddleware returned to it's original place && I18N middleware is now only in examples
This commit is contained in:
@ -1,6 +1,6 @@
|
||||
# Just a little example of middleware handlers
|
||||
|
||||
from telebot.asyncio_middlewares import BaseMiddleware
|
||||
from telebot.asyncio_handler_backends import BaseMiddleware
|
||||
from telebot.async_telebot import AsyncTeleBot
|
||||
from telebot.async_telebot import CancelUpdate
|
||||
bot = AsyncTeleBot('TOKEN')
|
||||
|
@ -7,7 +7,6 @@
|
||||
# But this example just to show the work of middlewares.
|
||||
|
||||
import telebot
|
||||
import telebot.asyncio_middlewares
|
||||
from telebot.async_telebot import AsyncTeleBot
|
||||
from telebot import asyncio_handler_backends
|
||||
import logging
|
||||
@ -28,7 +27,7 @@ TRANSLATIONS = {
|
||||
bot = AsyncTeleBot('TOKEN')
|
||||
|
||||
|
||||
class LanguageMiddleware(telebot.asyncio_middlewares.BaseMiddleware):
|
||||
class LanguageMiddleware(asyncio_handler_backends.BaseMiddleware):
|
||||
def __init__(self):
|
||||
self.update_types = ['message'] # Update types that will be handled by this middleware.
|
||||
async def pre_process(self, message, data):
|
||||
|
@ -0,0 +1,120 @@
|
||||
import contextvars
|
||||
import gettext
|
||||
import os
|
||||
|
||||
from telebot.asyncio_handler_backends import BaseMiddleware
|
||||
|
||||
try:
|
||||
from babel.support import LazyProxy
|
||||
|
||||
babel_imported = True
|
||||
except ImportError:
|
||||
babel_imported = False
|
||||
|
||||
|
||||
class I18N(BaseMiddleware):
|
||||
"""
|
||||
This middleware provides high-level tool for internationalization
|
||||
It is based on gettext util.
|
||||
"""
|
||||
|
||||
context_lang = contextvars.ContextVar('language', default=None)
|
||||
|
||||
def __init__(self, translations_path, domain_name: str):
|
||||
super().__init__()
|
||||
self.update_types = self.process_update_types()
|
||||
|
||||
self.path = translations_path
|
||||
self.domain = domain_name
|
||||
self.translations = self.find_translations()
|
||||
|
||||
@property
|
||||
def available_translations(self):
|
||||
return list(self.translations)
|
||||
|
||||
def gettext(self, text: str, lang: str = None):
|
||||
"""
|
||||
Singular translations
|
||||
"""
|
||||
|
||||
if lang is None:
|
||||
lang = self.context_lang.get()
|
||||
|
||||
if lang not in self.translations:
|
||||
return text
|
||||
|
||||
translator = self.translations[lang]
|
||||
return translator.gettext(text)
|
||||
|
||||
def ngettext(self, singular: str, plural: str, lang: str = None, n=1):
|
||||
"""
|
||||
Plural translations
|
||||
"""
|
||||
if lang is None:
|
||||
lang = self.context_lang.get()
|
||||
|
||||
if lang not in self.translations:
|
||||
if n == 1:
|
||||
return singular
|
||||
return plural
|
||||
|
||||
translator = self.translations[lang]
|
||||
return translator.ngettext(singular, plural, n)
|
||||
|
||||
def lazy_gettext(self, text: str, lang: str = None):
|
||||
if not babel_imported:
|
||||
raise RuntimeError('babel module is not imported. Check that you installed it.')
|
||||
return LazyProxy(self.gettext, text, lang, enable_cache=False)
|
||||
|
||||
def lazy_ngettext(self, singular: str, plural: str, lang: str = None, n=1):
|
||||
if not babel_imported:
|
||||
raise RuntimeError('babel module is not imported. Check that you installed it.')
|
||||
return LazyProxy(self.ngettext, singular, plural, lang, n, enable_cache=False)
|
||||
|
||||
async def get_user_language(self, obj):
|
||||
"""
|
||||
You need to override this method and return user language
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def process_update_types(self) -> list:
|
||||
"""
|
||||
You need to override this method and return any update types which you want to be processed
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
async def pre_process(self, message, data):
|
||||
"""
|
||||
context language variable will be set each time when update from 'process_update_types' comes
|
||||
value is the result of 'get_user_language' method
|
||||
"""
|
||||
self.context_lang.set(await self.get_user_language(obj=message))
|
||||
|
||||
async def post_process(self, message, data, exception):
|
||||
pass
|
||||
|
||||
def find_translations(self):
|
||||
"""
|
||||
Looks for translations with passed 'domain' in passed 'path'
|
||||
"""
|
||||
if not os.path.exists(self.path):
|
||||
raise RuntimeError(f"Translations directory by path: {self.path!r} was not found")
|
||||
|
||||
result = {}
|
||||
|
||||
for name in os.listdir(self.path):
|
||||
translations_path = os.path.join(self.path, name, 'LC_MESSAGES')
|
||||
|
||||
if not os.path.isdir(translations_path):
|
||||
continue
|
||||
|
||||
po_file = os.path.join(translations_path, self.domain + '.po')
|
||||
mo_file = po_file[:-2] + 'mo'
|
||||
|
||||
if os.path.isfile(po_file) and not os.path.isfile(mo_file):
|
||||
raise FileNotFoundError(f"Translations for: {name!r} were not compiled!")
|
||||
|
||||
with open(mo_file, 'rb') as file:
|
||||
result[name] = gettext.GNUTranslations(file)
|
||||
|
||||
return result
|
@ -56,7 +56,7 @@ import keyboards
|
||||
from telebot import types
|
||||
from telebot.async_telebot import AsyncTeleBot
|
||||
from telebot.asyncio_filters import TextMatchFilter, TextFilter
|
||||
from telebot.asyncio_middlewares import I18N
|
||||
from i18n_base_midddleware import I18N
|
||||
from telebot.asyncio_storage.memory_storage import StateMemoryStorage
|
||||
|
||||
|
||||
@ -87,7 +87,7 @@ class I18NMiddleware(I18N):
|
||||
|
||||
|
||||
storage = StateMemoryStorage()
|
||||
bot = AsyncTeleBot("", state_storage=storage)
|
||||
bot = AsyncTeleBot("1254795383:AAE7gbj1aas4lEDHB1eVuZZhSGPWcH1B5ds", state_storage=storage)
|
||||
|
||||
i18n = I18NMiddleware(translations_path='locales', domain_name='messages')
|
||||
_ = i18n.gettext # for singular translations
|
||||
|
Reference in New Issue
Block a user