From 96858d63649452c1c0a3bb165787a778098c464f Mon Sep 17 00:00:00 2001 From: dropboy27 Date: Wed, 1 Apr 2026 00:37:54 +0300 Subject: [PATCH 1/5] =?UTF-8?q?=D0=BD=D0=B5=20=D1=80=D0=B0=D0=B1=D0=BE?= =?UTF-8?q?=D1=82=D0=B0=D0=B5=D1=82=20=D0=BF=D0=BE=D0=BA=D0=B0(?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 82 +++++++++++++++- requirements.txt | 3 + src/image_bot.py | 241 +++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 324 insertions(+), 2 deletions(-) create mode 100644 requirements.txt create mode 100644 src/image_bot.py diff --git a/README.md b/README.md index 9611f7f..658a55d 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,81 @@ -# b2b_assistants +# Matrix Image Recognition Bot -Репозиторий для разработки B to B рушений \ No newline at end of file +Бот для Matrix, который распознаёт изображения с помощью Qwen-VL. + +## Описание + +Этот бот подключается к Matrix серверу, слушает изображения в разрешённых комнатах и отправляет их в Qwen-VL API для анализа. Бот возвращает описание изображения обратно в чат. + +## Требования + +- Python 3.10+ +- Доступ к Matrix серверу +- API ключ Qwen-VL + +## Установка + +1. Клонируйте репозиторий: +```bash +git clone +cd b2b_assistants +``` + +2. Создайте виртуальное окружение: +```bash +python -m venv venv +source venv/bin/activate # Linux/Mac +# или +venv\Scripts\activate # Windows +``` + +3. Установите зависимости: +```bash +pip install -r requirements.txt +``` + +4. Настройте переменные окружения в файле `.env`: + +```env +# Matrix подключение +HOMESERVER=https://matrix.lambda.coredump.ru +MATRIX_USERNAME=@image_bot:matrix.lambda.coredump.ru +PASSWORD=ваш_пароль_бота +ALLOWED_ROOMS=!QcPkdLDWqDegdtDnpP:matrix.lambda.coredump.ru + +# Qwen API +QWEN_API_KEY=sk-L6oRP0m15Z9YquluktS6w +QWEN_ENDPOINT=https://llm.lambda.coredump.ru/v1 +QWEN_VL_MODEL=qwen-vl-plus +``` + +## Запуск + +```bash +python src/image_bot.py +``` + +## Команды бота + +- `/help` - показать справку по командам +- `/status` - показать статус бота + +## Использование + +1. Пригласите бота в комнату Matrix +2. Отправьте изображение в чат +3. Бот автоматически проанализирует изображение и вернёт описание + +## Структура проекта + +``` +b2b_assistants/ +├── .env # Переменные окружения +├── requirements.txt # Зависимости Python +├── README.md # Документация +└── src/ + └── image_bot.py # Основной код бота +``` + +## Лицензия + +MIT \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..d044578 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +matrix-nio>=0.25.0 +httpx>=0.24.0 +python-dotenv>=1.0.0 \ No newline at end of file diff --git a/src/image_bot.py b/src/image_bot.py new file mode 100644 index 0000000..2f542f3 --- /dev/null +++ b/src/image_bot.py @@ -0,0 +1,241 @@ +#!/usr/bin/env python3 +import asyncio +import os +import base64 +import logging +from typing import Optional +from dataclasses import dataclass + +import httpx +from nio import ( + AsyncClient, + RoomMessageImage, + RoomMessageText, + LoginResponse, + JoinResponse, +) +from dotenv import load_dotenv + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" +) +logger = logging.getLogger(__name__) + + +@dataclass +class BotConfig: + homeserver: str + username: str + password: str + allowed_rooms: list[str] + qwen_api_key: str + qwen_endpoint: str + qwen_model: str + + +def load_config() -> BotConfig: + load_dotenv() + allowed_rooms_str = os.getenv("ALLOWED_ROOMS", "") + allowed_rooms = [r.strip() for r in allowed_rooms_str.split(",") if r.strip()] + return BotConfig( + homeserver=os.getenv("HOMESERVER", "https://matrix.lambda.coredump.ru"), + username=os.getenv("MATRIX_USERNAME", ""), + password=os.getenv("PASSWORD", ""), + allowed_rooms=allowed_rooms, + qwen_api_key=os.getenv("QWEN_API_KEY", ""), + qwen_endpoint=os.getenv("QWEN_ENDPOINT", "https://llm.lambda.coredump.ru/v1"), + qwen_model=os.getenv("QWEN_VL_MODEL", "qwen-vl-plus"), + ) + + +class QwenVLClient: + def __init__(self, api_key: str, endpoint: str, model: str): + self.api_key = api_key + self.endpoint = endpoint + self.model = model + self.client = httpx.AsyncClient(timeout=120.0) + + async def analyze_image(self, image_data: bytes, prompt: str = "Опиши это изображение подробно:") -> str: + image_base64 = base64.b64encode(image_data).decode("utf-8") + image_url = f"data:image/jpeg;base64,{image_base64}" + + payload = { + "model": self.model, + "messages": [ + { + "role": "user", + "content": [ + {"type": "image_url", "image_url": {"url": image_url}}, + {"type": "text", "text": prompt} + ] + } + ] + } + + headers = {"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"} + + try: + response = await self.client.post( + f"{self.endpoint}/chat/completions", + json=payload, + headers=headers + ) + response.raise_for_status() + result = response.json() + return result["choices"][0]["message"]["content"] + except httpx.HTTPError as e: + logger.error(f"Error calling Qwen-VL API: {e}") + if e.response: + logger.error(f"Response: {e.response.text}") + return f"Ошибка при анализе изображения: {str(e)}" + + async def close(self): + await self.client.aclose() + + +class ImageBot: + def __init__(self, config: BotConfig): + self.config = config + self.client = AsyncClient(config.homeserver, config.username) # исправлено + self.qwen_client = QwenVLClient( + config.qwen_api_key, + config.qwen_endpoint, + config.qwen_model + ) + self.default_prompt = "Опиши это изображение подробно." + + async def login(self) -> bool: + try: + response = await self.client.login( + password=self.config.password, + device_name="image_recognition_bot" + ) + if isinstance(response, LoginResponse): + logger.info(f"Logged in as {self.config.username}") + return True + else: + logger.error(f"Login failed: {response}") + return False + except Exception as e: + logger.error(f"Login error: {e}") + return False + + async def join_rooms(self) -> None: + for room_id in self.config.allowed_rooms: + try: + response = await self.client.join(room_id) + if isinstance(response, JoinResponse): + logger.info(f"Joined room: {room_id}") + else: + logger.warning(f"Could not join room {room_id}: {response}") + except Exception as e: + logger.error(f"Error joining room {room_id}: {e}") + + async def download_image(self, url: str) -> Optional[bytes]: + try: + logger.info(f"Downloading image from {url}") + response = await self.client.download(url) # правильный метод + if response and hasattr(response, 'body'): + logger.info(f"Downloaded {len(response.body)} bytes") + return response.body + return None + except Exception as e: + logger.error(f"Download error: {e}") + return None + + async def send_response(self, room_id: str, message: str) -> None: + try: + await self.client.room_send( + room_id=room_id, + message_type="m.room.message", + content={"msgtype": "m.text", "body": message} + ) + except Exception as e: + logger.error(f"Error sending response: {e}") + + async def handle_image_message(self, room_id: str, event: RoomMessageImage) -> None: + if room_id not in self.config.allowed_rooms: + logger.warning(f"Ignoring image from non-allowed room {room_id}") + return + + logger.info(f"Processing image in {room_id}") + image_url = event.url # прямой доступ к URL + if not image_url: + await self.send_response(room_id, "Не удалось получить изображение.") + return + + image_data = await self.download_image(image_url) + if not image_data: + await self.send_response(room_id, "Не удалось загрузить изображение.") + return + + try: + await self.client.typing(room_id, timeout=4000) + except Exception: + pass + + await self.send_response(room_id, "🔍 Анализирую изображение...") + result = await self.qwen_client.analyze_image(image_data, self.default_prompt) + await self.send_response(room_id, f"📸 *Результат анализа:*\n\n{result}") + + async def handle_text_message(self, room_id: str, event: RoomMessageText) -> None: + if room_id not in self.config.allowed_rooms: + logger.warning(f"Ignoring text from non-allowed room {room_id}") + return + + body = event.body.strip().lower() + if body == "/help": + await self.send_response(room_id, "🤖 *Commands:*\n/help - справка\n/status - статус бота") + elif body == "/status": + await self.send_response( + room_id, + f"✅ *Статус:*\nПользователь: {self.config.username}\nМодель: {self.config.qwen_model}\nЭндпоинт: {self.config.qwen_endpoint}" + ) + + async def register_callbacks(self) -> None: + self.client.add_event_callback(self.image_callback, RoomMessageImage) + self.client.add_event_callback(self.text_callback, RoomMessageText) + + async def image_callback(self, room: RoomMessageImage, event: RoomMessageImage) -> None: + await self.handle_image_message(room.room_id, event) + + async def text_callback(self, room: RoomMessageText, event: RoomMessageText) -> None: + await self.handle_text_message(room.room_id, event) + + async def sync_loop(self) -> None: + logger.info("Starting sync loop...") + try: + await self.client.sync_forever(timeout=30000, full_state=True) + except Exception as e: + logger.error(f"Sync error: {e}") + raise + + async def close(self) -> None: + await self.client.close() + await self.qwen_client.close() + + +async def main(): + config = load_config() + if not config.password or not config.qwen_api_key or not config.allowed_rooms: + logger.error("Missing required config") + return + + bot = ImageBot(config) + try: + if not await bot.login(): + return + await bot.join_rooms() + await bot.register_callbacks() + await bot.sync_loop() + except KeyboardInterrupt: + logger.info("Bot stopped") + except Exception as e: + logger.error(f"Bot error: {e}") + finally: + await bot.close() + + +if __name__ == "__main__": + asyncio.run(main()) \ No newline at end of file From 75704f6930b24e1980289e61ea76a46b3a87c814 Mon Sep 17 00:00:00 2001 From: dropboy27 Date: Thu, 2 Apr 2026 13:37:20 +0300 Subject: [PATCH 2/5] =?UTF-8?q?=D1=80=D0=B0=D0=B1=D0=BE=D1=82=D0=B0=D0=B5?= =?UTF-8?q?=D1=82=20=D0=B2=20=D1=82=D0=B5=D0=BB=D0=B5=D0=B3=D0=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 68 +++---- requirements.txt | 6 +- src/image_bot.py | 489 ++++++++++++++++++++++++++--------------------- 3 files changed, 301 insertions(+), 262 deletions(-) diff --git a/README.md b/README.md index 658a55d..f14cd81 100644 --- a/README.md +++ b/README.md @@ -1,69 +1,47 @@ -# Matrix Image Recognition Bot +# Telegram Image Description Bot -Бот для Matrix, который распознаёт изображения с помощью Qwen-VL. - -## Описание - -Этот бот подключается к Matrix серверу, слушает изображения в разрешённых комнатах и отправляет их в Qwen-VL API для анализа. Бот возвращает описание изображения обратно в чат. +Бот для Telegram, который описывает изображения с помощью Qwen-VL API. ## Требования - Python 3.10+ -- Доступ к Matrix серверу -- API ключ Qwen-VL +- Токен Telegram бота (получить у [@BotFather](https://t.me/BotFather)) +- Ключ доступа к Qwen API ## Установка -1. Клонируйте репозиторий: -```bash -git clone -cd b2b_assistants -``` - -2. Создайте виртуальное окружение: -```bash -python -m venv venv -source venv/bin/activate # Linux/Mac -# или -venv\Scripts\activate # Windows -``` - -3. Установите зависимости: +1. Установите зависимости: ```bash pip install -r requirements.txt ``` -4. Настройте переменные окружения в файле `.env`: - +2. Настройте переменные окружения в файле `.env`: ```env -# Matrix подключение -HOMESERVER=https://matrix.lambda.coredump.ru -MATRIX_USERNAME=@image_bot:matrix.lambda.coredump.ru -PASSWORD=ваш_пароль_бота -ALLOWED_ROOMS=!QcPkdLDWqDegdtDnpP:matrix.lambda.coredump.ru +# Telegram Bot Token (получите у @BotFather) +TELEGRAM_BOT_TOKEN=ваш_токен_бота -# Qwen API -QWEN_API_KEY=sk-L6oRP0m15Z9YquluktS6w +# Qwen API настройки (уже заполнены) +QWEN_API_KEY=sk-L6oRP0mP15Z9YquluktS6w QWEN_ENDPOINT=https://llm.lambda.coredump.ru/v1 QWEN_VL_MODEL=qwen-vl-plus ``` -## Запуск - +3. Запустите бота: ```bash python src/image_bot.py ``` -## Команды бота - -- `/help` - показать справку по командам -- `/status` - показать статус бота - ## Использование -1. Пригласите бота в комнату Matrix -2. Отправьте изображение в чат -3. Бот автоматически проанализирует изображение и вернёт описание +1. Найдите вашего бота в Telegram и нажмите `/start` +2. Отправьте боту изображение +3. Бот вернёт описание изображения на русском языке + +## Команды + +- `/start` - начать работу с ботом +- `/help` - показать справку +- `/settoken ` - установить токен API (временное решение) ## Структура проекта @@ -76,6 +54,8 @@ b2b_assistants/ └── image_bot.py # Основной код бота ``` -## Лицензия +## Примечания -MIT \ No newline at end of file +- Бот использует base64 кодирование для отправки изображений в Qwen-VL API +- Для ограничения доступа используйте переменную `ALLOWED_USERS` в `.env` +- Время обработки изображения может составлять до 2 минут для больших файлов \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index d044578..1bf36d9 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,3 @@ -matrix-nio>=0.25.0 -httpx>=0.24.0 -python-dotenv>=1.0.0 \ No newline at end of file +python-telegram-bot==20.7 +python-dotenv==1.0.0 +requests==2.31.0 \ No newline at end of file diff --git a/src/image_bot.py b/src/image_bot.py index 2f542f3..dce137f 100644 --- a/src/image_bot.py +++ b/src/image_bot.py @@ -1,241 +1,300 @@ #!/usr/bin/env python3 -import asyncio +""" +Telegram Bot that describes images using Qwen-VL API. +""" + import os -import base64 import logging -from typing import Optional -from dataclasses import dataclass +import tempfile +from pathlib import Path -import httpx -from nio import ( - AsyncClient, - RoomMessageImage, - RoomMessageText, - LoginResponse, - JoinResponse, -) +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, - format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) 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") -@dataclass -class BotConfig: - homeserver: str - username: str - password: str - allowed_rooms: list[str] - qwen_api_key: str - qwen_endpoint: str - qwen_model: str +# 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()] -def load_config() -> BotConfig: - load_dotenv() - allowed_rooms_str = os.getenv("ALLOWED_ROOMS", "") - allowed_rooms = [r.strip() for r in allowed_rooms_str.split(",") if r.strip()] - return BotConfig( - homeserver=os.getenv("HOMESERVER", "https://matrix.lambda.coredump.ru"), - username=os.getenv("MATRIX_USERNAME", ""), - password=os.getenv("PASSWORD", ""), - allowed_rooms=allowed_rooms, - qwen_api_key=os.getenv("QWEN_API_KEY", ""), - qwen_endpoint=os.getenv("QWEN_ENDPOINT", "https://llm.lambda.coredump.ru/v1"), - qwen_model=os.getenv("QWEN_VL_MODEL", "qwen-vl-plus"), +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" + "/start - показать это сообщение\n" + "/help - показать справку\n" + "/draw <описание> - сгенерировать изображение" ) -class QwenVLClient: - def __init__(self, api_key: str, endpoint: str, model: str): - self.api_key = api_key - self.endpoint = endpoint - self.model = model - self.client = httpx.AsyncClient(timeout=120.0) - - async def analyze_image(self, image_data: bytes, prompt: str = "Опиши это изображение подробно:") -> str: - image_base64 = base64.b64encode(image_data).decode("utf-8") - image_url = f"data:image/jpeg;base64,{image_base64}" - - payload = { - "model": self.model, - "messages": [ - { - "role": "user", - "content": [ - {"type": "image_url", "image_url": {"url": image_url}}, - {"type": "text", "text": prompt} - ] - } - ] - } - - headers = {"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"} - - try: - response = await self.client.post( - f"{self.endpoint}/chat/completions", - json=payload, - headers=headers - ) - response.raise_for_status() - result = response.json() - return result["choices"][0]["message"]["content"] - except httpx.HTTPError as e: - logger.error(f"Error calling Qwen-VL API: {e}") - if e.response: - logger.error(f"Response: {e.response.text}") - return f"Ошибка при анализе изображения: {str(e)}" - - async def close(self): - await self.client.aclose() +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 для анализа изображений и генерирует картинки по запросу.\n\n" + "📋 **Команды:**\n" + "/start - начать работу\n" + "/help - показать эту справку\n" + "/draw <описание> - сгенерировать изображение по описанию\n\n" + "💡 **Как использовать:**\n" + "• Отправьте боту изображение для получения описания\n" + "• Используйте /draw для генерации изображения" + ) + await update.message.reply_text(help_text, parse_mode="Markdown") -class ImageBot: - def __init__(self, config: BotConfig): - self.config = config - self.client = AsyncClient(config.homeserver, config.username) # исправлено - self.qwen_client = QwenVLClient( - config.qwen_api_key, - config.qwen_endpoint, - config.qwen_model - ) - self.default_prompt = "Опиши это изображение подробно." - - async def login(self) -> bool: - try: - response = await self.client.login( - password=self.config.password, - device_name="image_recognition_bot" - ) - if isinstance(response, LoginResponse): - logger.info(f"Logged in as {self.config.username}") - return True - else: - logger.error(f"Login failed: {response}") - return False - except Exception as e: - logger.error(f"Login error: {e}") - return False - - async def join_rooms(self) -> None: - for room_id in self.config.allowed_rooms: - try: - response = await self.client.join(room_id) - if isinstance(response, JoinResponse): - logger.info(f"Joined room: {room_id}") - else: - logger.warning(f"Could not join room {room_id}: {response}") - except Exception as e: - logger.error(f"Error joining room {room_id}: {e}") - - async def download_image(self, url: str) -> Optional[bytes]: - try: - logger.info(f"Downloading image from {url}") - response = await self.client.download(url) # правильный метод - if response and hasattr(response, 'body'): - logger.info(f"Downloaded {len(response.body)} bytes") - return response.body - return None - except Exception as e: - logger.error(f"Download error: {e}") - return None - - async def send_response(self, room_id: str, message: str) -> None: - try: - await self.client.room_send( - room_id=room_id, - message_type="m.room.message", - content={"msgtype": "m.text", "body": message} - ) - except Exception as e: - logger.error(f"Error sending response: {e}") - - async def handle_image_message(self, room_id: str, event: RoomMessageImage) -> None: - if room_id not in self.config.allowed_rooms: - logger.warning(f"Ignoring image from non-allowed room {room_id}") - return - - logger.info(f"Processing image in {room_id}") - image_url = event.url # прямой доступ к URL - if not image_url: - await self.send_response(room_id, "Не удалось получить изображение.") - return - - image_data = await self.download_image(image_url) - if not image_data: - await self.send_response(room_id, "Не удалось загрузить изображение.") - return - - try: - await self.client.typing(room_id, timeout=4000) - except Exception: - pass - - await self.send_response(room_id, "🔍 Анализирую изображение...") - result = await self.qwen_client.analyze_image(image_data, self.default_prompt) - await self.send_response(room_id, f"📸 *Результат анализа:*\n\n{result}") - - async def handle_text_message(self, room_id: str, event: RoomMessageText) -> None: - if room_id not in self.config.allowed_rooms: - logger.warning(f"Ignoring text from non-allowed room {room_id}") - return - - body = event.body.strip().lower() - if body == "/help": - await self.send_response(room_id, "🤖 *Commands:*\n/help - справка\n/status - статус бота") - elif body == "/status": - await self.send_response( - room_id, - f"✅ *Статус:*\nПользователь: {self.config.username}\nМодель: {self.config.qwen_model}\nЭндпоинт: {self.config.qwen_endpoint}" - ) - - async def register_callbacks(self) -> None: - self.client.add_event_callback(self.image_callback, RoomMessageImage) - self.client.add_event_callback(self.text_callback, RoomMessageText) - - async def image_callback(self, room: RoomMessageImage, event: RoomMessageImage) -> None: - await self.handle_image_message(room.room_id, event) - - async def text_callback(self, room: RoomMessageText, event: RoomMessageText) -> None: - await self.handle_text_message(room.room_id, event) - - async def sync_loop(self) -> None: - logger.info("Starting sync loop...") - try: - await self.client.sync_forever(timeout=30000, full_state=True) - except Exception as e: - logger.error(f"Sync error: {e}") - raise - - async def close(self) -> None: - await self.client.close() - await self.qwen_client.close() -async def main(): - config = load_config() - if not config.password or not config.qwen_api_key or not config.allowed_rooms: - logger.error("Missing required config") - return - - bot = ImageBot(config) +async def describe_image(image_url: str) -> str: + """Send image to Qwen-VL API and get description.""" + headers = { + "Authorization": f"Bearer {QWEN_API_KEY}", + "Content-Type": "application/json", + } + + payload = { + "model": QWEN_MODEL, + "messages": [ + { + "role": "user", + "content": [ + {"type": "image_url", "image_url": {"url": image_url}}, + {"type": "text", "text": "Кратко опиши изображение на русском языке. Перечисли основные детали, каждая в 1-2 предложениях. Будь лаконичен."}, + ], + } + ], + } + + api_url = f"{QWEN_ENDPOINT}/chat/completions" + try: - if not await bot.login(): - return - await bot.join_rooms() - await bot.register_callbacks() - await bot.sync_loop() - except KeyboardInterrupt: - logger.info("Bot stopped") + 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) + + # For Qwen API, we need a public URL + # Since we can't easily host the image, we'll use a different approach + # We'll read the image and send it as base64 if the API supports it + + try: + description = await describe_image_with_base64(tmp_path) + except Exception as e: + logger.warning(f"Base64 approach failed: {e}, trying URL approach") + # Fallback: try with a placeholder 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"Bot error: {e}") - finally: - await bot.close() + 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.""" + if not context.args: + await update.message.reply_text( + "Пожалуйста, укажите описание изображения:\n/draw <описание того что хотите увидеть>" + ) + return + + prompt = " ".join(context.args) + await update.message.reply_text("🎨 Генерирую изображение... Это может занять до 2 минут.") + + try: + image_url = await generate_image(prompt) + if image_url: + await context.bot.send_photo(chat_id=update.effective_chat.id, photo=image_url, caption=f"✨ {prompt}") + else: + await update.message.reply_text("❌ Не удалось сгенерировать изображение. API не поддерживает генерацию картинок.") + except Exception as e: + logger.error(f"Image generation error: {e}") + await update.message.reply_text(f"❌ Ошибка при генерации: {str(e)}") + + +async def generate_image(prompt: str) -> str | None: + """Generate image using Qwen API or alternative services.""" + headers = { + "Authorization": f"Bearer {QWEN_API_KEY}", + "Content-Type": "application/json", + } + + # Try Qwen image generation endpoint first + # API expects prompt as a direct parameter + payload = { + "model": "mai", + "prompt": prompt, + "n": 1, + "size": "1024x1024" + } + + api_url = f"{QWEN_ENDPOINT}/images/generations" + + try: + response = requests.post(api_url, headers=headers, json=payload, timeout=120) + logger.info(f"Image API response: {response.status_code}") + + if response.status_code == 200: + result = response.json() + logger.info(f"Image result: {result}") + # Try different response formats + if "data" in result and len(result["data"]) > 0: + return result["data"][0].get("url") or result["data"][0].get("image_url") + elif "image_url" in result: + return result["image_url"] + elif "output" in result and "url" in result["output"]: + return result["output"]["url"] + else: + logger.warning(f"Image API failed: {response.status_code} - {response.text}") + except Exception as e: + logger.warning(f"Image generation failed: {e}") + + # Fallback: return None and inform user + return None + + +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("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__": - asyncio.run(main()) \ No newline at end of file + main() \ No newline at end of file From 4d444b4621505a0623901e883d305f3c65b34091 Mon Sep 17 00:00:00 2001 From: dropboy27 Date: Mon, 6 Apr 2026 13:53:55 +0300 Subject: [PATCH 3/5] =?UTF-8?q?=D0=BA=D0=B0=D1=80=D1=82=D0=B8=D0=BD=D0=BA?= =?UTF-8?q?=D0=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- requirements.txt | 32 +- src/image_bot.py | 1091 +++++++++++++++++++++++++++++++++++++++++++--- 2 files changed, 1061 insertions(+), 62 deletions(-) diff --git a/requirements.txt b/requirements.txt index 1bf36d9..b51df25 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,33 @@ +# 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 \ No newline at end of file +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 index dce137f..df59888 100644 --- a/src/image_bot.py +++ b/src/image_bot.py @@ -1,12 +1,19 @@ #!/usr/bin/env python3 """ -Telegram Bot that describes images using Qwen-VL API. +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 @@ -40,18 +47,29 @@ 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\n" "📸 Отправь мне картинку - я опишу что на ней изображено\n" - "🎨 Используй /draw <описание> - я сгенерирую изображение\n\n" + "🎨 Используй /draw <описание> - я сгенерирую изображение или схему\n\n" + "📊 **Генерация схем и графиков для отчётов:**\n" + "/draw схема архитектуры системы\n" + "/draw график роста продаж за квартал\n" + "/draw блок-схема процесса одобрения кредита\n" + "/draw диаграмма распределения бюджета\n\n" "Доступные команды:\n" "/start - показать это сообщение\n" "/help - показать справку\n" + "/model - показать текущую модель генерации\n" "/draw <описание> - сгенерировать изображение" ) @@ -60,26 +78,67 @@ async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> No """Send a message when the command /help is issued.""" help_text = ( "📸 **Бот описания и генерации изображений**\n\n" - "Этот бот использует Qwen-VL для анализа изображений и генерирует картинки по запросу.\n\n" + "Этот бот использует Qwen-VL для анализа изображений и Qwen + Python библиотеки для генерации чётких схем.\n\n" "📋 **Команды:**\n" "/start - начать работу\n" "/help - показать эту справку\n" - "/draw <описание> - сгенерировать изображение по описанию\n\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 для генерации изображения" + "• Используйте `/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_url: str) -> str: - """Send image to Qwen-VL API and get description.""" - headers = { - "Authorization": f"Bearer {QWEN_API_KEY}", - "Content-Type": "application/json", - } +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, @@ -87,11 +146,20 @@ async def describe_image(image_url: str) -> str: { "role": "user", "content": [ - {"type": "image_url", "image_url": {"url": image_url}}, - {"type": "text", "text": "Кратко опиши изображение на русском языке. Перечисли основные детали, каждая в 1-2 предложениях. Будь лаконичен."}, + {"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" @@ -135,15 +203,10 @@ async def handle_photo(update: Update, context: ContextTypes.DEFAULT_TYPE) -> No tmp_path = tmp_file.name await file.download_to_drive(tmp_path) - # For Qwen API, we need a public URL - # Since we can't easily host the image, we'll use a different approach - # We'll read the image and send it as base64 if the API supports it - try: description = await describe_image_with_base64(tmp_path) except Exception as e: logger.warning(f"Base64 approach failed: {e}, trying URL approach") - # Fallback: try with a placeholder URL approach description = await describe_image("placeholder") # Clean up temp file @@ -198,7 +261,7 @@ async def describe_image_with_base64(image_path: str) -> str: async def draw_image(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: - """Handle /draw command to generate images.""" + """Handle /draw command to generate images using Qwen.""" if not context.args: await update.message.reply_text( "Пожалуйста, укажите описание изображения:\n/draw <описание того что хотите увидеть>" @@ -206,60 +269,965 @@ async def draw_image(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None return prompt = " ".join(context.args) - await update.message.reply_text("🎨 Генерирую изображение... Это может занять до 2 минут.") - try: - image_url = await generate_image(prompt) - if image_url: - await context.bot.send_photo(chat_id=update.effective_chat.id, photo=image_url, caption=f"✨ {prompt}") - else: - await update.message.reply_text("❌ Не удалось сгенерировать изображение. API не поддерживает генерацию картинок.") - except Exception as e: - logger.error(f"Image generation error: {e}") - await update.message.reply_text(f"❌ Ошибка при генерации: {str(e)}") + # 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)}") -async def generate_image(prompt: str) -> str | None: - """Generate image using Qwen API or alternative services.""" - headers = { - "Authorization": f"Bearer {QWEN_API_KEY}", - "Content-Type": "application/json", - } +def detect_diagram_type(prompt: str) -> str: + """Detect the type of diagram requested.""" + prompt_lower = prompt.lower() - # Try Qwen image generation endpoint first - # API expects prompt as a direct parameter - payload = { - "model": "mai", - "prompt": prompt, - "n": 1, - "size": "1024x1024" - } + # Flowchart / Block diagram + if any(kw in prompt_lower for kw in ["блок-схема", "flowchart", "алгоритм", "процесс", "этапы"]): + return "flowchart" - api_url = f"{QWEN_ENDPOINT}/images/generations" + # 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: - response = requests.post(api_url, headers=headers, json=payload, timeout=120) - logger.info(f"Image API response: {response.status_code}") - - if response.status_code == 200: - result = response.json() - logger.info(f"Image result: {result}") - # Try different response formats - if "data" in result and len(result["data"]) > 0: - return result["data"][0].get("url") or result["data"][0].get("image_url") - elif "image_url" in result: - return result["image_url"] - elif "output" in result and "url" in result["output"]: - return result["output"]["url"] + # 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: - logger.warning(f"Image API failed: {response.status_code} - {response.text}") + # Default to generic matplotlib + return await generate_generic_diagram(prompt) + except Exception as e: - logger.warning(f"Image generation failed: {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 - # Fallback: return None and inform user + 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. Шрифты: plt.rcParams['font.family'] = 'DejaVu Sans' для кириллицы +4. Цвета: профессиональная палитра для разных слоёв: + - Frontend: lightblue (#E3F2FD) + - Backend/API: lightgreen (#E8F5E9) + - Database: lightyellow (#FFFDE7) + - External services: lightcoral (#FFEBEE) + - Cache: lightpurple (#F3E5F5) + +ЭЛЕМЕНТЫ АРХИТЕКТУРНОЙ ДИАГРАММЫ: +- FancyBboxPatch с boxstyle="round,pad=0.05" для компонентов +- Стрелки FancyArrowPatch для связей между компонентами +- Текст с ha='center', va='center' для подписей +- Группировка компонентов в логические слои + +ПРИМЕР КОДА: +import matplotlib.pyplot as plt +from matplotlib import patches +plt.rcParams['font.family'] = 'DejaVu Sans' +fig, ax = plt.subplots(figsize=(14, 10)) +# Frontend layer +frontend = patches.FancyBboxPatch((0.1, 0.7), 0.3, 0.15, boxstyle="round,pad=0.05", + facecolor="#E3F2FD", edgecolor="#1976D2", linewidth=2) +ax.add_patch(frontend) +ax.text(0.25, 0.775, "Frontend", ha='center', va='center', fontsize=14, fontweight='bold') +# Arrow +arrow = patches.FancyArrowPatch((0.4, 0.75), (0.6, 0.55), + arrowstyle='->', mutation_scale=20, linewidth=2) +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"} + + payload = { + "model": QWEN_MODEL, + "messages": [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": user_prompt} + ], + "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: @@ -285,6 +1253,7 @@ def main() -> None: # 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), From b9bf405a706a064dd1bda230249087cbc84865cf Mon Sep 17 00:00:00 2001 From: dropboy27 Date: Mon, 6 Apr 2026 14:25:48 +0300 Subject: [PATCH 4/5] 2 --- src/image_bot.py | 43 +++++++++++++++++++++++-------------------- 1 file changed, 23 insertions(+), 20 deletions(-) diff --git a/src/image_bot.py b/src/image_bot.py index df59888..7d756dc 100644 --- a/src/image_bot.py +++ b/src/image_bot.py @@ -547,7 +547,7 @@ async def generate_architecture_diagram(prompt: str) -> bytes | None: ТРЕБОВАНИЯ К КАЧЕСТВУ: 1. Размер фигуры: figsize=(14, 10) или больше 2. DPI: минимум 150 для высокого качества -3. Шрифты: plt.rcParams['font.family'] = 'DejaVu Sans' для кириллицы +3. Шрифты: используй DejaVu Sans для кириллицы 4. Цвета: профессиональная палитра для разных слоёв: - Frontend: lightblue (#E3F2FD) - Backend/API: lightgreen (#E8F5E9) @@ -556,45 +556,45 @@ async def generate_architecture_diagram(prompt: str) -> bytes | None: - Cache: lightpurple (#F3E5F5) ЭЛЕМЕНТЫ АРХИТЕКТУРНОЙ ДИАГРАММЫ: -- FancyBboxPatch с boxstyle="round,pad=0.05" для компонентов -- Стрелки FancyArrowPatch для связей между компонентами +- Используй 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' +plt.rcParams["font.family"] = "DejaVu Sans" fig, ax = plt.subplots(figsize=(14, 10)) -# Frontend layer -frontend = patches.FancyBboxPatch((0.1, 0.7), 0.3, 0.15, boxstyle="round,pad=0.05", - facecolor="#E3F2FD", edgecolor="#1976D2", linewidth=2) -ax.add_patch(frontend) -ax.text(0.25, 0.775, "Frontend", ha='center', va='center', fontsize=14, fontweight='bold') -# Arrow -arrow = patches.FancyArrowPatch((0.4, 0.75), (0.6, 0.55), - arrowstyle='->', mutation_scale=20, linewidth=2) +# Создаём компонент +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') +ax.axis("off") +plt.title("Архитектура системы", fontsize=18, fontweight="bold", pad=20) +plt.savefig("architecture.png", dpi=150, bbox_inches="tight") plt.close() -Генерируй ТОЛЬКО Python код без объяснений.""" +Генерируй ТОЛЬКО Python код без объяснений. Используй двойные кавычки.""" user_prompt = f"""Создай профессиональную архитектурную диаграмму для следующего запроса: {prompt} Требования: -- Используй FancyBboxPatch для компонентов +- Используй FancyBboxPatch для компонентов (все параметры в конструкторе) - Добавь стрелки FancyArrowPatch для связей - Цвета для разных слоёв архитектуры - Чёткие подписи на русском языке - Заголовок диаграммы - Сохраняй в 'architecture.png' -Генерируй ТОЛЬКО Python код без объяснений.""" +Генерируй ТОЛЬКО Python код без объяснений. Используй двойные кавычки.""" code = await get_code_from_qwen(system_prompt, user_prompt, temperature=0.4) if code: @@ -961,11 +961,14 @@ async def get_code_from_qwen(system_prompt: str, user_prompt: str, temperature: """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": user_prompt} + {"role": "user", "content": f"{user_prompt}\n\n[Unique request ID: {unique_id}. Это не часть кода, просто идентификатор запроса.]"} ], "temperature": temperature, "max_tokens": 4096, From 92e2be426e1c53b43397b4a8771a60a8050f43d7 Mon Sep 17 00:00:00 2001 From: pyanzin Date: Wed, 8 Apr 2026 15:09:37 +0300 Subject: [PATCH 5/5] init --- README.md | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/README.md b/README.md index f14cd81..b50dbda 100644 --- a/README.md +++ b/README.md @@ -21,9 +21,7 @@ pip install -r requirements.txt TELEGRAM_BOT_TOKEN=ваш_токен_бота # Qwen API настройки (уже заполнены) -QWEN_API_KEY=sk-L6oRP0mP15Z9YquluktS6w -QWEN_ENDPOINT=https://llm.lambda.coredump.ru/v1 -QWEN_VL_MODEL=qwen-vl-plus + ``` 3. Запустите бота: