Инструкция для создание маркетингового ассистента с использованием Telegram бота и ChatGPT - Python код с объяснениями логики

import asyncio from telegram import Update, Bot from telegram.ext import Application, MessageHandler, filters, ContextTypes, CommandHandler from openai import AsyncOpenAI import asyncpg from datetime import datetime # ... другие импорты

logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__)

async def db_connect(): try: connection = await asyncpg.connect(user='...', password='...', database='...', host='...') logger.info("Успешное подключение к базе данных") return connection except Exception as e: logger.error(f"Ошибка подключения к базе данных: {e}") raise

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE): # ... код обработки команды /start async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE): # ... код обработки входящих сообщений

client = AsyncOpenAI(api_key=openai_api_key) async def update_chat_history(chat_id, role, content, username): # ... код обновления истории чата chat_completion = await client.chat.completions.create( model="gpt-3.5-turbo", messages=messages_with_context, ) answer = chat_completion.choices[0].message.content

async def save_user_data_if_not_exists(context, telegram_id, username, full_name): # ... код сохранения данных пользователя async def save_message_to_db(telegram_id, username, content, role): # ... код сохранения сообщения в базу данных

async def send_broadcast_message(bot, message_text): # ... код для массовой рассылки сообщений async def sendall_command(update: Update, context: ContextTypes.DEFAULT_TYPE): # ... код обработки команды рассылки

import asyncio from telegram import Update, Bot from telegram.ext import Application, MessageHandler, filters, ContextTypes, CommandHandler from openai import AsyncOpenAI import csv import logging from telegram import InlineKeyboardButton, InlineKeyboardMarkup from functools import partial import asyncpg from datetime import datetime logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) admin_state = {} async def db_connect(): try: connection = await asyncpg.connect(user='имя пользователя', password='пароль', database='название базы данных', host='localhost') logger.info("Успешное подключение к базе данных") return connection except Exception as e: logger.error(f"Ошибка подключения к базе данных: {e}") raise # Словарь, сопоставляющий команды с URL каналов. Можно заменить или удалить channels = { 'channel1': ('https://t.me/+JsobbT1ArkY3M2Zi', 'Эй-Ай 🤖 Искусственный интеллект'), 'channel2': ('https://t.me/+oWkVbBbGQtFiYTAy', 'ИИ инструменты в маркетинге'), 'channel3': ('https://t.me/+Z_20lkPs2ItlNDcy', 'Богатый ремесленник') } async def channel_command(update: Update, context: ContextTypes.DEFAULT_TYPE, channel_url, channel_name): keyboard = [[InlineKeyboardButton("Перейти в канал", url=channel_url)]] reply_markup = InlineKeyboardMarkup(keyboard) await context.bot.send_message(chat_id=update.effective_chat.id, text=f"Подписаться на канал: {channel_name}", reply_markup=reply_markup) def get_main_menu(): keyboard = [ [InlineKeyboardButton("Эй-Ай 🤖 Искусственный интеллект", url="https://t.me/+JsobbT1ArkY3M2Zi")], [InlineKeyboardButton("ИИ инструменты в маркетинге", url="https://t.me/+oWkVbBbGQtFiYTAy")], [InlineKeyboardButton("Богатый ремесленник", url="https://t.me/+Z_20lkPs2ItlNDcy")] ] return InlineKeyboardMarkup(keyboard) async def start(update: Update, context: ContextTypes.DEFAULT_TYPE): full_name = update.effective_user.full_name telegram_id = update.effective_user.id username = update.effective_user.username welcome_message = f"""Привет, {full_name}. Этот бот открывает доступ к ChatGPT 3.5 Turbo - БЕСПЛАТНО🚀 - без ограничений по количеству запросов. Работу бота настроил и оплачивает маркетолог <a href="https://t.me/aglamov">Динар Аглямов</a>. Так же, бот может вас проконсультировать какие маркетинговые услуги можно приобрести у Динара. ✅ Чат-бот умеет: - Ответы на вопросы (для быстрого поиска информации) - Генерация текста (от SMM до сказок детям) - Создание кода (например, этого бота написал при поддержке ChatGPT 3.5) - Помощь с математическими задачами - Генерация идей и контента (создание изображений будет в другом боте) - Обработка текстов на разных языках - Интерактивный диалог (просто поговорить =)) - И многое другое 👉🏻 Для запуска бота, напишите в чат вопрос на любом языке Приятного пользования 💭 Теперь, можно задать свой вопрос исскуственному интеллекту!""" # Отправляем приветственное сообщение с использованием HTML await context.bot.send_message( chat_id=update.effective_chat.id, text=welcome_message, parse_mode='HTML', disable_web_page_preview=True # Отключаем предварительный просмотр веб-страницы ) # Отправляем меню с кнопками menu = get_main_menu() await context.bot.send_message( chat_id=update.effective_chat.id, text="Выберите один из каналов, чтобы подписаться:", reply_markup=menu ) await save_user_data_if_not_exists(context, telegram_id, username, full_name) async def notify_admin(context: ContextTypes.DEFAULT_TYPE, message: str): admin_chat_id = 'chat_id администратора' # Замените на chat_id администратора или пользователя try: await context.bot.send_message(chat_id=admin_chat_id, text=message) except Exception as e: logger.error(f"Ошибка при отправке сообщения администратору: {e}") async def save_user_data_if_not_exists(context, telegram_id, username, full_name): conn = await db_connect() try: user_exists = await conn.fetchval('SELECT EXISTS(SELECT 1 FROM users WHERE telegram_id = $1)', telegram_id) if not user_exists: await conn.execute('INSERT INTO users(telegram_id, username, full_name) VALUES ($1, $2, $3)', telegram_id, username, full_name) logger.info(f"Новый пользователь сохранен: {full_name} (@{username})") await notify_admin(context, f"Новый пользователь: {full_name} (@{username})") except Exception as e: logger.error(f"Ошибка при сохранении данных пользователя: {e}") finally: await conn.close() async def save_message_to_db(telegram_id, username, content, role): conn = await db_connect() try: await conn.execute( ''' INSERT INTO messages(telegram_id, username, content, role, timestamp) VALUES($1, $2, $3, $4, $5) ''', telegram_id, username, content, role, datetime.now() ) logger.info("Сообщение успешно сохранено в базу данных.") except Exception as e: logger.error(f"Ошибка при сохранении сообщения в базу данных: {e}") finally: await conn.close() async def send_broadcast_message(bot, message_text): conn = await db_connect() try: subscribers = await conn.fetch('SELECT telegram_id FROM users') logger.info("Получен список подписчиков для рассылки") except Exception as e: logger.error(f"Ошибка при получении списка подписчиков: {e}") return 0, 0, [] finally: await conn.close() success_count = 0 failed_count = 0 failed_subscribers = [] for record in subscribers: subscriber_id = record['telegram_id'] try: await bot.send_message(chat_id=subscriber_id, text=message_text) success_count += 1 except Exception as e: logger.error(f"Не удалось отправить сообщение {subscriber_id}: {e}") failed_count += 1 failed_subscribers.append(subscriber_id) logger.info(f"Рассылка завершена. Успешно: {success_count}, Не удалось: {failed_count}") return success_count, failed_count, failed_subscribers openai_api_key = "API ключ" telegram_token = "токен бота" bot = Bot(token=telegram_token) client = AsyncOpenAI(api_key=openai_api_key) chat_histories = {} # Функция для добавления сообщения в историю чата async def update_chat_history(chat_id, role, content, username): if chat_id not in chat_histories: chat_histories[chat_id] = [] if len(chat_histories[chat_id]) > 10: chat_histories[chat_id].pop(0) # Удаляем самое старое сообщение chat_histories[chat_id].append({"role": role, "content": content}) await save_message_to_db(chat_id, username, content, role) async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: telegram_id = update.effective_user.id user_message = update.message.text chat_id = update.effective_chat.id username = update.effective_user.username full_name = update.effective_user.full_name logger.info(f"Получено сообщение от {username}: {user_message}") # Логика для обработки текста рассылки от администратора if admin_state.get(telegram_id) == "awaiting_message_text": if telegram_id == телеграм id админа: success_count, failed_count, failed_subscribers = await send_broadcast_message(context.bot, user_message) admin_state[telegram_id] = None # Сброс состояния report_message = f"Рассылка выполнена.\nУспешно отправлено сообщений: {success_count}.\nНе удалось отправить сообщений: {failed_count}." if failed_count > 0: report_message += f"\nНе удалось отправить следующим пользователям: {', '.join(failed_subscribers)}" await update.message.reply_text(report_message) return # Сохраняем данные пользователя, если они еще не были сохранены await save_user_data_if_not_exists(chat_id, telegram_id, username, full_name) # Обновляем историю диалога для текущего чата await update_chat_history(chat_id, "user", user_message, username) # Предустановленный контекст и т.д. personal_context = [ {"role": "system", "content": "Ты ассистент маркетолога Динара Аглямова, твоя задача нативно и предлагать консультации у Динара, предлагать пройти мини курсы и подписываться на его телеграм каналы и отвечать на всевозможные вопросы пользователей. Динар Аглямов - маркетолог со стажем 13 лет. Проходил обучение у Максима Белоусова по разработке маркетинговых стратегий. Динар оказывает услуги разработки. маркетинговых стратегий, консультирует по вопросам запуска лидогенерации для B2B, имеет большой опыт (более 40 запусков) лидогенерации на франшизы. Работает с нишами в B2B секторе. Для связи с Динаром контакт в телеграм @aglamov. Динар записывает минин-курсы по маркетингку и внедрению искусственного интеллекта в бизнес процессы. Мини курсы можно посмотреть по ссылке: http://aglamovcourses.ru. Так же, если пользователь хочет больше узнать про маркетинг и задает вопросы касательно маркетиннка, нативно предлагай подписаться на телеграм канал Динара про маркетинг. Вот ссылка https://t.me/+oWkVbBbGQtFiYTAy. Если пользователи интересуются искусственным интелектом, нативно предлагай им подписаться на телеграм канал Динара Эй-Ай 🤖 Искусственный интеллект по ссылке https://t.me/+JsobbT1ArkY3M2Zi. Будь не навязчив. Выполняй роль Ассистента Динара Аглямова, который умеет отвечает на все вопросы. Общайся только на русском языке. Коротко о проектах: с 2011 по 2015 съемка видео. Реклама и семейная съемка, 2014 год - аренда камер Go Pro, 2014 год - продажа игрушек BRUDER, с 2013 по 2014 купил пилораму занимался лесопереработкой, с 2013 по 2017 кондитерская Мистер Макарун. С кухни съемной квартиры до производства в 90м2, в мае 2016 году - пробую собирать и продавать Дизайнерскую мебель, с декабря 2017 по июнь 2017 открываю под ключ пекарню по франшизе Хлеб из Тандыра, в мае 2017 продажа мягких игрушек ЖДУН, в июне 2017 регистрирую домен aglamov.biz, где выкладываю свои статьи про бизнес, фриланс и продвижение. Начинаю углубленное изучение SEO продвижения. августе 2018 зарегистрировал ИП в третий раз. Начал официально заниматься продвижением бизнеса в интернете. в 2022 прошел курсы у Максима Белоусова. в 2024 погрузился в изучение искуссвенного интеллекта. Сейчас занимается разработкой программы для автоматизированного анализа звонков отдела продаж и обслуживания на основе ии. Часть кейсов и портфолио Динара можно найти по ссылке: https://vc.ru/u/1047023-dinar-aglyamov-pro-marketing, "}, ] messages_with_context = personal_context + chat_histories.get(chat_id, []) chat_completion = await client.chat.completions.create( model="gpt-3.5-turbo", messages=messages_with_context, ) answer = chat_completion.choices[0].message.content logger.info(f"Ответ от ChatGPT: {answer}") await update_chat_history(chat_id, "assistant", answer, username) # Отправляем ответ пользователю await context.bot.send_message(chat_id=chat_id, text=answer) async def sendall_command(update: Update, context: ContextTypes.DEFAULT_TYPE): telegram_id = update.effective_user.id if telegram_id == Telegram ID администратора: # Замените на ваш Telegram ID администратора admin_state[telegram_id] = "awaiting_message_text" await update.message.reply_text("Пожалуйста, отправьте текст для рассылки.") else: await update.message.reply_text("У вас нет прав для выполнения этой команды.") if __name__ == '__main__': application = Application.builder().token(telegram_token).build() application.add_handler(CommandHandler('start', start)) application.add_handler(CommandHandler('sendall', sendall_command)) for command, (url, name) in channels.items(): handler = CommandHandler(command, partial(channel_command, channel_url=url, channel_name=name)) application.add_handler(handler) application.add_handler(MessageHandler(filters.TEXT, handle_message)) loop = asyncio.get_event_loop() loop.run_until_complete(application.run_polling())