Untitled
unknown
plain_text
9 months ago
28 kB
13
Indexable
import logging
import sqlite3
import pandas as pd
import os
from datetime import datetime
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import Application, CommandHandler, CallbackQueryHandler, MessageHandler, filters, ContextTypes, ConversationHandler
# Настройка логирования
logging.basicConfig(
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=logging.INFO
)
logger = logging.getLogger(__name__)
# 🔑 ЗАМЕНИТЕ НА ВАШ ТОКЕН
BOT_TOKEN = "ВАШ_НОВЫЙ_ТОКЕН_ЗДЕСЬ"
# 👥 Добавьте ваш ID
ALLOWED_USERS = [6947728011]
# Состояния для ConversationHandler
BRAND, MODEL, LICENSE_PLATE, PURCHASE_PRICE, EXPENSE_NAME, EXPENSE_AMOUNT, SALE_PRICE = range(7)
class Database:
def __init__(self):
self.init_database()
def init_database(self):
"""Инициализация базы данных"""
conn = sqlite3.connect('autoservice.db')
c = conn.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS cars
(id INTEGER PRIMARY KEY AUTOINCREMENT,
brand TEXT NOT NULL,
model TEXT NOT NULL,
license_plate TEXT UNIQUE NOT NULL,
status TEXT DEFAULT 'active',
purchase_price REAL DEFAULT 0,
sale_price REAL DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)''')
c.execute('''CREATE TABLE IF NOT EXISTS expenses
(id INTEGER PRIMARY KEY AUTOINCREMENT,
car_id INTEGER NOT NULL,
description TEXT NOT NULL,
amount REAL NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (car_id) REFERENCES cars(id))''')
conn.commit()
conn.close()
logger.info("✅ База данных инициализирована")
def execute_query(self, query, params=()):
"""Выполнить запрос и вернуть результат"""
try:
conn = sqlite3.connect('autoservice.db')
c = conn.cursor()
c.execute(query, params)
result = c.fetchall()
conn.commit()
conn.close()
return result
except Exception as e:
logger.error(f"Ошибка запроса: {e}")
return None
def add_car(self, brand, model, license_plate):
"""Добавить автомобиль"""
return self.execute_query(
"INSERT INTO cars (brand, model, license_plate) VALUES (?, ?, ?)",
(brand, model, license_plate.upper())
)
def get_active_cars(self):
"""Получить активные автомобили"""
return self.execute_query(
"SELECT id, brand, model, license_plate FROM cars WHERE status = 'active'"
)
def get_car_by_id(self, car_id):
"""Получить автомобиль по ID"""
result = self.execute_query("SELECT * FROM cars WHERE id = ?", (car_id,))
return result[0] if result else None
def get_car_expenses(self, car_id):
"""Получить расходы автомобиля"""
return self.execute_query(
"SELECT description, amount FROM expenses WHERE car_id = ?", (car_id,)
)
def update_purchase_price(self, car_id, price):
"""Обновить цену покупки"""
return self.execute_query(
"UPDATE cars SET purchase_price = ? WHERE id = ?", (price, car_id)
)
def update_sale_price(self, car_id, price):
"""Обновить цену продажи"""
return self.execute_query(
"UPDATE cars SET sale_price = ? WHERE id = ?", (price, car_id)
)
def add_expense(self, car_id, description, amount):
"""Добавить расход"""
return self.execute_query(
"INSERT INTO expenses (car_id, description, amount) VALUES (?, ?, ?)",
(car_id, description, amount)
)
def move_to_archive(self, car_id):
"""Переместить в архив"""
return self.execute_query(
"UPDATE cars SET status = 'archived' WHERE id = ?", (car_id,)
)
def get_archived_cars(self):
"""Получить архивные автомобили"""
return self.execute_query("SELECT * FROM cars WHERE status = 'archived'")
def delete_archived_car(self, car_id):
"""Удалить автомобиль из архива"""
# Сначала удаляем связанные расходы
self.execute_query("DELETE FROM expenses WHERE car_id = ?", (car_id,))
# Затем удаляем автомобиль
return self.execute_query("DELETE FROM cars WHERE id = ? AND status = 'archived'", (car_id,))
def clear_archive(self):
"""Очистить весь архив"""
# Удаляем расходы архивных автомобилей
self.execute_query("DELETE FROM expenses WHERE car_id IN (SELECT id FROM cars WHERE status = 'archived')")
# Удаляем архивные автомобили
return self.execute_query("DELETE FROM cars WHERE status = 'archived'")
# Инициализация базы данных
db = Database()
def check_access(user_id):
"""Проверка доступа пользователя"""
return user_id in ALLOWED_USERS
# ===== ОСНОВНЫЕ ОБРАБОТЧИКИ =====
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Главное меню"""
if not check_access(update.effective_user.id):
await update.message.reply_text("❌ Доступ запрещен")
return
keyboard = [
[InlineKeyboardButton("🚗 Добавить авто", callback_data="add_car")],
[InlineKeyboardButton("📊 Активные авто", callback_data="active_cars")],
[InlineKeyboardButton("📁 Архив", callback_data="archive")]
]
reply_markup = InlineKeyboardMarkup(keyboard)
await update.message.reply_text("🏠 Главное меню автосервиса:", reply_markup=reply_markup)
async def handle_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Обработчик всех callback'ов"""
query = update.callback_query
await query.answer()
if not check_access(query.from_user.id):
await query.edit_message_text("❌ Доступ запрещен")
return
data = query.data
if data == "main_menu":
await show_main_menu(update, context)
elif data == "active_cars":
await show_active_cars(update, context)
elif data == "archive":
await show_archive_menu(update, context)
elif data.startswith("car_"):
car_id = int(data.split("_")[1])
await show_car_details(update, context, car_id)
elif data.startswith("add_purchase_"):
car_id = int(data.split("_")[2])
context.user_data["current_car_id"] = car_id
await query.edit_message_text("Введите цену покупки автомобиля:")
return PURCHASE_PRICE
elif data.startswith("add_expense_"):
car_id = int(data.split("_")[2])
context.user_data["current_car_id"] = car_id
await query.edit_message_text("Введите описание расхода:")
return EXPENSE_NAME
elif data.startswith("add_sale_"):
car_id = int(data.split("_")[2])
context.user_data["current_car_id"] = car_id
await query.edit_message_text("Введите цену продажи автомобиля:")
return SALE_PRICE
elif data.startswith("archive_car_"):
car_id = int(data.split("_")[2])
db.move_to_archive(car_id)
await query.edit_message_text("✅ Автомобиль перемещен в архив!")
await show_active_cars(update, context)
elif data == "generate_excel":
await generate_excel_report(update, context)
elif data.startswith("delete_car_"):
car_id = int(data.split("_")[2])
db.delete_archived_car(car_id)
await query.edit_message_text("✅ Автомобиль удален из архива!")
await show_archive_menu(update, context)
elif data == "clear_archive":
keyboard = [
[InlineKeyboardButton("✅ Да, очистить", callback_data="confirm_clear")],
[InlineKeyboardButton("❌ Нет, отмена", callback_data="archive")]
]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.edit_message_text(
"⚠️ Вы уверены, что хотите полностью очистить архив? Это действие нельзя отменить!",
reply_markup=reply_markup
)
elif data == "confirm_clear":
db.clear_archive()
await query.edit_message_text("✅ Архив полностью очищен!")
await show_main_menu(update, context)
else:
await query.edit_message_text(f"Неизвестная команда: {data}")
async def show_main_menu(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Показать главное меню"""
keyboard = [
[InlineKeyboardButton("🚗 Добавить авто", callback_data="add_car")],
[InlineKeyboardButton("📊 Активные авто", callback_data="active_cars")],
[InlineKeyboardButton("📁 Архив", callback_data="archive")]
]
reply_markup = InlineKeyboardMarkup(keyboard)
if update.callback_query:
await update.callback_query.edit_message_text("🏠 Главное меню автосервиса:", reply_markup=reply_markup)
else:
await update.message.reply_text("🏠 Главное меню автосервиса:", reply_markup=reply_markup)
async def show_active_cars(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Показать активные автомобили"""
query = update.callback_query
await query.answer()
cars = db.get_active_cars()
if not cars:
keyboard = [[InlineKeyboardButton("🔙 Назад", callback_data="main_menu")]]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.edit_message_text("🚫 Нет активных автомобилей", reply_markup=reply_markup)
return
keyboard = []
for car_id, brand, model, license_plate in cars:
button_text = f"{brand} {model} ({license_plate})"
keyboard.append([InlineKeyboardButton(button_text, callback_data=f"car_{car_id}")])
keyboard.append([InlineKeyboardButton("🔙 Назад", callback_data="main_menu")])
reply_markup = InlineKeyboardMarkup(keyboard)
await query.edit_message_text("📊 Активные автомобили:", reply_markup=reply_markup)
async def show_car_details(update: Update, context: ContextTypes.DEFAULT_TYPE, car_id: int):
"""Показать детали автомобиля и меню действий"""
query = update.callback_query
await query.answer()
car = db.get_car_by_id(car_id)
if not car:
await query.edit_message_text("❌ Автомобиль не найден")
return
car_id, brand, model, license_plate, status, purchase_price, sale_price, created_at = car
# Получаем расходы
expenses = db.get_car_expenses(car_id)
total_expenses = sum(expense[1] for expense in expenses) if expenses else 0
# Расчет прибыли
profit = (sale_price or 0) - (purchase_price or 0) - total_expenses
# Формируем текст
text = f"🚗 {brand} {model} ({license_plate})\n\n"
text += f"💰 Цена покупки: {purchase_price or 'не указана'}\n"
text += f"📊 Общие расходы: {total_expenses}\n"
text += f"🎯 Цена продажи: {sale_price or 'не указана'}\n"
text += f"💵 Прибыль: {profit}\n\n"
if expenses:
text += "📋 Расходы:\n"
for desc, amount in expenses:
text += f" • {desc}: {amount}\n"
# Кнопки действий
keyboard = [
[InlineKeyboardButton("💰 Добавить цену покупки", callback_data=f"add_purchase_{car_id}")],
[InlineKeyboardButton("💸 Добавить расход", callback_data=f"add_expense_{car_id}")],
[InlineKeyboardButton("🎯 Добавить цену продажи", callback_data=f"add_sale_{car_id}")],
[InlineKeyboardButton("📁 Перенести в архив", callback_data=f"archive_car_{car_id}")],
[InlineKeyboardButton("🔙 Назад к активным", callback_data="active_cars")]
]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.edit_message_text(text, reply_markup=reply_markup)
async def show_archive_menu(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Показать меню архива"""
query = update.callback_query
await query.answer()
archived_cars = db.get_archived_cars()
if not archived_cars:
keyboard = [
[InlineKeyboardButton("📊 Выгрузить Excel", callback_data="generate_excel")],
[InlineKeyboardButton("🔙 Назад", callback_data="main_menu")]
]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.edit_message_text("📁 Архив пуст", reply_markup=reply_markup)
return
# Расчет общей прибыли
total_profit = 0
car_details = []
for car in archived_cars:
car_id, brand, model, license_plate, status, purchase_price, sale_price, created_at = car
# Сумма расходов для этого автомобиля
expenses = db.get_car_expenses(car_id)
total_expenses = sum(expense[1] for expense in expenses) if expenses else 0
profit = (sale_price or 0) - (purchase_price or 0) - total_expenses
total_profit += profit
car_details.append({
'id': car_id,
'brand': brand,
'model': model,
'license_plate': license_plate,
'profit': profit
})
# Формируем текст
text = f"📁 Архив автомобилей\n\n"
text += f"💵 Общая прибыль: {total_profit:.2f}\n"
text += f"🚗 Автомобилей в архиве: {len(archived_cars)}\n\n"
# Кнопки для каждого автомобиля
keyboard = []
for car in car_details:
profit_color = "🟢" if car['profit'] >= 0 else "🔴"
button_text = f"{car['brand']} {car['model']} ({profit_color}{car['profit']})"
keyboard.append([InlineKeyboardButton(button_text, callback_data=f"car_{car['id']}")])
# Кнопка удаления для каждого авто
keyboard.append([InlineKeyboardButton(f"🗑️ Удалить {car['brand']} {car['model']}", callback_data=f"delete_car_{car['id']}")])
# Общие кнопки архива
keyboard.extend([
[InlineKeyboardButton("📊 Выгрузить Excel", callback_data="generate_excel")],
[InlineKeyboardButton("🗑️ Очистить весь архив", callback_data="clear_archive")],
[InlineKeyboardButton("🔙 Назад", callback_data="main_menu")]
])
reply_markup = InlineKeyboardMarkup(keyboard)
await query.edit_message_text(text, reply_markup=reply_markup)
async def generate_excel_report(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Генерация Excel отчета"""
query = update.callback_query
await query.answer()
try:
archived_cars = db.get_archived_cars()
if not archived_cars:
await query.edit_message_text("❌ В архиве нет данных для отчета")
return
# Подготовка данных
data = []
total_profit = 0
for car in archived_cars:
car_id, brand, model, license_plate, status, purchase_price, sale_price, created_at = car
# Получаем расходы
expenses = db.get_car_expenses(car_id)
total_expenses = sum(expense[1] for expense in expenses) if expenses else 0
# Расчет прибыли
profit = (sale_price or 0) - (purchase_price or 0) - total_expenses
total_profit += profit
data.append({
'Марка': brand,
'Модель': model,
'Номер': license_plate,
'Цена покупки': purchase_price or 0,
'Цена продажи': sale_price or 0,
'Общие расходы': total_expenses,
'Прибыль': profit,
'Дата добавления': created_at
})
# Создаем DataFrame
df = pd.DataFrame(data)
# Создаем Excel файл
filename = f"archive_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx"
with pd.ExcelWriter(filename, engine='openpyxl') as writer:
# Лист с автомобилями
df.to_excel(writer, sheet_name='Автомобили', index=False)
# Сводный отчет
summary_data = {
'Показатель': ['Общая прибыль', 'Количество автомобилей', 'Средняя прибыль', 'Максимальная прибыль', 'Минимальная прибыль'],
'Значение': [
total_profit,
len(df),
df['Прибыль'].mean() if not df.empty else 0,
df['Прибыль'].max() if not df.empty else 0,
df['Прибыль'].min() if not df.empty else 0
]
}
summary_df = pd.DataFrame(summary_data)
summary_df.to_excel(writer, sheet_name='Сводка', index=False)
# Отправляем файл
await query.message.reply_document(
document=open(filename, 'rb'),
filename=filename
)
# Удаляем временный файл
os.remove(filename)
await query.edit_message_text("✅ Excel отчет сгенерирован и отправлен!")
except Exception as e:
logger.error(f"Ошибка генерации отчета: {e}")
await query.edit_message_text("❌ Ошибка при генерации отчета")
# ===== CONVERSATION HANDLERS =====
async def add_car_start(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Начало добавления автомобиля"""
query = update.callback_query
await query.answer()
await query.edit_message_text("Введите марку автомобиля:")
return BRAND
async def add_car_brand(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Получение марки автомобиля"""
context.user_data['brand'] = update.message.text
await update.message.reply_text("Введите модель автомобиля:")
return MODEL
async def add_car_model(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Получение модели автомобиля"""
context.user_data['model'] = update.message.text
await update.message.reply_text("Введите номерной знак:")
return LICENSE_PLATE
async def add_car_license_plate(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Сохранение автомобиля"""
license_plate = update.message.text.upper()
try:
db.add_car(context.user_data['brand'], context.user_data['model'], license_plate)
await update.message.reply_text(
f"✅ Автомобиль {context.user_data['brand']} {context.user_data['model']} "
f"с номером {license_plate} добавлен в активные!"
)
except Exception as e:
await update.message.reply_text("❌ Ошибка: автомобиль с таким номером уже существует")
context.user_data.clear()
await show_main_menu(update, context)
return ConversationHandler.END
async def save_purchase_price(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Сохранение цены покупки"""
try:
price = float(update.message.text)
car_id = context.user_data['current_car_id']
db.update_purchase_price(car_id, price)
await update.message.reply_text(f"✅ Цена покупки {price} добавлена!")
except ValueError:
await update.message.reply_text("❌ Пожалуйста, введите корректное число")
return PURCHASE_PRICE
context.user_data.clear()
await show_car_details(update, context, car_id)
return ConversationHandler.END
async def save_expense_name(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Сохранение названия расхода"""
context.user_data['expense_name'] = update.message.text
await update.message.reply_text("Введите сумму расхода:")
return EXPENSE_AMOUNT
async def save_expense_amount(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Сохранение суммы расхода"""
try:
amount = float(update.message.text)
car_id = context.user_data['current_car_id']
description = context.user_data['expense_name']
db.add_expense(car_id, description, amount)
await update.message.reply_text(f"✅ Расход '{description}' на сумму {amount} добавлен!")
except ValueError:
await update.message.reply_text("❌ Пожалуйста, введите корректное число")
return EXPENSE_AMOUNT
context.user_data.clear()
await show_car_details(update, context, car_id)
return ConversationHandler.END
async def save_sale_price(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Сохранение цены продажи"""
try:
price = float(update.message.text)
car_id = context.user_data['current_car_id']
db.update_sale_price(car_id, price)
await update.message.reply_text(f"✅ Цена продажи {price} добавлена!")
except ValueError:
await update.message.reply_text("❌ Пожалуйста, введите корректное число")
return SALE_PRICE
context.user_data.clear()
await show_car_details(update, context, car_id)
return ConversationHandler.END
async def cancel(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Отмена операции"""
await update.message.reply_text("❌ Операция отменена")
context.user_data.clear()
await show_main_menu(update, context)
return ConversationHandler.END
def main():
"""Основная функция запуска бота"""
try:
application = Application.builder().token(BOT_TOKEN).build()
# Обработчики команд
application.add_handler(CommandHandler("start", start))
# Обработчики callback'ов для главного меню
application.add_handler(CallbackQueryHandler(handle_callback, pattern="^main_menu$"))
application.add_handler(CallbackQueryHandler(handle_callback, pattern="^active_cars$"))
application.add_handler(CallbackQueryHandler(handle_callback, pattern="^archive$"))
application.add_handler(CallbackQueryHandler(handle_callback, pattern="^car_"))
application.add_handler(CallbackQueryHandler(handle_callback, pattern="^add_purchase_"))
application.add_handler(CallbackQueryHandler(handle_callback, pattern="^add_expense_"))
application.add_handler(CallbackQueryHandler(handle_callback, pattern="^add_sale_"))
application.add_handler(CallbackQueryHandler(handle_callback, pattern="^archive_car_"))
application.add_handler(CallbackQueryHandler(handle_callback, pattern="^generate_excel$"))
application.add_handler(CallbackQueryHandler(handle_callback, pattern="^delete_car_"))
application.add_handler(CallbackQueryHandler(handle_callback, pattern="^clear_archive$"))
application.add_handler(CallbackQueryHandler(handle_callback, pattern="^confirm_clear$"))
# ConversationHandler для добавления автомобиля
add_car_conv = ConversationHandler(
entry_points=[CallbackQueryHandler(add_car_start, pattern="^add_car$")],
states={
BRAND: [MessageHandler(filters.TEXT & ~filters.COMMAND, add_car_brand)],
MODEL: [MessageHandler(filters.TEXT & ~filters.COMMAND, add_car_model)],
LICENSE_PLATE: [MessageHandler(filters.TEXT & ~filters.COMMAND, add_car_license_plate)],
},
fallbacks=[CommandHandler('cancel', cancel)]
)
application.add_handler(add_car_conv)
# ConversationHandler для добавления цены покупки
purchase_conv = ConversationHandler(
entry_points=[CallbackQueryHandler(handle_callback, pattern="^add_purchase_")],
states={
PURCHASE_PRICE: [MessageHandler(filters.TEXT & ~filters.COMMAND, save_purchase_price)],
},
fallbacks=[CommandHandler('cancel', cancel)]
)
application.add_handler(purchase_conv)
# ConversationHandler для добавления расхода
expense_conv = ConversationHandler(
entry_points=[CallbackQueryHandler(handle_callback, pattern="^add_expense_")],
states={
EXPENSE_NAME: [MessageHandler(filters.TEXT & ~filters.COMMAND, save_expense_name)],
EXPENSE_AMOUNT: [MessageHandler(filters.TEXT & ~filters.COMMAND, save_expense_amount)],
},
fallbacks=[CommandHandler('cancel', cancel)]
)
application.add_handler(expense_conv)
# ConversationHandler для добавления цены продажи
sale_conv = ConversationHandler(
entry_points=[CallbackQueryHandler(handle_callback, pattern="^add_sale_")],
states={
SALE_PRICE: [MessageHandler(filters.TEXT & ~filters.COMMAND, save_sale_price)],
},
fallbacks=[CommandHandler('cancel', cancel)]
)
application.add_handler(sale_conv)
logger.info("🤖 Бот запускается...")
print("=" * 50)
print("🚀 ПОЛНЫЙ АВТОСЕРВИС БОТ ЗАПУЩЕН!")
print("➡️ Напишите /start в Telegram")
print("=" * 50)
application.run_polling()
except Exception as e:
logger.error(f"❌ Ошибка запуска бота: {e}")
if __name__ == "__main__":
main()Editor is loading...
Leave a Comment