1272 lines
No EOL
58 KiB
Python
1272 lines
No EOL
58 KiB
Python
#!/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> id: INT PK|<name> name: VARCHAR|<email> 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|<oid> id: INT PK|<user_id> user_id: INT FK|<total> 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() |