diff --git a/README.md b/README.md index b50dbda..9611f7f 100644 --- a/README.md +++ b/README.md @@ -1,59 +1,3 @@ -# Telegram Image Description Bot +# b2b_assistants -Бот для Telegram, который описывает изображения с помощью Qwen-VL API. - -## Требования - -- Python 3.10+ -- Токен Telegram бота (получить у [@BotFather](https://t.me/BotFather)) -- Ключ доступа к Qwen API - -## Установка - -1. Установите зависимости: -```bash -pip install -r requirements.txt -``` - -2. Настройте переменные окружения в файле `.env`: -```env -# Telegram Bot Token (получите у @BotFather) -TELEGRAM_BOT_TOKEN=ваш_токен_бота - -# Qwen API настройки (уже заполнены) - -``` - -3. Запустите бота: -```bash -python src/image_bot.py -``` - -## Использование - -1. Найдите вашего бота в Telegram и нажмите `/start` -2. Отправьте боту изображение -3. Бот вернёт описание изображения на русском языке - -## Команды - -- `/start` - начать работу с ботом -- `/help` - показать справку -- `/settoken ` - установить токен API (временное решение) - -## Структура проекта - -``` -b2b_assistants/ -├── .env # Переменные окружения -├── requirements.txt # Зависимости Python -├── README.md # Документация -└── src/ - └── image_bot.py # Основной код бота -``` - -## Примечания - -- Бот использует base64 кодирование для отправки изображений в Qwen-VL API -- Для ограничения доступа используйте переменную `ALLOWED_USERS` в `.env` -- Время обработки изображения может составлять до 2 минут для больших файлов \ No newline at end of file +Репозиторий для разработки B to B рушений \ No newline at end of file diff --git a/requirements.txt b/requirements.txt deleted file mode 100644 index b51df25..0000000 --- a/requirements.txt +++ /dev/null @@ -1,33 +0,0 @@ -# Telegram Bot -python-telegram-bot==20.7 - -# Matrix Bot (use matrix-nio for Python 3.12+ compatibility) -matrix-nio==0.25.2 - -# Common -python-dotenv==1.0.0 -requests==2.31.0 - -# Matplotlib for diagram generation -matplotlib==3.9.0 - -# Plotly for interactive charts and advanced visualizations -plotly==5.24.0 - -# Graphviz for flowcharts and diagrams -graphviz==0.20.3 - -# NetworkX for network/graph diagrams -networkx==3.3 - -# PyDot for graph visualization -pydot==3.0.1 - -# Pillow for image processing -Pillow==10.4.0 - -# Pandas for data manipulation in charts -pandas==2.2.2 - -# NumPy for numerical operations -numpy==1.26.4 \ No newline at end of file diff --git a/src/image_bot.py b/src/image_bot.py deleted file mode 100644 index 7d756dc..0000000 --- a/src/image_bot.py +++ /dev/null @@ -1,1272 +0,0 @@ -#!/usr/bin/env python3 -""" -Telegram Bot that describes images using Qwen-VL API and generates diagrams using Qwen + Python libraries. -All image generation is done through Qwen API + matplotlib/graphviz/networkx. -""" - -import os -import logging -import tempfile -from pathlib import Path -import random -import time -import base64 -import re -import io -import textwrap - -import requests -from dotenv import load_dotenv -from telegram import Update -from telegram.ext import ( - Application, - CommandHandler, - MessageHandler, - filters, - ContextTypes, -) - -# Load environment variables -load_dotenv() - -# Configure logging -logging.basicConfig( - format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", - level=logging.INFO, -) -logger = logging.getLogger(__name__) - -# Qwen API configuration -QWEN_API_KEY = os.getenv("QWEN_API_KEY") -QWEN_ENDPOINT = os.getenv("QWEN_ENDPOINT") -QWEN_MODEL = os.getenv("QWEN_VL_MODEL", "qwen-vl-plus") - -# Telegram bot token -TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN") - -# Allowed user IDs (empty list means all users allowed) -ALLOWED_USERS = [int(user.strip()) for user in os.getenv("ALLOWED_USERS", "").split(",") if user.strip()] - -# Configuration for image generation -MAX_RETRIES = 3 -RETRY_DELAY = 2 # seconds -CODE_EXECUTION_TIMEOUT = 180 - - -async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: - """Send a message when the command /start is issued.""" - user = update.effective_user - await update.message.reply_text( - f"Привет, {user.first_name}!\n\n" - "Я бот для работы с изображениями и создания визуальных материалов для отчётов.\n\n" - "📸 Отправь мне картинку - я опишу что на ней изображено\n" - "🎨 Используй /draw <описание> - я сгенерирую изображение или схему\n\n" - "📊 **Генерация схем и графиков для отчётов:**\n" - "/draw схема архитектуры системы\n" - "/draw график роста продаж за квартал\n" - "/draw блок-схема процесса одобрения кредита\n" - "/draw диаграмма распределения бюджета\n\n" - "Доступные команды:\n" - "/start - показать это сообщение\n" - "/help - показать справку\n" - "/model - показать текущую модель генерации\n" - "/draw <описание> - сгенерировать изображение" - ) - - -async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: - """Send a message when the command /help is issued.""" - help_text = ( - "📸 **Бот описания и генерации изображений**\n\n" - "Этот бот использует Qwen-VL для анализа изображений и Qwen + Python библиотеки для генерации чётких схем.\n\n" - "📋 **Команды:**\n" - "/start - начать работу\n" - "/help - показать эту справку\n" - "/model - показать текущую модель генерации\n" - "/draw <описание> - сгенерировать изображение\n\n" - "📊 **Генерация схем и графиков для отчётов:**\n\n" - "🔹 **Блок-схемы и алгоритмы:**\n" - "• `/draw блок-схема процесса одобрения кредита` - создаст flowchart\n" - "• `/draw алгоритм обработки заказа` - создаст блок-схему алгоритма\n" - "• `/draw процесс onboarding сотрудника` - создаст диаграмму процесса\n\n" - "🔹 **Архитектурные диаграммы:**\n" - "• `/draw архитектура системы микросервисов` - создаст архитектурную схему\n" - "• `/draw схема веб-приложения с базами данных` - создаст диаграмму\n" - "• `/draw архитектура cloud-инфраструктуры` - создаст схему инфраструктуры\n\n" - "🔹 **Графики данных:**\n" - "• `/draw график роста продаж за квартал` - создаст линейный график\n" - "• `/draw сравнение показателей отделов` - создаст столбчатую диаграмму\n" - "• `/draw распределение бюджета по статьям` - создаст круговую диаграмму\n" - "• `/draw корреляция между параметрами` - создаст точечную диаграмму\n\n" - "🔹 **Специализированные диаграммы:**\n" - "• `/draw ER-диаграмма базы данных пользователей` - создаст схему БД\n" - "• `/draw сеть серверов и их соединения` - создаст сетевую диаграмму\n" - "• `/draw топология локальной сети` - создаст схему сети\n\n" - "💡 **Как использовать:**\n" - "• Отправьте боту изображение для получения описания\n" - "• Используйте `/draw` для генерации изображения\n" - "• Для схем используйте ключевые слова: *блок-схема, архитектура, график, диаграмма*\n" - "• Изображения генерируются в профессиональном стиле для бизнес-отчётов\n\n" - "🔧 **Техническая информация:**\n" - "• Блок-схемы: Qwen + graphviz\n" - "• Архитектурные диаграммы: Qwen + matplotlib\n" - "• Графики: Qwen + matplotlib\n" - "• ER-диаграммы: Qwen + graphviz\n" - "• Сетевые диаграммы: Qwen + networkx" - ) - await update.message.reply_text(help_text, parse_mode="Markdown") - - -async def show_model(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: - """Show current image generation model configuration.""" - model_info = ( - "🔧 **Текущая конфигурация генерации изображений:**\n\n" - f"• **Qwen модель:** `{QWEN_MODEL}`\n" - f"• **Qwen API Endpoint:** `{QWEN_ENDPOINT}`\n\n" - "📊 **Генерация схем и графиков:**\n" - "Все схемы создаются через Qwen + Python библиотеки:\n" - "• matplotlib - для графиков и архитектурных диаграмм\n" - "• graphviz - для блок-схем и ER-диаграмм\n" - "• networkx - для сетевых диаграмм" - ) - await update.message.reply_text(model_info, parse_mode="Markdown") - - -async def describe_image(image_data: bytes) -> str: - headers = {"Authorization": f"Bearer {QWEN_API_KEY}", "Content-Type": "application/json"} - image_encoded = base64.b64encode(image_data).decode("utf-8") - image_uri = f"data:image/jpeg;base64,{image_encoded}" - - # Уникальная соль для обхода кеша - unique_salt = f" Уникальный идентификатор запроса: {int(time.time())}-{random.randint(1000,9999)}. Не учитывай это в ответе." - - payload = { - "model": QWEN_MODEL, - "messages": [ - { - "role": "user", - "content": [ - {"type": "image_url", "image_url": {"url": image_uri}}, - {"type": "text", "text": ( - "Ты — аналитик. Проанализируй КОНКРЕТНО ЭТО изображение. " - "Опиши то, что уникально для этой картинки: все текстовые надписи, цифры, стрелки, цвета, названия блоков. " - "Если видишь архитектурную диаграмму — перечисли фактические названия слоёв и компонентов, которые написаны на картинке. " - "Не используй общие фразы типа 'основные детали схемы' или 'это архитектурная диаграмма системы ', если на изображении нет точной такой надписи. " - "Заверши выводом: какие конкретные факты из этого изображения нужно записать в отчёт." - f"{unique_salt}" - )}, - ], - } - ], - "temperature": 0.9, - "top_p": 0.95, - } - - api_url = f"{QWEN_ENDPOINT}/chat/completions" - - try: - response = requests.post(api_url, headers=headers, json=payload, timeout=60) - - if response.status_code != 200: - logger.error(f"API error: {response.status_code} - {response.text}") - return f"Ошибка API: {response.status_code}. {response.text[:200]}" - - result = response.json() - description = result["choices"][0]["message"]["content"] - return description - except requests.exceptions.RequestException as e: - logger.error(f"API request failed: {e}") - return f"Ошибка при запросе к API: {str(e)}" - except (KeyError, IndexError) as e: - logger.error(f"Unexpected response format: {e}") - return "Ошибка при обработке ответа от API" - - -async def handle_photo(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: - """Handle photo messages.""" - # Check if user is allowed - if ALLOWED_USERS and update.effective_user.id not in ALLOWED_USERS: - await update.message.reply_text("У вас нет доступа к этому боту.") - return - - photo = update.message.photo[-1] # Get the highest resolution photo - user = update.effective_user - - await update.message.reply_text("🔍 Анализирую изображение...") - - try: - # Download the photo - file = await context.bot.get_file(photo.file_id) - - # Create a temporary file - with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as tmp_file: - tmp_path = tmp_file.name - await file.download_to_drive(tmp_path) - - try: - description = await describe_image_with_base64(tmp_path) - except Exception as e: - logger.warning(f"Base64 approach failed: {e}, trying URL approach") - description = await describe_image("placeholder") - - # Clean up temp file - Path(tmp_path).unlink(missing_ok=True) - - await update.message.reply_text(f"📝 Описание:\n\n{description}") - - except Exception as e: - logger.error(f"Error processing photo: {e}") - await update.message.reply_text(f"❌ Произошла ошибка при обработке изображения: {str(e)}") - - -async def describe_image_with_base64(image_path: str) -> str: - """Send image to Qwen-VL API using base64 encoding.""" - import base64 - - headers = { - "Authorization": f"Bearer {QWEN_API_KEY}", - "Content-Type": "application/json", - } - - # Read and encode image - with open(image_path, "rb") as f: - image_data = base64.b64encode(f.read()).decode("utf-8") - - # Create data URI - image_uri = f"data:image/jpeg;base64,{image_data}" - - payload = { - "model": QWEN_MODEL, - "messages": [ - { - "role": "user", - "content": [ - {"type": "image_url", "image_url": {"url": image_uri}}, - {"type": "text", "text": "Кратко опиши изображение на русском языке. Перечисли основные детали, каждая в 1-2 предложениях. Будь лаконичен."}, - ], - } - ], - } - - api_url = f"{QWEN_ENDPOINT}/chat/completions" - - response = requests.post(api_url, headers=headers, json=payload, timeout=120) - - if response.status_code != 200: - logger.error(f"API error: {response.status_code} - {response.text}") - raise Exception(f"API error: {response.status_code} - {response.text}") - - result = response.json() - return result["choices"][0]["message"]["content"] - - -async def draw_image(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: - """Handle /draw command to generate images using Qwen.""" - if not context.args: - await update.message.reply_text( - "Пожалуйста, укажите описание изображения:\n/draw <описание того что хотите увидеть>" - ) - return - - prompt = " ".join(context.args) - - # Check if this is a diagram/chart request - diagram_keywords = ["схема", "график", "диаграмма", "архитектура", "блок-схема", - "flowchart", "diagram", "architecture", "chart", "graph", - "pie chart", "bar chart", "line chart"] - is_diagram_request = any(keyword in prompt.lower() for keyword in diagram_keywords) - - if is_diagram_request: - await update.message.reply_text("📊 Генерирую схему/график через Qwen + Python библиотеки... Это может занять до 2 минут.") - try: - image_bytes = await generate_diagram_via_qwen(prompt) - if image_bytes: - # Add text overlay to the image - image_bytes = add_text_to_image(image_bytes, prompt) - await context.bot.send_photo( - chat_id=update.effective_chat.id, - photo=image_bytes - ) - else: - await update.message.reply_text("❌ Не удалось сгенерировать схему. Попробуйте переформулировать запрос.") - except Exception as e: - logger.error(f"Diagram generation error: {e}") - await update.message.reply_text(f"❌ Ошибка при генерации схемы: {str(e)}") - else: - await update.message.reply_text("🎨 Генерирую изображение через Qwen + matplotlib... Это может занять до 2 минут.") - try: - image_bytes = await generate_image_via_qwen(prompt) - if image_bytes: - # Add text overlay to the image - image_bytes = add_text_to_image(image_bytes, prompt) - await context.bot.send_photo( - chat_id=update.effective_chat.id, - photo=image_bytes - ) - else: - await update.message.reply_text("❌ Не удалось сгенерировать изображение. Попробуйте переформулировать запрос.") - except Exception as e: - logger.error(f"Image generation error: {e}") - await update.message.reply_text(f"❌ Ошибка при генерации: {str(e)}") - - -def detect_diagram_type(prompt: str) -> str: - """Detect the type of diagram requested.""" - prompt_lower = prompt.lower() - - # Flowchart / Block diagram - if any(kw in prompt_lower for kw in ["блок-схема", "flowchart", "алгоритм", "процесс", "этапы"]): - return "flowchart" - - # Architecture diagram - if any(kw in prompt_lower for kw in ["архитектура", "architecture", "микросервисы", "система", "слои"]): - return "architecture" - - # ER diagram - if any(kw in prompt_lower for kw in ["ер-диаграмма", "er-diagram", "база данных", "таблицы", "связи"]): - return "er_diagram" - - # Network diagram - if any(kw in prompt_lower for kw in ["сеть", "network", "узлы", "соединения", "топология"]): - return "network" - - # Line chart - if any(kw in prompt_lower for kw in ["линейный график", "line chart", "тренд", "динамика", "время"]): - return "line_chart" - - # Bar chart - if any(kw in prompt_lower for kw in ["столбчатая", "bar chart", "гистограмма", "сравнение"]): - return "bar_chart" - - # Pie chart - if any(kw in prompt_lower for kw in ["круговая", "pie chart", "доля", "процент", "распределение"]): - return "pie_chart" - - # Scatter plot - if any(kw in prompt_lower for kw in ["точечный", "scatter", "корреляция", "зависимость"]): - return "scatter" - - # Default to matplotlib generic - return "matplotlib_generic" - - -async def generate_diagram_via_qwen(prompt: str) -> bytes | None: - """Generate a diagram using Qwen to create code, then execute it.""" - if not QWEN_API_KEY: - logger.error("QWEN_API_KEY is not set") - return None - - logger.info(f"Generating diagram via Qwen for prompt: {prompt}") - - # Detect the type of diagram requested - diagram_type = detect_diagram_type(prompt) - logger.info(f"Detected diagram type: {diagram_type}") - - try: - # Route to appropriate generator based on diagram type - if diagram_type == "flowchart": - return await generate_flowchart(prompt) - elif diagram_type == "architecture": - return await generate_architecture_diagram(prompt) - elif diagram_type == "er_diagram": - return await generate_er_diagram(prompt) - elif diagram_type == "network": - return await generate_network_diagram(prompt) - elif diagram_type == "line_chart": - return await generate_line_chart(prompt) - elif diagram_type == "bar_chart": - return await generate_bar_chart(prompt) - elif diagram_type == "pie_chart": - return await generate_pie_chart(prompt) - elif diagram_type == "scatter": - return await generate_scatter_plot(prompt) - else: - # Default to generic matplotlib - return await generate_generic_diagram(prompt) - - except Exception as e: - logger.error(f"Diagram generation failed: {e}") - return None - - -async def generate_image_via_qwen(prompt: str) -> bytes | None: - """Generate a general image/illustration using Qwen + matplotlib.""" - if not QWEN_API_KEY: - logger.error("QWEN_API_KEY is not set") - return None - - system_prompt = """Ты — эксперт по визуализации данных и созданию профессиональных иллюстраций. Твоя задача — создавать чёткие, визуально привлекательные изображения с помощью Python matplotlib. - -ТРЕБОВАНИЯ К КАЧЕСТВУ ИЗОБРАЖЕНИЯ: -1. Размер: figsize=(14, 10) или больше для детализации -2. DPI: минимум 150 для высокого качества -3. Шрифты: используй DejaVu Sans для поддержки кириллицы -4. Цвета: используй профессиональную палитру (steelblue, coral, teal, orange, purple) -5. Линии: linewidth >= 2 для чёткости -6. Маркеры: markersize >= 8 для видимости - -ОБЯЗАТЕЛЬНЫЕ ЭЛЕМЕНТЫ: -- Заголовок с plt.title() - крупный, жирный шрифт (fontsize=16-20, fontweight='bold') -- Подписи осей с plt.xlabel() и plt.ylabel() - fontsize=12-14 -- Сетка с plt.grid(True, alpha=0.3) для лучшей читаемости -- Легенду если есть несколько серий данных -- Отступы: bbox_inches='tight' при сохранении - -ФОРМАТ КОДА: -1. Импорт: import matplotlib.pyplot as plt -2. Настройка шрифтов: plt.rcParams['font.family'] = 'DejaVu Sans' -3. Создание фигуры: plt.figure(figsize=(14, 10)) -4. Добавление элементов графика -5. Настройка заголовка и подписей -6. Сохранение: plt.savefig('image.png', dpi=150, bbox_inches='tight') -7. Закрытие: plt.close() - -Генерируй ТОЛЬКО Python код без объяснений.""" - - user_prompt = f"""Создай профессиональное визуальное изображение для следующего запроса: - -{prompt} - -Требования: -- Высокое качество (dpi=150, figsize>=12) -- Профессиональные цвета и стили -- Чёткие подписи на русском языке -- Заголовок и легенда при необходимости -- Сохраняй в 'image.png' - -Генерируй ТОЛЬКО Python код без объяснений.""" - - code = await get_code_from_qwen(system_prompt, user_prompt, temperature=0.4) - if code: - return execute_matplotlib_code(code) - return None - - -async def generate_generic_diagram(prompt: str) -> bytes | None: - """Generate a generic diagram using Qwen + matplotlib.""" - system_prompt = """Ты — эксперт по визуализации данных и созданию профессиональных схем. Твоя задача — создавать чёткие, визуально привлекательные схемы и графики с помощью Python matplotlib. - -ТРЕБОВАНИЯ К КАЧЕСТВУ: -1. Размер фигуры: figsize=(14, 10) или больше -2. DPI: минимум 150 для высокого качества -3. Шрифты: plt.rcParams['font.family'] = 'DejaVu Sans' для кириллицы -4. Цвета: профессиональная палитра (steelblue, coral, teal, orange, purple, darkgreen) -5. Линии: linewidth >= 2, alpha для прозрачности -6. Маркеры: markersize >= 8 для видимости - -ОБЯЗАТЕЛЬНЫЕ ЭЛЕМЕНТЫ: -- Заголовок: plt.title('Название', fontsize=18, fontweight='bold', pad=20) -- Подписи осей: fontsize=14, fontweight='semibold' -- Сетка: plt.grid(True, alpha=0.3, linestyle='--', linewidth=0.5) -- Легенда: если несколько серий данных -- Отступы: bbox_inches='tight' при сохранении - -СТРУКТУРА КОДА: -1. import matplotlib.pyplot as plt -2. plt.rcParams['font.family'] = 'DejaVu Sans' -3. fig, ax = plt.subplots(figsize=(14, 10)) -4. Добавление элементов графика -5. Настройка заголовка и подписей -6. plt.savefig('diagram.png', dpi=150, bbox_inches='tight') -7. plt.close() - -Генерируй ТОЛЬКО Python код без объяснений.""" - - user_prompt = f"""Создай профессиональную схему/график для следующего запроса: - -{prompt} - -Требования: -- Высокое качество (dpi=150, figsize>=12) -- Профессиональные цвета и стили -- Чёткие подписи на русском языке -- Заголовок и сетка для читаемости -- Сохраняй в 'diagram.png' - -Генерируй ТОЛЬКО Python код без объяснений.""" - - code = await get_code_from_qwen(system_prompt, user_prompt, temperature=0.4) - if code: - return execute_matplotlib_code(code) - return None - - -async def generate_flowchart(prompt: str) -> bytes | None: - """Generate a flowchart using Qwen + graphviz.""" - system_prompt = '''Ты — эксперт по созданию профессиональных блок-схем. Твоя задача — создавать чёткие, визуально привлекательные блок-схемы с помощью Python graphviz. - -ТРЕБОВАНИЯ К КАЧЕСТВУ: -1. Размер: используйте dot.attr(size='12,8!') для хорошего размера -2. DPI: format='png' с высоким разрешением -3. Направление: rankdir='TB' (сверху вниз) или rankdir='LR' (слева направо) -4. Стили узлов: используйте box, roundbox, circle, ellipse для разных типов шагов -5. Цвета: профессиональная палитра (lightblue, lightgreen, lightyellow, salmon) -6. Шрифты: fontsize=12-14 для читаемости - -СТРУКТУРА БЛОК-СХЕМЫ: -- Прямоугольник (box): для процессов/действий -- Ромб (diamond): для решений/условий -- Овал (ellipse/circle): для начала/конца -- Параллелограмм: для ввода/вывода - -ПРИМЕР КОДА: -from graphviz import Digraph -dot = Digraph("Flowchart", format="png") -dot.attr(rankdir="TB", size="12,8!", bgcolor="white") -dot.node("start", "Начало", shape="ellipse", style="filled", fillcolor="lightgreen") -dot.node("process1", "Процесс 1", shape="box", style="filled", fillcolor="lightblue") -dot.node("decision", "Условие?", shape="diamond", style="filled", fillcolor="lightyellow") -dot.edge("start", "process1") -dot.edge("process1", "decision") -dot.render("flowchart", cleanup=True) - -Генерируй ТОЛЬКО Python код без объяснений. Используй двойные кавычки для всех строк.''' - - user_prompt = f"""Создай профессиональную блок-схему для следующего запроса: - -{prompt} - -Требования: -- Используй правильные формы для разных типов шагов -- Добавь цвета для визуальной иерархии -- Чёткие подписи на русском языке -- Сохраняй в 'flowchart.png' - -Генерируй ТОЛЬКО Python код без объяснений. Используй двойные кавычки для всех строк.""" - - code = await get_code_from_qwen(system_prompt, user_prompt, temperature=0.4) - if code: - return execute_graphviz_code(code) - return None - - -async def generate_architecture_diagram(prompt: str) -> bytes | None: - """Generate an architecture diagram using Qwen + matplotlib.""" - system_prompt = """Ты — эксперт по созданию профессиональных архитектурных диаграмм. Твоя задача — создавать чёткие, визуально привлекательные архитектурные схемы с помощью Python matplotlib. - -ТРЕБОВАНИЯ К КАЧЕСТВУ: -1. Размер фигуры: figsize=(14, 10) или больше -2. DPI: минимум 150 для высокого качества -3. Шрифты: используй DejaVu Sans для кириллицы -4. Цвета: профессиональная палитра для разных слоёв: - - Frontend: lightblue (#E3F2FD) - - Backend/API: lightgreen (#E8F5E9) - - Database: lightyellow (#FFFDE7) - - External services: lightcoral (#FFEBEE) - - Cache: lightpurple (#F3E5F5) - -ЭЛЕМЕНТЫ АРХИТЕКТУРНОЙ ДИАГРАММЫ: -- Используй patches.FancyBboxPatch для компонентов -- Стрелки FancyArrowPatch для связей -- Текст с ha='center', va='center' для подписей - -ВАЖНО: Синтаксис FancyBboxPatch - все параметры передаются в конструктор: -patches.FancyBboxPatch(x, y, width, height, boxstyle="round,pad=0.05", facecolor="color", edgecolor="color", linewidth=2) - -ПРИМЕР КОДА: -import matplotlib.pyplot as plt -from matplotlib import patches -plt.rcParams["font.family"] = "DejaVu Sans" -fig, ax = plt.subplots(figsize=(14, 10)) -# Создаём компонент -rect = patches.FancyBboxPatch((0.1, 0.7), 0.3, 0.15, boxstyle="round,pad=0.05", facecolor="#E3F2FD", edgecolor="#1976D2", linewidth=2) -ax.add_patch(rect) -ax.text(0.25, 0.775, "Frontend", ha="center", va="center", fontsize=14, fontweight="bold") -# Создаём стрелку -arrow = patches.FancyArrowPatch((0.45, 0.75), (0.6, 0.55), arrowstyle="->", mutation_scale=20, linewidth=2, color="gray") -ax.add_patch(arrow) -ax.axis("off") -plt.title("Архитектура системы", fontsize=18, fontweight="bold", pad=20) -plt.savefig("architecture.png", dpi=150, bbox_inches="tight") -plt.close() - -Генерируй ТОЛЬКО Python код без объяснений. Используй двойные кавычки.""" - - user_prompt = f"""Создай профессиональную архитектурную диаграмму для следующего запроса: - -{prompt} - -Требования: -- Используй FancyBboxPatch для компонентов (все параметры в конструкторе) -- Добавь стрелки FancyArrowPatch для связей -- Цвета для разных слоёв архитектуры -- Чёткие подписи на русском языке -- Заголовок диаграммы -- Сохраняй в 'architecture.png' - -Генерируй ТОЛЬКО Python код без объяснений. Используй двойные кавычки.""" - - code = await get_code_from_qwen(system_prompt, user_prompt, temperature=0.4) - if code: - return execute_matplotlib_code(code) - return None - - -async def generate_er_diagram(prompt: str) -> bytes | None: - """Generate an ER diagram using Qwen + graphviz.""" - system_prompt = '''Ты — эксперт по созданию профессиональных ER-диаграмм баз данных. Твоя задача — создавать чёткие, визуально привлекательные ER-диаграммы с помощью Python graphviz. - -ТРЕБОВАНИЯ К КАЧЕСТВУ: -1. Размер: dot.attr(size="14,10!") для хорошего размера -2. Направление: rankdir="LR" (слева направо) для таблиц -3. Стили: используйте record или MRE для структуры таблиц -4. Цвета: lightblue для таблиц, lightgreen для связей -5. Шрифты: fontsize=11-13 для читаемости - -СТРУКТУРА ТАБЛИЦЫ: -- Используйте subgraph для группировки -- Заголовок таблицы жирным -- Атрибуты с типами данных -- PK/FK обозначения - -ПРИМЕР КОДА: -from graphviz import Digraph -dot = Digraph("ER_Diagram", format="png") -dot.attr(rankdir="LR", size="14,10!", bgcolor="white") -dot.attr("node", shape="record", fontsize="12") -# Таблица Users -with dot.subgraph(name="cluster_0") as c: - c.attr(label="Users", style="filled", color="lightblue", fillcolor="lightblue") - c.node("users", "{Users| id: INT PK| name: VARCHAR| email: VARCHAR}") -# Таблица Orders -with dot.subgraph(name="cluster_1") as c: - c.attr(label="Orders", style="filled", color="lightblue", fillcolor="lightblue") - c.node("orders", "{Orders| id: INT PK| user_id: INT FK| total: DECIMAL}") -# Связь -dot.edge("users:name", "orders:user_id", label="1:N", fontsize="10") -dot.render("er_diagram", cleanup=True) - -Генерируй ТОЛЬКО Python код без объяснений. Используй двойные кавычки для всех строк.''' - - user_prompt = f"""Создай профессиональную ER-диаграмму базы данных для следующего запроса: - -{prompt} - -Требования: -- Используй record shape для таблиц -- Покажи все атрибуты с типами данных -- Обозначь PK и FK -- Добавь связи с типами (1:1, 1:N, N:M) -- Цвета для визуальной иерархии -- Сохраняй в 'er_diagram.png' - -Генерируй ТОЛЬКО Python код без объяснений. Используй двойные кавычки для всех строк.""" - - code = await get_code_from_qwen(system_prompt, user_prompt, temperature=0.4) - if code: - return execute_graphviz_code(code) - return None - - -async def generate_network_diagram(prompt: str) -> bytes | None: - """Generate a network diagram using Qwen + networkx.""" - system_prompt = """Ты — эксперт по созданию профессиональных сетевых диаграмм. Твоя задача — создавать чёткие, визуально привлекательные сетевые схемы с помощью Python networkx и matplotlib. - -ТРЕБОВАНИЯ К КАЧЕСТВУ: -1. Размер фигуры: figsize=(14, 10) или больше -2. DPI: минимум 150 для высокого качества -3. Шрифты: используй DejaVu Sans для кириллицы -4. Узлы: разные цвета и размеры для разных типов устройств -5. Расположение: используй nx.spring_layout или nx.kamada_kawai_layout -6. Стрелки: directed графы для направленных связей - -ТИПЫ УЗЛОВ И ЦВЕТА: -- Router/Gateway: красный (#D32F2F), размер 800 -- Server: синий (#1976D2), размер 600 -- Switch: зелёный (#388E3C), размер 500 -- Client/PC: оранжевый (#F57C00), размер 400 -- Cloud: фиолетовый (#7B1FA2), размер 700 - -ПРИМЕР КОДА: -import networkx as nx -import matplotlib.pyplot as plt -plt.rcParams["font.family"] = "DejaVu Sans" -G = nx.DiGraph() -G.add_node("Интернет", type="cloud") -G.add_node("Router", type="router") -G.add_node("Switch", type="switch") -G.add_node("Server1", type="server") -G.add_node("PC1", type="client") -G.add_edge("Интернет", "Router") -G.add_edge("Router", "Switch") -G.add_edge("Switch", "Server1") -G.add_edge("Switch", "PC1") -pos = nx.kamada_kawai_layout(G) -node_colors = {"cloud": "#7B1FA2", "router": "#D32F2F", "switch": "#388E3C", "server": "#1976D2", "client": "#F57C00"} -sizes = {"cloud": 700, "router": 800, "switch": 500, "server": 600, "client": 400} -colors = [node_colors[G.nodes[n]["type"]] for n in G.nodes()] -sizes_list = [sizes[G.nodes[n]["type"]] for n in G.nodes()] -plt.figure(figsize=(14, 10)) -nx.draw(G, pos, with_labels=True, node_color=colors, node_size=sizes_list, font_size=12, font_weight="bold", edge_color="gray", arrowsize=20) -plt.title("Сетевая топология", fontsize=18, fontweight="bold") -plt.savefig("network.png", dpi=150, bbox_inches="tight") -plt.close() - -Генерируй ТОЛЬКО Python код без объяснений. Используй двойные кавычки для всех строк.""" - - user_prompt = f"""Создай профессиональную сетевую диаграмму для следующего запроса: - -{prompt} - -Требования: -- Используй networkx для графа -- Разные цвета для разных типов устройств -- Чёткие подписи на русском -- Стрелки для направленных связей -- Заголовок диаграммы -- Сохраняй в 'network.png' - -Генерируй ТОЛЬКО Python код без объяснений. Используй двойные кавычки для всех строк.""" - - code = await get_code_from_qwen(system_prompt, user_prompt, temperature=0.4) - if code: - return execute_networkx_code(code) - return None - - -async def generate_line_chart(prompt: str) -> bytes | None: - """Generate a line chart using Qwen + matplotlib.""" - system_prompt = """Ты — эксперт по визуализации данных. Твоя задача — создавать профессиональные линейные графики с помощью Python matplotlib. - -ТРЕБОВАНИЯ К КАЧЕСТВУ: -1. Размер фигуры: figsize=(14, 8) или больше -2. DPI: минимум 150 для высокого качества -3. Шрифты: plt.rcParams['font.family'] = 'DejaVu Sans' для кириллицы -4. Линии: linewidth >= 2, разные цвета для каждой линии -5. Маркеры: marker='o', markersize >= 8 для видимости точек -6. Сетка: plt.grid(True, alpha=0.3, linestyle='--') - -ОБЯЗАТЕЛЬНЫЕ ЭЛЕМЕНТЫ: -- Заголовок: plt.title('Название', fontsize=18, fontweight='bold') -- Подписи осей: plt.xlabel(), plt.ylabel() с fontsize=14 -- Легенда: plt.legend() с описанием линий -- Форматирование осей при необходимости - -ПРИМЕР КОДА: -import matplotlib.pyplot as plt -import numpy as np -plt.rcParams['font.family'] = 'DejaVu Sans' -x = np.array([1, 2, 3, 4, 5]) -y1 = np.array([10, 15, 13, 17, 20]) -y2 = np.array([8, 12, 14, 11, 16]) -plt.figure(figsize=(14, 8)) -plt.plot(x, y1, marker='o', linewidth=2, markersize=8, label='Серия 1', color='steelblue') -plt.plot(x, y2, marker='s', linewidth=2, markersize=8, label='Серия 2', color='coral') -plt.xlabel('Период', fontsize=14) -plt.ylabel('Значение', fontsize=14) -plt.title('Линейный график', fontsize=18, fontweight='bold') -plt.grid(True, alpha=0.3, linestyle='--') -plt.legend(fontsize=12) -plt.savefig('line_chart.png', dpi=150, bbox_inches='tight') -plt.close() - -Генерируй ТОЛЬКО Python код без объяснений.""" - - user_prompt = f"""Создай профессиональный линейный график для следующего запроса: - -{prompt} - -Требования: -- Чёткие линии с маркерами -- Профессиональные цвета -- Подписи осей и заголовок на русском -- Сетка для читаемости -- Легенда если несколько линий -- Сохраняй в 'line_chart.png' - -Генерируй ТОЛЬКО Python код без объяснений.""" - - code = await get_code_from_qwen(system_prompt, user_prompt, temperature=0.4) - if code: - return execute_matplotlib_code(code) - return None - - -async def generate_bar_chart(prompt: str) -> bytes | None: - """Generate a bar chart using Qwen + matplotlib.""" - system_prompt = """Ты — эксперт по визуализации данных. Твоя задача — создавать профессиональные столбчатые диаграммы с помощью Python matplotlib. - -ТРЕБОВАНИЯ К КАЧЕСТВУ: -1. Размер фигуры: figsize=(14, 8) или больше -2. DPI: минимум 150 для высокого качества -3. Шрифты: plt.rcParams['font.family'] = 'DejaVu Sans' для кириллицы -4. Столбцы: ширина 0.6-0.8, профессиональные цвета -5. Подписи значений: добавь plt.bar_label() для точных значений -6. Отступы: bbox_inches='tight' при сохранении - -ОБЯЗАТЕЛЬНЫЕ ЭЛЕМЕНТЫ: -- Заголовок: plt.title('Название', fontsize=18, fontweight='bold') -- Подписи осей: fontsize=14 -- Сетка по оси Y: plt.grid(True, axis='y', alpha=0.3) -- Цвета: используй градиент или контрастные цвета - -ПРИМЕР КОДА: -import matplotlib.pyplot as plt -plt.rcParams['font.family'] = 'DejaVu Sans' -categories = ['Январь', 'Февраль', 'Март', 'Апрель', 'Май'] -values = [25, 40, 30, 45, 35] -colors = ['#1976D2', '#388E3C', '#F57C00', '#D32F2F', '#7B1FA2'] -plt.figure(figsize=(14, 8)) -bars = plt.bar(categories, values, color=colors, edgecolor='white', linewidth=1.5) -plt.xlabel('Месяц', fontsize=14) -plt.ylabel('Значение', fontsize=14) -plt.title('Столбчатая диаграмма', fontsize=18, fontweight='bold') -plt.grid(True, axis='y', alpha=0.3, linestyle='--') -for bar in bars: - height = bar.get_height() - plt.text(bar.get_x() + bar.get_width()/2., height, f'{height}', ha='center', va='bottom', fontsize=12) -plt.savefig('bar_chart.png', dpi=150, bbox_inches='tight') -plt.close() - -Генерируй ТОЛЬКО Python код без объяснений.""" - - user_prompt = f"""Создай профессиональную столбчатую диаграмму для следующего запроса: - -{prompt} - -Требования: -- Чёткие столбцы с подписями значений -- Профессиональные цвета -- Подписи осей и заголовок на русском -- Сетка по оси Y -- Сохраняй в 'bar_chart.png' - -Генерируй ТОЛЬКО Python код без объяснений.""" - - code = await get_code_from_qwen(system_prompt, user_prompt, temperature=0.4) - if code: - return execute_matplotlib_code(code) - return None - - -async def generate_pie_chart(prompt: str) -> bytes | None: - """Generate a pie chart using Qwen + matplotlib.""" - system_prompt = """Ты — эксперт по визуализации данных. Твоя задача — создавать профессиональные круговые диаграммы с помощью Python matplotlib. - -ТРЕБОВАНИЯ К КАЧЕСТВУ: -1. Размер фигуры: figsize=(12, 12) для хорошей детализации -2. DPI: минимум 150 для высокого качества -3. Шрифты: plt.rcParams['font.family'] = 'DejaVu Sans' для кириллицы -4. Сектора: autopct='%1.1f%%', pctdistance=0.85 -5. Взрыв: explode для выделения важных секторов -6. Тень: shadow=True для объёма - -ОБЯЗАТЕЛЬНЫЕ ЭЛЕМЕНТЫ: -- Заголовок: plt.title('Название', fontsize=18, fontweight='bold', pad=20) -- Процентные метки: autopct с форматированием -- Легенда: plt.legend() с описанием секторов -- startangle=90 для правильного начала - -ПРИМЕР КОДА: -import matplotlib.pyplot as plt -plt.rcParams['font.family'] = 'DejaVu Sans' -labels = ['Отдел A', 'Отдел B', 'Отдел C', 'Отдел D'] -sizes = [35, 25, 20, 20] -colors = ['#1976D2', '#388E3C', '#F57C00', '#D32F2F'] -explode = (0.05, 0, 0, 0) # Выделяем первый сектор -plt.figure(figsize=(12, 12)) -wedges, texts, autotexts = plt.pie(sizes, explode=explode, labels=labels, colors=colors, - autopct='%1.1f%%', startangle=90, shadow=True, - textprops={'fontsize': 14, 'family': 'DejaVu Sans'}) -plt.setp(autotexts, size=12, weight='bold') -plt.title('Распределение по отделам', fontsize=18, fontweight='bold', pad=20) -plt.legend(wedges, labels, loc='center left', bbox_to_anchor=(1, 0.5), fontsize=12) -plt.savefig('pie_chart.png', dpi=150, bbox_inches='tight') -plt.close() - -Генерируй ТОЛЬКО Python код без объяснений.""" - - user_prompt = f"""Создай профессиональную круговую диаграмму для следующего запроса: - -{prompt} - -Требования: -- Чёткие секторы с процентами -- Профессиональные цвета -- Подписи и заголовок на русском -- Легенда для описания секторов -- Тень для объёма -- Сохраняй в 'pie_chart.png' - -Генерируй ТОЛЬКО Python код без объяснений.""" - - code = await get_code_from_qwen(system_prompt, user_prompt, temperature=0.4) - if code: - return execute_matplotlib_code(code) - return None - - -async def generate_scatter_plot(prompt: str) -> bytes | None: - """Generate a scatter plot using Qwen + matplotlib.""" - system_prompt = """Ты — эксперт по визуализации данных. Твоя задача — создавать профессиональные точечные диаграммы с помощью Python matplotlib. - -ТРЕБОВАНИЯ К КАЧЕСТВУ: -1. Размер фигуры: figsize=(14, 10) или больше -2. DPI: минимум 150 для высокого качества -3. Шрифты: plt.rcParams['font.family'] = 'DejaVu Sans' для кириллицы -4. Точки: s=100-200 для размера, alpha=0.6-0.8 для прозрачности -5. Цвета: профессиональная палитра с градиентом при необходимости -6. Линии тренда: добавь np.polyfit() для корреляции - -ОБЯЗАТЕЛЬНЫЕ ЭЛЕМЕНТЫ: -- Заголовок: plt.title('Название', fontsize=18, fontweight='bold') -- Подписи осей: fontsize=14 с указанием единиц измерения -- Сетка: plt.grid(True, alpha=0.3, linestyle='--') -- Легенда: если несколько серий данных - -ПРИМЕР КОДА: -import matplotlib.pyplot as plt -import numpy as np -plt.rcParams['font.family'] = 'DejaVu Sans' -np.random.seed(42) -x = np.random.rand(50) * 100 -y = 0.5 * x + np.random.randn(50) * 10 -plt.figure(figsize=(14, 10)) -plt.scatter(x, y, alpha=0.6, s=150, c='steelblue', edgecolors='white', linewidth=1) -# Линия тренда -z = np.polyfit(x, y, 1) -p = np.poly1d(z) -plt.plot(x, p(x), "r--", linewidth=2, label=f'Тренд: y={z[0]:.2f}x+{z[1]:.2f}') -plt.xlabel('Параметр X', fontsize=14) -plt.ylabel('Параметр Y', fontsize=14) -plt.title('Точечная диаграмма корреляции', fontsize=18, fontweight='bold') -plt.grid(True, alpha=0.3, linestyle='--') -plt.legend(fontsize=12) -plt.savefig('scatter.png', dpi=150, bbox_inches='tight') -plt.close() - -Генерируй ТОЛЬКО Python код без объяснений.""" - - user_prompt = f"""Создай профессиональную точечную диаграмму для следующего запроса: - -{prompt} - -Требования: -- Чёткие точки с прозрачностью -- Профессиональные цвета -- Подписи осей и заголовок на русском -- Сетка для читаемости -- Линия тренда при необходимости -- Сохраняй в 'scatter.png' - -Генерируй ТОЛЬКО Python код без объяснений.""" - - code = await get_code_from_qwen(system_prompt, user_prompt, temperature=0.4) - if code: - return execute_matplotlib_code(code) - return None - - -async def get_code_from_qwen(system_prompt: str, user_prompt: str, temperature: float = 0.3, max_retries: int = MAX_RETRIES) -> str | None: - """Get code from Qwen API with retry logic.""" - headers = {"Authorization": f"Bearer {QWEN_API_KEY}", "Content-Type": "application/json"} - - # Add unique timestamp to bypass any caching - unique_id = f"{int(time.time() * 1000)}-{random.randint(10000, 99999)}" - - payload = { - "model": QWEN_MODEL, - "messages": [ - {"role": "system", "content": system_prompt}, - {"role": "user", "content": f"{user_prompt}\n\n[Unique request ID: {unique_id}. Это не часть кода, просто идентификатор запроса.]"} - ], - "temperature": temperature, - "max_tokens": 4096, - } - - api_url = f"{QWEN_ENDPOINT}/chat/completions" - - for attempt in range(max_retries): - try: - response = requests.post(api_url, headers=headers, json=payload, timeout=120) - - if response.status_code != 200: - logger.error(f"Qwen API error (attempt {attempt + 1}/{max_retries}): {response.status_code} - {response.text[:200]}") - if attempt < max_retries - 1: - time.sleep(RETRY_DELAY * (attempt + 1)) - continue - return None - - result = response.json() - code = result["choices"][0]["message"]["content"] - code = extract_code_from_markdown(code) - - # Validate that we got actual code - if not code or len(code.strip()) < 20: - logger.warning(f"Generated code too short, retrying...") - if attempt < max_retries - 1: - time.sleep(RETRY_DELAY) - continue - return None - - logger.info(f"Generated code (attempt {attempt + 1}):\n{code[:300]}...") - return code - - except requests.exceptions.Timeout: - logger.warning(f"Request timeout (attempt {attempt + 1}/{max_retries})") - if attempt < max_retries - 1: - time.sleep(RETRY_DELAY * (attempt + 1)) - continue - return None - except Exception as e: - logger.error(f"Qwen API call failed (attempt {attempt + 1}/{max_retries}): {e}") - if attempt < max_retries - 1: - time.sleep(RETRY_DELAY * (attempt + 1)) - continue - return None - - return None - - -def extract_code_from_markdown(code: str) -> str: - """Extract Python code from markdown code blocks.""" - # Look for ```python ... ``` or ``` ... ``` blocks - pattern = r'```(?:python)?\s*\n(.*?)```' - match = re.search(pattern, code, re.DOTALL) - if match: - return match.group(1).strip() - return code.strip() - - -def execute_matplotlib_code(code: str) -> bytes | None: - """Execute matplotlib code and return the image as bytes.""" - import matplotlib - matplotlib.use('Agg') # Use non-interactive backend - import matplotlib.pyplot as plt - from matplotlib import patches - - # Create a temporary file for the output - tmp_path = tempfile.mktemp(suffix='.png') - - try: - # Modify the code to save to our temp path - # Replace various filenames with our temp path - code = code.replace("diagram.png", tmp_path) - code = code.replace("image.png", tmp_path) - code = code.replace("'diagram.png'", f"'{tmp_path}'") - code = code.replace('"diagram.png"', f'"{tmp_path}"') - code = code.replace("'image.png'", f"'{tmp_path}'") - code = code.replace('"image.png"', f'"{tmp_path}"') - - # Add plt.close() at the end to prevent memory issues - if "plt.savefig" in code and "plt.close()" not in code: - code += "\nplt.close()" - - # Execute the code in a restricted namespace - exec_globals = { - '__builtins__': __builtins__, - 'matplotlib': matplotlib, - 'plt': plt, - 'patches': patches, - } - - exec(code, exec_globals) - - # Read the generated image - if os.path.exists(tmp_path): - with open(tmp_path, 'rb') as f: - image_bytes = f.read() - return image_bytes - else: - logger.error("Image file was not created") - return None - - except Exception as e: - logger.error(f"Matplotlib code execution failed: {e}") - # Try to create a simple fallback diagram - return create_fallback_diagram(str(e)) - finally: - # Clean up temp file - if os.path.exists(tmp_path): - os.remove(tmp_path) - - -def execute_graphviz_code(code: str) -> bytes | None: - """Execute graphviz code and return the image as bytes.""" - from graphviz import Digraph - - tmp_path = tempfile.mktemp(suffix='.png') - base_path = tmp_path.rsplit('.', 1)[0] - - try: - # Execute the code in a restricted namespace - exec_globals = { - '__builtins__': __builtins__, - 'Digraph': Digraph, - } - - # Replace render calls to use our temp path - code = re.sub(r"dot\.render\(['\"]\w+['\"]", f"dot.render('{base_path}'", code) - code = re.sub(r"dot\.render\(['\"]\w+['\"],", f"dot.render('{base_path}',", code) - - exec(code, exec_globals) - - # Read the generated image - output_path = f"{base_path}.png" - if os.path.exists(output_path): - with open(output_path, 'rb') as f: - image_bytes = f.read() - os.remove(output_path) - return image_bytes - else: - logger.error("Image file was not created") - return None - - except Exception as e: - logger.error(f"Graphviz code execution failed: {e}") - return create_fallback_diagram(str(e)) - finally: - # Clean up temp files - for ext in ['', '.dot', '.png']: - path = f"{base_path}{ext}" - if os.path.exists(path): - try: - os.remove(path) - except: - pass - - -def execute_networkx_code(code: str) -> bytes | None: - """Execute networkx code and return the image as bytes.""" - import matplotlib - matplotlib.use('Agg') - import matplotlib.pyplot as plt - import networkx as nx - - tmp_path = tempfile.mktemp(suffix='.png') - - try: - # Execute the code in a restricted namespace - exec_globals = { - '__builtins__': __builtins__, - 'matplotlib': matplotlib, - 'plt': plt, - 'nx': nx, - } - - # Replace savefig calls to use our temp path - code = re.sub(r"plt\.savefig\(['\"]\w+\.png['\"]", f"plt.savefig('{tmp_path}'", code) - code = re.sub(r"plt\.savefig\(['\"]\w+\.png['\"],", f"plt.savefig('{tmp_path}',", code) - - if "plt.savefig" not in code: - code += f"\nplt.savefig('{tmp_path}', dpi=150, bbox_inches='tight')" - - if "plt.close()" not in code: - code += "\nplt.close()" - - exec(code, exec_globals) - - # Read the generated image - if os.path.exists(tmp_path): - with open(tmp_path, 'rb') as f: - image_bytes = f.read() - return image_bytes - else: - logger.error("Image file was not created") - return None - - except Exception as e: - logger.error(f"NetworkX code execution failed: {e}") - return create_fallback_diagram(str(e)) - finally: - # Clean up temp file - if os.path.exists(tmp_path): - try: - os.remove(tmp_path) - except: - pass - - -def add_text_to_image(image_bytes: bytes, text: str) -> bytes: - """Return image bytes unchanged - text overlay removed.""" - # Simply return the original image without any modifications - return image_bytes - - -def create_fallback_diagram(error: str) -> bytes | None: - """Create a professional fallback diagram when code execution fails.""" - import matplotlib - matplotlib.use('Agg') - import matplotlib.pyplot as plt - - fig, ax = plt.subplots(figsize=(12, 8)) - - # Градиентный фон - ax.set_facecolor('#F5F5F5') - fig.patch.set_facecolor('#F5F5F5') - - # Иконка ошибки - ax.text(0.5, 0.85, '⚠️', ha='center', fontsize=60, va='center') - - # Заголовок ошибки - ax.text(0.5, 0.72, 'Не удалось сгенерировать схему', - ha='center', fontsize=18, fontweight='bold', va='center', color='#1976D2') - - # Рекомендация - ax.text(0.5, 0.62, 'Попробуйте переформулировать запрос или использовать более простое описание', - ha='center', fontsize=14, va='center', color='#424242') - - # Детали ошибки в рамке - error_text = f'Детали ошибки:\n{error[:200]}' if error else 'Неизвестная ошибка' - ax.text(0.5, 0.45, error_text, - ha='center', fontsize=11, va='center', color='#757575', - bbox=dict(boxstyle='round', facecolor='white', edgecolor='#BDBDBD', alpha=0.8)) - - # Подсказка - ax.text(0.5, 0.25, '💡 Подсказка: Используйте ключевые слова "схема", "график", "диаграмма"', - ha='center', fontsize=12, va='center', color='#388E3C', fontweight='semibold') - - ax.axis('off') - - buf = io.BytesIO() - try: - fig.savefig(buf, format='png', dpi=150, bbox_inches='tight', - facecolor=fig.get_facecolor(), edgecolor='none') - buf.seek(0) - return buf.read() - except Exception as e: - logger.error(f"Failed to create fallback diagram: {e}") - return None - finally: - plt.close(fig) - - -def main() -> None: - """Start the bot.""" - if not TELEGRAM_BOT_TOKEN: - logger.error("TELEGRAM_BOT_TOKEN not set. Please set it in .env file.") - print("❌ Ошибка: TELEGRAM_BOT_TOKEN не установлен.") - print("\nПолучите токен у @BotFather в Telegram:") - print("1. Напишите @BotFather") - print("2. Отправьте команду /newbot") - print("3. Следуйте инструкциям и скопируйте токен") - print("\nДобавьте токен в файл .env:") - print("TELEGRAM_BOT_TOKEN=ваш_токен_от_BotFather") - return - - # Create application with increased timeout configuration - application = Application.builder() \ - .token(TELEGRAM_BOT_TOKEN) \ - .connect_timeout(60) \ - .read_timeout(60) \ - .write_timeout(60) \ - .pool_timeout(60) \ - .build() - - # Add handlers - application.add_handler(CommandHandler("start", start)) - application.add_handler(CommandHandler("help", help_command)) - application.add_handler(CommandHandler("model", show_model)) - application.add_handler(CommandHandler("draw", draw_image)) - application.add_handler( - MessageHandler(filters.PHOTO, handle_photo), - ) - - # Start the bot - logger.info("Бот запущен...") - print("🤖 Бот запущен! Ожидание сообщений...") - application.run_polling(allowed_updates=Update.ALL_TYPES) - - -if __name__ == "__main__": - main() \ No newline at end of file