"""
Telegram бот для работы с отгрузками МойСклад
Функции:
- Поиск отгрузок (по номеру заказа или отгрузки)
- Пагинация списка отгрузок
- Валидация данных отгрузки
- Генерация документов (XML/Excel)
- Выбор печатных форм из МойСклад
"""
import os
import sys
import logging
from pathlib import Path
from telegram import (
Update,
ReplyKeyboardMarkup,
InlineKeyboardButton,
InlineKeyboardMarkup
)
from telegram.ext import (
Application,
CommandHandler,
MessageHandler,
CallbackQueryHandler,
ContextTypes,
ConversationHandler,
filters
)
# Добавляем путь к modules
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'modules'))
from moysklad_client import MoySkladClient # noqa: E402
from validator import OrderValidator # noqa: E402
from xml_generator import XMLGenerator # noqa: E402
from user_manager import UserManager # noqa: E402
from config import settings # noqa: E402
# Настройка логирования
logging.basicConfig(
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=logging.INFO
)
logger = logging.getLogger(__name__)
# Состояния разговора
(
AUTH_LOGIN,
AUTH_PASSWORD,
MAIN_MENU,
SEARCH_ORDER,
SHOW_ORDERS,
SELECT_ORDER,
VALIDATE_ORDER,
SELECT_FORMAT,
SELECT_TEMPLATE,
DOWNLOAD_DOCUMENT
) = range(10)
# Менеджер пользователей
user_manager = UserManager()
def get_moysklad_client(user_id: int) -> MoySkladClient:
"""
Получает клиента МойСклад для конкретного пользователя
Args:
user_id: ID пользователя Telegram
Returns:
MoySkladClient с учетными данными пользователя
"""
credentials = user_manager.get_user_credentials(user_id)
if not credentials:
raise ValueError(f"Пользователь {user_id} не авторизован")
return MoySkladClient(
credentials['login'],
credentials['password'],
settings.moysklad_api_url
)
def get_main_keyboard():
"""Возвращает основную клавиатуру"""
keyboard = [
["🔍 Поиск отгрузки", "📋 Список отгрузок"],
["❓ Помощь"]
]
return ReplyKeyboardMarkup(keyboard, resize_keyboard=True)
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Обработчик команды /start"""
user = update.effective_user
user_id = user.id
# Проверяем, авторизован ли пользователь
if not user_manager.is_user_registered(user_id):
await update.message.reply_text(
f"Привет, {user.first_name}! 👋\n\n"
f"Для начала работы необходимо авторизоваться в МойСклад.\n\n"
f"Пожалуйста, введите ваш логин от МойСклад:",
parse_mode='HTML'
)
return AUTH_LOGIN
# Пользователь авторизован - показываем главное меню
await update.message.reply_text(
f"С возвращением, {user.first_name}! 👋\n\n"
f"Выберите действие:",
reply_markup=get_main_keyboard()
)
# Очищаем данные пользователя
context.user_data.clear()
return MAIN_MENU
async def auth_login(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Обработчик ввода логина"""
login = update.message.text.strip()
# Сохраняем логин временно
context.user_data['temp_login'] = login
await update.message.reply_text(
"✅ Логин принят.\n\n"
"Теперь введите ваш пароль от МойСклад:\n\n"
"⚠️ Пароль будет зашифрован и сохранён безопасно.",
parse_mode='HTML'
)
return AUTH_PASSWORD
async def auth_password(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Обработчик ввода пароля и завершение авторизации"""
user = update.effective_user
user_id = user.id
password = update.message.text.strip()
login = context.user_data.get('temp_login')
# Удаляем сообщение с паролем для безопасности
try:
await update.message.delete()
except Exception:
pass
if not login:
await update.message.reply_text(
"❌ Ошибка: логин не найден. Начните заново с команды /start"
)
return ConversationHandler.END
# Проверяем учетные данные
await update.message.reply_text("⏳ Проверяю учетные данные...")
try:
# Создаём временный клиент для проверки
test_client = MoySkladClient(
login, password, settings.moysklad_api_url
)
# Пробуем получить данные для проверки авторизации
result = test_client.get_orders(limit=1)
if result and 'rows' in result:
# Авторизация успешна - сохраняем пользователя
success = user_manager.register_user(
user_id,
login,
password,
user.username
)
if success:
# Очищаем временные данные
context.user_data.pop('temp_login', None)
await update.message.reply_text(
"✅ Авторизация успешна!\n\n"
"Ваши учетные данные сохранены и зашифрованы.\n\n"
"Выберите действие:",
parse_mode='HTML',
reply_markup=get_main_keyboard()
)
logger.info(f"Пользователь {user_id} успешно авторизован")
return MAIN_MENU
else:
await update.message.reply_text(
"❌ Ошибка сохранения данных. "
"Попробуйте позже или обратитесь к администратору."
)
return ConversationHandler.END
else:
await update.message.reply_text(
"❌ Ошибка авторизации\n\n"
"Неверный логин или пароль. Попробуйте снова командой /start",
parse_mode='HTML'
)
context.user_data.pop('temp_login', None)
return ConversationHandler.END
except Exception as e:
logger.error(f"Ошибка авторизации пользователя {user_id}: {e}")
await update.message.reply_text(
"❌ Ошибка авторизации\n\n"
f"Не удалось подключиться к МойСклад.\n"
f"Проверьте правильность логина и пароля.\n\n"
f"Детали: {str(e)[:100]}",
parse_mode='HTML'
)
context.user_data.pop('temp_login', None)
return ConversationHandler.END
async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Показывает помощь"""
help_text = (
"📖 Инструкция по использованию\n\n"
"🔍 Поиск отгрузки\n"
"Введите номер заказа или отгрузки для поиска.\n"
"Например: 951 или 00001\n\n"
"📋 Список отгрузок\n"
"Показывает последние отгрузки с пагинацией.\n"
"Используйте кнопки для навигации.\n\n"
"🔐 Учетные данные\n"
"/logout - Выйти из аккаунта МойСклад\n"
"/change_credentials - Сменить логин/пароль\n\n"
"✅ Валидация\n"
"После выбора отгрузки бот автоматически проверит:\n"
"- Заполненность обязательных полей\n"
"- Корректность ИНН/КПП\n"
"- Соответствие сумм\n\n"
"📄 Генерация документов\n"
"После успешной валидации можно:\n"
"- Скачать XML для ЭДО\n"
"- Сгенерировать Excel (УПД, ТН и др.)"
)
await update.message.reply_text(
help_text,
parse_mode='HTML',
reply_markup=get_main_keyboard()
)
async def logout_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Выход из аккаунта МойСклад"""
user_id = update.effective_user.id
if user_manager.delete_user(user_id):
await update.message.reply_text(
"✅ Вы вышли из аккаунта.\n\n"
"Ваши учетные данные удалены.\n\n"
"Используйте /start для новой авторизации."
)
logger.info(f"Пользователь {user_id} вышел из аккаунта")
else:
await update.message.reply_text(
"❌ Вы не были авторизованы."
)
return ConversationHandler.END
async def change_credentials_command(
update: Update,
context: ContextTypes.DEFAULT_TYPE
):
"""Смена учетных данных"""
user_id = update.effective_user.id
if not user_manager.is_user_registered(user_id):
await update.message.reply_text(
"❌ Вы не авторизованы. Используйте /start"
)
return ConversationHandler.END
# Удаляем старые данные
user_manager.delete_user(user_id)
await update.message.reply_text(
"🔄 Смена учетных данных\n\n"
"Введите новый логин от МойСклад:",
parse_mode='HTML'
)
return AUTH_LOGIN
return MAIN_MENU
async def search_order_prompt(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Запрашивает номер заказа для поиска"""
await update.message.reply_text(
"🔍 Введите номер заказа или отгрузки:\n\n"
"Например: 951, 00001 или полный UUID",
reply_markup=ReplyKeyboardMarkup(
[["⬅️ Назад"]],
resize_keyboard=True
)
)
return SEARCH_ORDER
async def search_order(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Ищет заказ по номеру"""
user_id = update.effective_user.id
query = update.message.text.strip()
if query == "⬅️ Назад":
return await start(update, context)
# Показываем индикатор загрузки
msg = await update.message.reply_text("🔄 Поиск...")
try:
# Получаем клиента для пользователя
moysklad_client = get_moysklad_client(user_id)
# Ищем через API
search_result = moysklad_client.search_orders(query, limit=10)
orders = search_result.get("rows", [])
if not orders:
await msg.edit_text(
"❌ Отгрузки не найдены.\n\n"
"Попробуйте другой запрос:"
)
await update.message.reply_text(
"Введите номер заказа или отгрузки:",
reply_markup=ReplyKeyboardMarkup(
[["⬅️ Назад"]],
resize_keyboard=True
)
)
return SEARCH_ORDER
# Сохраняем результаты
context.user_data['orders'] = orders
context.user_data['page'] = 0
# Показываем результаты
await msg.delete()
await show_orders_page(update, context)
return SELECT_ORDER
except Exception as e:
logger.error(f"Ошибка поиска: {e}")
await msg.edit_text(
f"❌ Ошибка поиска: {str(e)}\n\n"
f"Попробуйте снова:"
)
await update.message.reply_text(
"Введите номер заказа или отгрузки:",
reply_markup=ReplyKeyboardMarkup(
[["⬅️ Назад"]],
resize_keyboard=True
)
)
return SEARCH_ORDER
async def show_orders_list(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Показывает список последних отгрузок"""
user_id = update.effective_user.id
msg = await update.message.reply_text("🔄 Загрузка списка...")
try:
# Получаем клиента для пользователя
moysklad_client = get_moysklad_client(user_id)
# Получаем последние отгрузки
result = moysklad_client.get_orders(limit=25, offset=0)
orders = result.get("rows", [])
if not orders:
await msg.edit_text(
"❌ Отгрузки не найдены.",
reply_markup=get_main_keyboard()
)
return MAIN_MENU
# Сохраняем результаты
context.user_data['orders'] = orders
context.user_data['page'] = 0
context.user_data['total'] = result.get("meta", {}).get("size", len(orders))
# Показываем первую страницу
await msg.delete()
await show_orders_page(update, context)
return SELECT_ORDER
except Exception as e:
logger.error(f"Ошибка получения списка: {e}")
await msg.edit_text(
f"❌ Ошибка: {str(e)}",
reply_markup=get_main_keyboard()
)
return MAIN_MENU
async def show_orders_page(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Показывает страницу с отгрузками"""
orders = context.user_data.get('orders', [])
page = context.user_data.get('page', 0)
page_size = 5
start = page * page_size
end = start + page_size
page_orders = orders[start:end]
if not page_orders:
await update.message.reply_text(
"❌ Нет отгрузок на этой странице.",
reply_markup=get_main_keyboard()
)
return MAIN_MENU
# Формируем текст со списком
text = f"📋 Отгрузки (стр. {page + 1})\n\n"
# Создаём inline кнопки для каждой отгрузки
keyboard = []
for i, order in enumerate(page_orders, start=start+1):
order_name = order.get('name', 'N/A')
order_date = order.get('moment', '')[:10] # YYYY-MM-DD
sum_val = order.get('sum', 0) / 100 # Копейки в рубли
agent = order.get('agent', {})
agent_name = agent.get('name', 'N/A')
text += f"{i}. №{order_name} от {order_date}\n"
text += f" Контрагент: {agent_name}\n"
text += f" Сумма: {sum_val:,.2f} ₽\n\n"
# Кнопка выбора
order_id = order['meta']['href'].split('/')[-1].split('?')[0]
keyboard.append([
InlineKeyboardButton(
f"✅ Выбрать №{order_name}",
callback_data=f"select:{order_id}"
)
])
# Кнопки навигации
nav_buttons = []
if page > 0:
nav_buttons.append(
InlineKeyboardButton("⬅️ Назад", callback_data="page:prev")
)
if end < len(orders):
nav_buttons.append(
InlineKeyboardButton("➡️ Вперёд", callback_data="page:next")
)
if nav_buttons:
keyboard.append(nav_buttons)
# Кнопка отмены
keyboard.append([
InlineKeyboardButton("❌ Отмена", callback_data="cancel")
])
reply_markup = InlineKeyboardMarkup(keyboard)
# Отправляем или редактируем сообщение
if update.callback_query:
await update.callback_query.edit_message_text(
text,
parse_mode='HTML',
reply_markup=reply_markup
)
else:
await update.message.reply_text(
text,
parse_mode='HTML',
reply_markup=reply_markup
)
return SELECT_ORDER
async def handle_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Обработчик inline кнопок"""
query = update.callback_query
await query.answer()
data = query.data
if data == "cancel":
await query.edit_message_text(
"❌ Отменено.",
reply_markup=None
)
await query.message.reply_text(
"Выберите действие:",
reply_markup=get_main_keyboard()
)
# Очищаем данные пользователя
context.user_data.clear()
return MAIN_MENU
elif data.startswith("page:"):
# Пагинация
direction = data.split(":")[1]
page = context.user_data.get('page', 0)
if direction == "prev":
context.user_data['page'] = max(0, page - 1)
else: # next
context.user_data['page'] = page + 1
await show_orders_page(update, context)
return SELECT_ORDER
elif data.startswith("select:"):
# Выбрана отгрузка
order_id = data.split(":")[1]
context.user_data['selected_order_id'] = order_id
await query.edit_message_text(
"🔄 Загрузка данных отгрузки...",
reply_markup=None
)
# Переходим к валидации
return await validate_order(update, context)
elif data.startswith("format:"):
# Выбран формат документа
format_type = data.split(":")[1]
context.user_data['format'] = format_type
if format_type == "xml":
# XML - показываем выбор типа XML
return await show_xml_format_choice(update, context)
elif format_type == "package":
# Комплект документов - генерируем сразу
return await generate_package(update, context)
else:
# Excel - сбрасываем флаг и показываем шаблоны
context.user_data['show_all_templates'] = False
return await show_templates(update, context)
elif data.startswith("xml_format:"):
# Выбран тип XML
xml_format = data.split(":")[1]
context.user_data['xml_format'] = xml_format
return await generate_xml(update, context)
elif data == "generate_another":
# Генерировать другой файл - возвращаемся к выбору формата
order_id = context.user_data.get('selected_order_id')
logger.info(f"generate_another clicked, order_id: {order_id}")
if order_id:
# Убираем кнопки с предыдущего сообщения
await query.edit_message_reply_markup(reply_markup=None)
# Вызываем validate_order напрямую с query
# validate_order сам покажет данные валидации
return await validate_order(update, context)
else:
await query.answer("❌ Ошибка: отгрузка не найдена")
await query.message.reply_text(
"Выберите действие:",
reply_markup=get_main_keyboard()
)
context.user_data.clear()
return MAIN_MENU
elif data.startswith("template:"):
# Выбран шаблон
template_id = data.split(":")[1]
context.user_data['template_id'] = template_id
return await generate_excel(update, context)
elif data == "show_all_templates":
# Показать все шаблоны
context.user_data['show_all_templates'] = True
return await show_templates(update, context)
# Если callback не распознан - логируем и возвращаем текущее состояние
logger.warning(f"Неизвестный callback: {data}")
return SELECT_FORMAT
async def validate_order(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Валидация отгрузки"""
user_id = update.effective_user.id
order_id = context.user_data.get('selected_order_id')
try:
# Получаем клиента для пользователя
moysklad_client = get_moysklad_client(user_id)
# Получаем полные данные
order_data = moysklad_client.get_full_order_data(order_id)
context.user_data['order_data'] = order_data
# Проводим валидацию
validator = OrderValidator()
# Собираем данные для валидации
validation_data = {
'order': order_data['order'],
'organization': order_data['organization'],
'counterparty': order_data['counterparty'],
'positions': order_data['positions']
}
# Валидируем заказ
validation_results = validator.validate_order(validation_data)
order = order_data['order']
org = order_data['organization']
counterparty = order_data['counterparty']
positions = order_data['positions']
# Проверяем наличие критичных ошибок
has_critical = any(
r.level.name == "CRITICAL"
for r in validation_results
)
has_warnings = any(
r.level.name == "WARNING"
for r in validation_results
)
# Формируем текст с результатами валидации
if has_critical:
status_icon = "❌"
status_text = "Валидация не пройдена!"
elif has_warnings:
status_icon = "⚠️"
status_text = "Валидация выявила предупреждения"
else:
status_icon = "✅"
status_text = "Валидация пройдена успешно!"
validation_text = (
f"{status_icon} {status_text}\n\n"
f"Отгрузка №{order.get('name')}\n"
f"Дата: {order.get('moment', '')[:10]}\n"
f"Сумма: {order.get('sum', 0) / 100:,.2f} ₽\n\n"
f"Продавец: {org.get('name')}\n"
f"ИНН: {org.get('inn')}\n\n"
f"Покупатель: {counterparty.get('name')}\n"
f"ИНН: {counterparty.get('inn')}\n\n"
)
# Добавляем информацию о грузополучателе если есть
# Если грузополучатель не указан явно, используем контрагента
consignee = order_data.get('consignee')
if not consignee:
consignee = counterparty # Используем контрагента
if consignee:
consignee_name = consignee.get('name', 'Не указан')
consignee_inn = consignee.get('inn', 'Не указан')
validation_text += (
f"Грузополучатель: {consignee_name}\n"
f"ИНН: {consignee_inn}\n\n"
)
validation_text += f"Товаров: {len(positions)} шт.\n"
# Добавляем результаты валидации
if validation_results:
validation_text += "\nРезультаты проверки:\n"
for result in validation_results[:5]: # Показываем первые 5
level_icon = {
"VALID": "✅",
"WARNING": "⚠️",
"CRITICAL": "❌"
}.get(result.level.name, "ℹ️")
validation_text += f"{level_icon} {result.message}\n"
if len(validation_results) > 5:
remaining = len(validation_results) - 5
validation_text += f"\n... и ещё {remaining}\n"
validation_text += "\nВыберите формат документа:"
# Формируем клавиатуру - блокируем генерацию при критических ошибках
if has_critical:
keyboard = [
[InlineKeyboardButton(
"🔒 Генерация документов заблокирована",
callback_data="blocked"
)],
[InlineKeyboardButton("❌ Отмена", callback_data="cancel")]
]
else:
keyboard = [
[InlineKeyboardButton(
"🧾 XML",
callback_data="format:xml"
)],
[InlineKeyboardButton(
"📊 Excel",
callback_data="format:excel"
)],
[InlineKeyboardButton(
"📦 Комплект (XML 1С + Excel)",
callback_data="format:package"
)],
[InlineKeyboardButton("❌ Отмена", callback_data="cancel")]
]
if update.callback_query:
# Проверяем, есть ли у сообщения reply_markup
# Если кнопки уже убраны (generate_another), отправляем новое
try:
await update.callback_query.edit_message_text(
validation_text,
parse_mode='HTML',
reply_markup=InlineKeyboardMarkup(keyboard)
)
except Exception as e:
# Если редактирование не удалось, отправляем новое сообщение
logger.warning(f"Не удалось отредактировать сообщение: {e}")
await update.callback_query.message.reply_text(
validation_text,
parse_mode='HTML',
reply_markup=InlineKeyboardMarkup(keyboard)
)
else:
await update.message.reply_text(
validation_text,
parse_mode='HTML',
reply_markup=InlineKeyboardMarkup(keyboard)
)
return SELECT_FORMAT
except Exception as e:
logger.error(f"Ошибка валидации: {e}")
error_text = f"❌ Ошибка валидации:\n{str(e)}"
if update.callback_query:
await update.callback_query.edit_message_text(error_text)
else:
await update.message.reply_text(error_text)
await context.bot.send_message(
chat_id=update.effective_chat.id,
text="Выберите действие:",
reply_markup=get_main_keyboard()
)
return MAIN_MENU
async def show_xml_format_choice(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Показывает выбор формата XML"""
query = update.callback_query
text = (
"📄 Выберите формат XML:\n\n"
"🏛 ВерсФорм 5.03 (УПД/ТН)\n"
"Для ФНС и электронного документооборота\n"
"Стандартный формат для налоговой\n\n"
"💼 1С EnterpriseData 1.20\n"
"Для загрузки в программы 1С\n"
"Формат обмена с конфигурациями 1С"
)
keyboard = [
[InlineKeyboardButton(
"🏛 ВерсФорм 5.03 (УПД)",
callback_data="xml_format:versform"
)],
[InlineKeyboardButton(
"💼 1С EnterpriseData 1.20",
callback_data="xml_format:1c"
)],
[InlineKeyboardButton("❌ Отмена", callback_data="cancel")]
]
await query.edit_message_text(
text,
parse_mode='HTML',
reply_markup=InlineKeyboardMarkup(keyboard)
)
return SELECT_FORMAT
async def generate_xml(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Генерация XML документа"""
user_id = update.effective_user.id
query = update.callback_query
# Получаем выбранный формат XML
xml_format = context.user_data.get('xml_format', 'versform')
format_name = "ВерсФорм 5.03 (УПД)" if xml_format == 'versform' else "1С EnterpriseData 1.20"
await query.edit_message_text(
f"🔄 Генерация XML документа ({format_name})...",
reply_markup=None
)
try:
order_id = context.user_data.get('selected_order_id')
if not order_id:
raise ValueError("ID отгрузки не найден")
# Получаем клиента для пользователя
moysklad_client = get_moysklad_client(user_id)
# Получаем полные данные отгрузки
order_data = moysklad_client.get_full_order_data(order_id)
# Создаём генератор XML в зависимости от выбранного формата
if xml_format == '1c':
from xml_generator import XML1CGenerator
xml_gen = XML1CGenerator()
generate_method = xml_gen.generate_realization
caption_format = "1С EnterpriseData 1.20"
else:
xml_gen = XMLGenerator()
generate_method = xml_gen.generate_upd
caption_format = "УПД ВерсФорм 5.03"
# Создаём временную директорию
output_dir = Path("temp_documents")
output_dir.mkdir(exist_ok=True)
# Генерируем XML файл
filename = f"{order_data.get('name', order_id)}.xml"
output_path = str(output_dir / filename)
generate_method(order_data, output_path)
# Формируем информацию об отгрузке
order = order_data.get('order', {})
order_name = order.get('name', 'N/A')
order_moment = order.get('moment', '')[:10]
order_sum = order.get('sum', 0) / 100
counterparty = order_data.get('counterparty', {})
counterparty_name = counterparty.get('name', 'N/A')
caption = (
f"✅ XML документ создан!\n\n"
f"📄 Формат: {caption_format}\n"
f"📋 Отгрузка: №{order_name} от {order_moment}\n"
f"👤 Контрагент: {counterparty_name}\n"
f"💰 Сумма: {order_sum:,.2f} ₽"
)
# Кнопка для генерации другого файла
keyboard = [
[InlineKeyboardButton(
"📝 Сгенерировать другой файл",
callback_data="generate_another"
)],
[InlineKeyboardButton(
"🏠 Главное меню",
callback_data="cancel"
)]
]
# Отправляем файл с inline кнопками
await query.message.reply_document(
document=open(output_path, 'rb'),
caption=caption,
parse_mode='HTML',
reply_markup=InlineKeyboardMarkup(keyboard)
)
# Удаляем временный файл
os.remove(output_path)
# Показываем главное меню (ReplyKeyboard)
await context.bot.send_message(
chat_id=update.effective_chat.id,
text="Используйте кнопки ниже для продолжения работы:",
reply_markup=get_main_keyboard()
)
# НЕ очищаем user_data - нужны данные для "Сгенерировать другой"
return SELECT_FORMAT
except Exception as e:
logger.error(f"Ошибка генерации XML: {e}")
await query.message.reply_text(
f"❌ Ошибка генерации XML: {str(e)}\n\n"
f"Выберите действие:",
reply_markup=get_main_keyboard()
)
context.user_data.clear()
return MAIN_MENU
async def show_templates(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Показывает рекомендованный шаблон или список всех шаблонов"""
user_id = update.effective_user.id
query = update.callback_query
# Проверяем, запрошены ли все шаблоны
show_all = context.user_data.get('show_all_templates', False)
await query.edit_message_text(
"🔄 Загрузка шаблонов...",
reply_markup=None
)
try:
# Получаем клиента для пользователя
moysklad_client = get_moysklad_client(user_id)
# Получаем данные заказа для определения типа клиента
order_data = context.user_data.get('order_data')
counterparty = order_data.get('counterparty') if order_data else None
# Определяем тип клиента
# ИП: ИНН 12 цифр + (есть КПП ИЛИ "ИП" в названии) → УПД
# Юр.лицо: ИНН 10 цифр → УПД
# Физ.лицо: ИНН 12 цифр без КПП и без "ИП" в названии → ТН
is_individual = True # По умолчанию физ.лицо (ТН)
if counterparty:
inn = counterparty.get('inn', '')
kpp = counterparty.get('kpp', '')
name = counterparty.get('name', '').upper()
if len(inn) == 10:
# Юр.лицо (ИНН 10 цифр) → УПД
is_individual = False
elif len(inn) == 12:
# ИНН 12 цифр: проверяем КПП или "ИП" в названии
if kpp or 'ИП ' in name or name.startswith('ИП'):
# ИП → УПД
is_individual = False
# Иначе: физ.лицо → ТН
# Получаем все шаблоны (customtemplate + embeddedtemplate)
custom_templates = moysklad_client.get_templates(
"demand", "customtemplate"
)
embedded_templates = moysklad_client.get_templates(
"demand", "embeddedtemplate"
)
templates = custom_templates + embedded_templates
if not templates:
await query.message.reply_text(
"❌ Шаблоны не найдены.\n\n"
"Выберите действие:",
reply_markup=get_main_keyboard()
)
context.user_data.clear()
return MAIN_MENU
# Определяем рекомендованный шаблон по конкретному ID
recommended_template = None
# ID шаблонов по умолчанию
TN_DEFAULT_ID = "92743bbb-c8c8-42e2-8686-350eace81edd" # ТН НДС
UPD_DEFAULT_ID = "be6d441a-5982-48fe-ac21-46754e051f8c" # УПД тест
# Выбираем нужный ID в зависимости от типа клиента
default_id = TN_DEFAULT_ID if is_individual else UPD_DEFAULT_ID
# Ищем шаблон с нужным ID
for tpl in templates:
if tpl.get('id') == default_id:
recommended_template = tpl
break
# Если не нашли по ID - берём первый
if not recommended_template and templates:
recommended_template = templates[0]
keyboard = []
if show_all:
# Показываем все шаблоны
text = "📄 Все доступные шаблоны:\n\n"
for i, tpl in enumerate(templates, 1):
tpl_name = tpl.get('name', 'N/A')
tpl_id = tpl.get('id')
# Отмечаем рекомендованный
if tpl == recommended_template:
text += f"{i}. ⭐ {tpl_name} (рекомендовано)\n"
else:
text += f"{i}. {tpl_name}\n"
keyboard.append([
InlineKeyboardButton(
(f"⭐ {tpl_name}"
if tpl == recommended_template
else tpl_name),
callback_data=f"template:{tpl_id}"
)
])
keyboard.append([
InlineKeyboardButton("❌ Отмена", callback_data="cancel")
])
else:
# Показываем только рекомендованный шаблон
client_type = "физ. лица" if is_individual else "юр. лица/ИП"
tpl_name = recommended_template.get('name', 'N/A')
tpl_id = recommended_template.get('id')
text = (
f"📄 Рекомендованный шаблон\n\n"
f"Тип клиента: {client_type}\n"
f"Шаблон: {tpl_name}\n\n"
f"Используйте рекомендованный шаблон или выберите другой:"
)
keyboard = [
[InlineKeyboardButton(
f"✅ {tpl_name}",
callback_data=f"template:{tpl_id}"
)],
[InlineKeyboardButton(
"📋 Показать все шаблоны",
callback_data="show_all_templates"
)],
[InlineKeyboardButton("❌ Отмена", callback_data="cancel")]
]
await query.edit_message_text(
text,
parse_mode='HTML',
reply_markup=InlineKeyboardMarkup(keyboard)
)
return SELECT_TEMPLATE
except Exception as e:
logger.error(f"Ошибка получения шаблонов: {e}")
await query.message.reply_text(
f"❌ Ошибка: {str(e)}\n\n"
f"Выберите действие:",
reply_markup=get_main_keyboard()
)
context.user_data.clear()
return MAIN_MENU
async def generate_excel(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Генерация Excel документа"""
user_id = update.effective_user.id
query = update.callback_query
await query.edit_message_text(
"🔄 Генерация документа...",
reply_markup=None
)
try:
# Получаем клиента для пользователя
moysklad_client = get_moysklad_client(user_id)
order_id = context.user_data.get('selected_order_id')
template_id = context.user_data.get('template_id')
template_type = context.user_data.get(
'template_type', 'customtemplate'
)
# Создаём временную директорию если нужна
output_dir = Path("temp_documents")
output_dir.mkdir(exist_ok=True)
# Генерируем документ
output_path = moysklad_client.print_document(
entity_type="demand",
entity_id=order_id,
template_id=template_id,
template_type=template_type,
extension="xls",
output_path=str(output_dir / f"{order_id}.xls")
)
# Получаем информацию об отгрузке
order_data = context.user_data.get('order_data', {})
order = order_data.get('order', {})
order_name = order.get('name', 'N/A')
order_moment = order.get('moment', '')[:10]
order_sum = order.get('sum', 0) / 100
counterparty = order_data.get('counterparty', {})
counterparty_name = counterparty.get('name', 'N/A')
caption = (
f"✅ Excel документ создан!\n\n"
f"📋 Отгрузка: №{order_name} от {order_moment}\n"
f"👤 Контрагент: {counterparty_name}\n"
f"💰 Сумма: {order_sum:,.2f} ₽"
)
# Кнопка для генерации другого файла
keyboard = [
[InlineKeyboardButton(
"📝 Сгенерировать другой файл",
callback_data="generate_another"
)],
[InlineKeyboardButton(
"🏠 Главное меню",
callback_data="cancel"
)]
]
# Отправляем файл
await query.message.reply_document(
document=open(output_path, 'rb'),
caption=caption,
parse_mode='HTML',
reply_markup=InlineKeyboardMarkup(keyboard)
)
# Удаляем временный файл
os.remove(output_path)
# Показываем главное меню (ReplyKeyboard)
await context.bot.send_message(
chat_id=update.effective_chat.id,
text="Используйте кнопки ниже для продолжения работы:",
reply_markup=get_main_keyboard()
)
# НЕ очищаем user_data
return SELECT_FORMAT
except Exception as e:
logger.error(f"Ошибка генерации документа: {e}")
await query.message.reply_text(
f"❌ Ошибка: {str(e)}\n\n"
f"Выберите действие:",
reply_markup=get_main_keyboard()
)
context.user_data.clear()
return MAIN_MENU
async def generate_package(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Генерация комплекта документов (XML 1С + рекомендованный Excel)"""
user_id = update.effective_user.id
query = update.callback_query
await query.edit_message_text(
"🔄 Генерация комплекта документов...\n"
"Это может занять несколько секунд",
reply_markup=None
)
try:
order_id = context.user_data.get('selected_order_id')
if not order_id:
raise ValueError("ID отгрузки не найден")
# Получаем клиента для пользователя
moysklad_client = get_moysklad_client(user_id)
# Получаем полные данные отгрузки
order_data = context.user_data.get('order_data')
if not order_data:
order_data = moysklad_client.get_full_order_data(order_id)
context.user_data['order_data'] = order_data
order = order_data.get('order', {})
counterparty = order_data.get('counterparty', {})
# Создаём временную директорию
output_dir = Path("temp_documents")
output_dir.mkdir(exist_ok=True)
# 1. Генерируем XML 1С
from xml_generator import XML1CGenerator
xml_gen = XML1CGenerator()
xml_filename = f"{order.get('name', order_id)}_1c.xml"
xml_path = str(output_dir / xml_filename)
xml_gen.generate_realization(order_data, xml_path)
# 2. Определяем рекомендованный Excel шаблон
inn = counterparty.get('inn', '')
kpp = counterparty.get('kpp', '')
name = counterparty.get('name', '').upper()
# Определяем тип клиента
is_individual = True
if len(inn) == 10:
is_individual = False
elif len(inn) == 12:
if kpp or 'ИП ' in name or name.startswith('ИП'):
is_individual = False
# ID рекомендованных шаблонов
TN_DEFAULT_ID = "92743bbb-c8c8-42e2-8686-350eace81edd"
UPD_DEFAULT_ID = "be6d441a-5982-48fe-ac21-46754e051f8c"
template_id = TN_DEFAULT_ID if is_individual else UPD_DEFAULT_ID
# Генерируем Excel
excel_path = moysklad_client.print_document(
entity_type="demand",
entity_id=order_id,
template_id=template_id,
template_type="customtemplate",
extension="xls",
output_path=str(output_dir / f"{order_id}.xls")
)
# Формируем информацию об отгрузке
order_name = order.get('name', 'N/A')
order_moment = order.get('moment', '')[:10]
order_sum = order.get('sum', 0) / 100
counterparty_name = counterparty.get('name', 'N/A')
caption = (
f"✅ Комплект документов создан!\n\n"
f"📦 Включено:\n"
f" • XML 1С EnterpriseData 1.20\n"
f" • Excel (рекомендованный шаблон)\n\n"
f"📋 Отгрузка: №{order_name} от {order_moment}\n"
f"👤 Контрагент: {counterparty_name}\n"
f"💰 Сумма: {order_sum:,.2f} ₽"
)
# Кнопка для генерации другого файла
keyboard = [
[InlineKeyboardButton(
"📝 Сгенерировать другой файл",
callback_data="generate_another"
)],
[InlineKeyboardButton(
"🏠 Главное меню",
callback_data="cancel"
)]
]
# Отправляем XML файл
await query.message.reply_document(
document=open(xml_path, 'rb'),
caption="📄 XML 1С EnterpriseData 1.20"
)
# Отправляем Excel файл с полной информацией
await query.message.reply_document(
document=open(excel_path, 'rb'),
caption=caption,
parse_mode='HTML',
reply_markup=InlineKeyboardMarkup(keyboard)
)
# Удаляем временные файлы
os.remove(xml_path)
os.remove(excel_path)
# Показываем главное меню (ReplyKeyboard)
await context.bot.send_message(
chat_id=update.effective_chat.id,
text="Используйте кнопки ниже для продолжения работы:",
reply_markup=get_main_keyboard()
)
# НЕ очищаем user_data
return SELECT_FORMAT
except Exception as e:
logger.error(f"Ошибка генерации комплекта: {e}")
await query.message.reply_text(
f"❌ Ошибка генерации комплекта: {str(e)}\n\n"
f"Выберите действие:",
reply_markup=get_main_keyboard()
)
context.user_data.clear()
return MAIN_MENU
async def handle_text(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Обработчик текстовых сообщений"""
text = update.message.text
if text == "🔍 Поиск отгрузки":
return await search_order_prompt(update, context)
elif text == "📋 Список отгрузок":
return await show_orders_list(update, context)
elif text == "❓ Помощь":
return await help_command(update, context)
elif text == "⬅️ Назад":
return await start(update, context)
else:
await update.message.reply_text(
"❌ Неизвестная команда.\n"
"Используйте кнопки меню.",
reply_markup=get_main_keyboard()
)
return MAIN_MENU
def main():
"""Запуск бота"""
# Создаём приложение
application = Application.builder().token(
settings.telegram_bot_token
).build()
# Обработчики команд (не входящих в ConversationHandler)
application.add_handler(CommandHandler("help", help_command))
application.add_handler(CommandHandler("logout", logout_command))
application.add_handler(
CommandHandler("change_credentials", change_credentials_command)
)
# Обработчик состояний
conv_handler = ConversationHandler(
entry_points=[
CommandHandler("start", start),
MessageHandler(
filters.Regex("^🔍 Поиск отгрузки$"),
search_order_prompt
),
MessageHandler(
filters.Regex("^📋 Список отгрузок$"),
show_orders_list
),
MessageHandler(filters.TEXT & ~filters.COMMAND, handle_text),
],
states={
AUTH_LOGIN: [
MessageHandler(
filters.TEXT & ~filters.COMMAND,
auth_login
),
],
AUTH_PASSWORD: [
MessageHandler(
filters.TEXT & ~filters.COMMAND,
auth_password
),
],
MAIN_MENU: [
MessageHandler(filters.TEXT & ~filters.COMMAND, handle_text),
],
SEARCH_ORDER: [
MessageHandler(
filters.TEXT & ~filters.COMMAND,
search_order
),
],
SELECT_ORDER: [
CallbackQueryHandler(handle_callback),
MessageHandler(filters.TEXT & ~filters.COMMAND, handle_text),
],
SELECT_FORMAT: [
CallbackQueryHandler(handle_callback),
MessageHandler(filters.TEXT & ~filters.COMMAND, handle_text),
],
SELECT_TEMPLATE: [
CallbackQueryHandler(handle_callback),
MessageHandler(filters.TEXT & ~filters.COMMAND, handle_text),
],
},
fallbacks=[
CommandHandler("start", start),
],
)
application.add_handler(conv_handler)
# Запускаем бота
logger.info("🤖 Бот запущен!")
application.run_polling(allowed_updates=Update.ALL_TYPES)
if __name__ == "__main__":
main()