def main() -> None:
"""Запуск бота."""
# Инициализация приложения
application = Application.builder().token(TELEGRAM_TOKEN).build()
# Создание обработчика разговора для создания статьи
conv_handler = ConversationHandler(
entry_points=[CommandHandler("start", start)],
states={
TOPIC: [MessageHandler(filters.TEXT & ~filters.COMMAND, topic_handler)],
KEYWORDS: [MessageHandler(filters.TEXT & ~filters.COMMAND, keywords_handler)],
AUDIENCE: [CallbackQueryHandler(audience_handler)],
PURPOSE: [CallbackQueryHandler(purpose_handler)],
LENGTH: [CallbackQueryHandler(length_handler)],
RESEARCH_OPTION: [CallbackQueryHandler(research_option_handler)],
RESEARCH_TYPE: [CallbackQueryHandler(research_type_handler)],
RESEARCH_QUERY: [
MessageHandler(filters.TEXT & ~filters.COMMAND, research_query_handler),
CallbackQueryHandler(research_query_handler, pattern="^use_main_topic$")
],
CONFIRM: [CallbackQueryHandler(confirm_handler)]
},
fallbacks=[CommandHandler("cancel", cancel)]
)
# Добавление обработчиков
application.add_handler(conv_handler)
application.add_handler(CommandHandler("users", list_users))
application.add_handler(CommandHandler("add_user", add_user))
# Запуск бота
application.run_polling()
# Декоратор для проверки авторизации
def authorized_only(func):
"""Декоратор для функций, требующих авторизации"""
async def wrapper(update: Update, context: ContextTypes.DEFAULT_TYPE, *args, **kwargs):
user_id = update.effective_user.id
# Проверка авторизации в кэше Redis для уменьшения латентности
if await redis_client.sismember("authorized_users", str(user_id)):
return await func(update, context, *args, **kwargs)
# Если не в кэше - проверяем в основной БД
if is_user_authorized(user_id):
# Добавляем в кэш для ускорения будущих проверок
await redis_client.sadd("authorized_users", str(user_id))
return await func(update, context, *args, **kwargs)
else:
# Обработка неавторизованного доступа
return await unauthorized_handler(update, context)
return wrapper
# Пример создания динамического промпта для исследования
def create_research_prompt(topic, research_type):
"""Создает специализированный промпт для исследования"""
base_prompt = f"Проведи исследование по теме '{topic}'."
type_prompts = {
"general_info": "Собери общую информацию, основные концепции, определения и контекст.",
"facts_stats": "Сосредоточься на фактах, статистике, данных и измеримых результатах.",
"scientific": "Найди академические источники, исследования, теории и методологии.",
"perspectives": "Представь различные позиции, мнения, подходы и школы мысли."
}
format_instructions = """
Структурируй информацию в виде кратких абзацев с ключевыми фактами.
Указывай источники информации, где это возможно.
Организуй материал в логические разделы.
"""
return f"{base_prompt} {type_prompts.get(research_type, type_prompts['general_info'])} {format_instructions}"