June 26
Документация AssistantBot
- Описание проекта
- Архитектура системы
- Установка и настройка
- Конфигурация
- Команды бота
- API и интеграции
- База данных
- Развертывание
- Мониторинг и логирование
- Устранение неполадок
- Разработка
🎯 Описание проекта
AssistantBot — это автоматизированный Telegram-бот для управления клиентскими коммуникациями с интеграцией Google Sheets. Бот автоматически отправляет запланированные сообщения клиентам, синхронизирует данные с Google Таблицами и предоставляет администраторам мощные инструменты для мониторинга и управления.
✨ Основные возможности
- 🕐 Автоматическая отправка сообщений по расписанию
- 📊 Синхронизация с Google Sheets в реальном времени
- 👥 Управление клиентами через Telegram интерфейс
- 📱 Мониторинг просроченных задач и уведомлений
- 🔍 Детальное логирование всех операций
- 🛡️ Разграничение прав доступа для администраторов и пользователей
- 🔄 Автоматическое восстановление после сбоев
🏗️ Архитектура системы
Компоненты системы
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Telegram Bot │ │ Google Sheets │ │ SQLite DB │
│ │ │ │ │ │
│ • Commands │◄──►│ • Users Data │◄──►│ • Schedule │
│ • Callbacks │ │ • Schedule │ │ • Logs │
│ • Messages │ │ • Status │ │ • Users │
└─────────────────┘ └─────────────────┘ └─────────────────┘
▲ ▲ ▲
│ │ │
▼ ▼ ▼
┌─────────────────────────────────────────────────────────────────┐
│ APScheduler Engine │
│ │
│ • Task Scheduling • Job Management • Automatic Retry │
└─────────────────────────────────────────────────────────────────┘
Структура проекта
tg-assistant-bot/ ├── app/ │ ├── core/ # Основные настройки │ │ ├── config.py # Конфигурация │ │ └── logging_config.py # Настройки логирования │ ├── db/ # База данных │ │ ├── database.py # Подключение к БД │ │ ├── models.py # Модели данных │ │ └── crud.py # CRUD операции │ ├── handlers/ # Обработчики команд │ │ ├── admin_commands.py # Команды администратора │ │ ├── user_commands.py # Пользовательские команды │ │ └── callbacks.py # Callback обработчики │ ├── services/ # Внешние сервисы │ │ └── google_sheets.py # Интеграция с Google Sheets │ ├── business_logic.py # Бизнес-логика │ ├── scheduler_setup.py # Настройка планировщика │ └── main.py # Точка входа ├── logs/ # Логи приложения ├── requirements.txt # Python зависимости ├── service_account.json # Ключи Google API ├── .env # Переменные окружения └── README.md # Описание проекта
🚀 Установка и настройка
Системные требования
Установка зависимостей
# Клонирование репозитория git clone https://github.com/your-repo/tg-assistant-bot.git cd tg-assistant-bot # Создание виртуального окружения python -m venv venv # Активация окружения # Linux/macOS: source venv/bin/activate # Windows: venv\Scripts\activate # Установка зависимостей pip install -r requirements.txt
Основные зависимости
python-telegram-bot==20.7 google-api-python-client==2.108.0 gspread==5.12.0 SQLAlchemy==2.0.23 APScheduler==3.10.4 loguru==0.7.2 pydantic==2.5.0 pydantic-settings==2.1.0
⚙️ Конфигурация
Переменные окружения (.env)
# Telegram Bot Token BOT_TOKEN=your_bot_token_here # Администраторы (ID через запятую) ADMIN_TELEGRAM_IDS=123456789,987654321 # Google Sheets GOOGLE_SHEETS_ID=your_google_sheets_id SERVICE_ACCOUNT_FILE=service_account.json # База данных DB_URL=sqlite:///bot_database.db # Настройки TIMEZONE=Europe/Moscow LOG_LEVEL=INFO
Настройка Google Sheets API
- Создайте проект в Google Cloud Console
- Включите Google Sheets API
- Создайте Service Account
- Скачайте JSON ключ и сохраните как
service_account.json - Предоставьте доступ Service Account к вашей таблице
Структура Google Sheets
Лист "Users"
Листы клиентов (например, "101")
🤖 Команды бота
Пользовательские команды
/start 🚀 Запуск бота и приветствие
/roman_dusenko 👤 Информация о Романе Дусенко
Административные команды
/sync 🔄 Принудительная синхронизация данных
/expired ⏰ Просроченные сообщения
/nearest 📅 Ближайшие запланированные сообщения
/status 📊 Статус и информация о боте
/noty🔔 Расписание уведомлений клиентов
/add_user➕ Добавить нового клиента
/debug_expired🔍 Отладка просроченных сообщений
Интерактивные элементы
- Кнопки навигации по логам и сообщениям
- Быстрые действия для отправки просроченных сообщений
- Подтверждения для критических операций
🔗 API и интеграции
Google Sheets API
class GoogleSheetsService:
def fetch_users(self) -> List[Dict]
def fetch_schedule(self, tab_name: str) -> List[Dict]
def update_status(self, tab_name: str, target_datetime_str: str,
target_time_str: str, target_text: str, new_status: str)
def append_user_row(self, user_data: Dict) -> bool
def clone_example_tab(self, new_tab_name: str, start_date: str) -> bool
Telegram Bot API
- Webhook или Long Polling режимы
- Обработка текстовых сообщений
- Callback Query для интерактивных кнопок
- Message Reactions для быстрых ответов
- File uploads для логов и отчетов
💾 База данных
Модели данных
User (Пользователи)
CREATE TABLE users (
id INTEGER PRIMARY KEY,
name VARCHAR(255) NOT NULL,
username VARCHAR(255),
telegram_id INTEGER,
chat_id INTEGER,
thread_id INTEGER,
tab_name VARCHAR(255),
start_date DATE,
active BOOLEAN DEFAULT TRUE,
created_at TIMESTAMP,
updated_at TIMESTAMP
);
ScheduleEntry (Расписание)
CREATE TABLE schedule_entries (
id INTEGER PRIMARY KEY,
user_id INTEGER REFERENCES users(id),
scheduled_datetime_utc TIMESTAMP,
message_text TEXT,
status VARCHAR(50),
bot_message_id INTEGER,
sent_at TIMESTAMP,
answer_text TEXT,
answered_at TIMESTAMP,
created_at TIMESTAMP,
updated_at TIMESTAMP
);
BotLog (Логи)
CREATE TABLE bot_logs (
id INTEGER PRIMARY KEY,
level VARCHAR(20),
message TEXT,
event_type VARCHAR(100),
details TEXT,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CRUD операции
# Основные операции crud.create_user(db, user_data) crud.get_user_by_username(db, username) crud.get_active_users(db) crud.create_schedule_entry(db, entry_data) crud.update_schedule_entry_status(db, entry_id, new_status) crud.log_bot_event(db, level, message, event_type, details)
🚀 Развертывание
Локальный запуск
# Активация окружения source venv/bin/activate # Запуск бота python -m app.main
Развертывание на Ubuntu Server
1. Подготовка сервера
# Обновление системы sudo apt update && sudo apt upgrade -y # Установка Python и зависимостей sudo apt install python3 python3-pip python3-venv git -y # Клонирование проекта git clone https://github.com/your-repo/tg-assistant-bot.git cd tg-assistant-bot
2. Настройка systemd сервиса
# Создание файла сервиса sudo nano /etc/systemd/system/assistant-bot.service
[Unit] Description=🤖 AssistantBot - Telegram Bot Service After=network.target Wants=network.target [Service] Type=simple User=deploy Group=deploy WorkingDirectory=/home/deploy/tg-assistant-bot Environment=PATH=/home/deploy/tg-assistant-bot/venv/bin ExecStart=/home/deploy/tg-assistant-bot/venv/bin/python -m app.main Restart=always RestartSec=10 StandardOutput=journal StandardError=journal SyslogIdentifier=assistant-bot Environment=PYTHONPATH=/home/deploy/tg-assistant-bot Environment=PYTHONUNBUFFERED=1 [Install] WantedBy=multi-user.target
3. Запуск сервиса
# Перезагрузка конфигурации sudo systemctl daemon-reload # Включение автозапуска sudo systemctl enable assistant-bot.service # Запуск сервиса sudo systemctl start assistant-bot.service # Проверка статуса sudo systemctl status assistant-bot.service
Docker развертывание
Dockerfile
FROM python:3.11-slim RUN apt-get update && apt-get install -y gcc && rm -rf /var/lib/apt/lists/* RUN useradd --create-home --shell /bin/bash botuser WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . RUN chown -R botuser:botuser /app USER botuser CMD ["python", "-m", "app.main"]
docker-compose.yml
version: '3.8'
services:
assistant-bot:
build: .
container_name: assistant-bot
restart: unless-stopped
volumes:
- ./logs:/app/logs
- ./service_account.json:/app/service_account.json:ro
- ./bot_database.db:/app/bot_database.db
environment:
- PYTHONUNBUFFERED=1
networks:
- bot-network
networks:
bot-network:
driver: bridge
📊 Мониторинг и логирование
Система логирования
Бот использует loguru для красивого и структурированного логирования:
# Примеры логов
logger.info("🚀 Запуск AssistantBot...")
logger.warning("⏳ Таймаут соединения с Telegram API")
logger.error("❌ Ошибка при синхронизации данных")
Уровни логирования
- DEBUG: Детальная отладочная информация
- INFO: Общая информация о работе
- WARNING: Предупреждения о потенциальных проблемах
- ERROR: Ошибки в работе
- CRITICAL: Критические ошибки
Просмотр логов
# Логи systemd сервиса sudo journalctl -u assistant-bot.service -f # Файловые логи tail -f logs/bot.log # Логи в боте /log - команда для просмотра через Telegram
Мониторинг состояния
- Проверка здоровья каждые 60 секунд
- Автоматическое восстановление при сбоях
- Уведомления администраторов о критических ошибках
- Статистика планировщика задач
🔧 Устранение неполадок
Частые проблемы
1. Ошибки подключения к Telegram API
telegram.error.TimedOut: Timed out httpcore.ConnectTimeout
- Проверьте интернет соединение
- Убедитесь, что токен бота корректный
- Проверьте настройки прокси (если используется)
2. Ошибки Google Sheets API
gspread.exceptions.APIError: [403] Forbidden
- Проверьте права доступа Service Account к таблице
- Убедитесь, что Google Sheets API включен
- Проверьте корректность
service_account.json
3. Ошибки базы данных
sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) database is locked
- Остановите все экземпляры бота
- Проверьте права доступа к файлу БД
- Убедитесь, что только один процесс использует БД
4. Проблемы с планировщиком
Run time of job was missed by X days
- Это нормально при перезапуске бота
- Задачи автоматически переносятся или выполняются
- Используйте
/syncдля принудительной синхронизации
Диагностические команды
# Проверка статуса сервиса sudo systemctl status assistant-bot.service # Просмотр логов sudo journalctl -u assistant-bot.service -n 100 # Проверка процессов ps aux | grep python # Проверка портов netstat -tlnp | grep python # Тест подключения к Telegram curl -s "https://api.telegram.org/bot$BOT_TOKEN/getMe"
Восстановление после сбоя
- Остановите сервис
sudo systemctl stop assistant-bot.service - Проверьте логи
sudo journalctl -u assistant-bot.service -n 50 - Исправьте проблему (конфигурация, права доступа и т.д.)
- Перезапустите сервис
sudo systemctl start assistant-bot.service
👨💻 Разработка
Структура кода
Основные модули
main.py: Точка входа, инициализация ботаbusiness_logic.py: Основная бизнес-логикаscheduler_setup.py: Настройка планировщика задач
Обработчики
admin_commands.py: Команды для администраторовuser_commands.py: Пользовательские командыcallbacks.py: Обработчики inline кнопок
Сервисы
Добавление новых команд
# В admin_commands.py или user_commands.py
async def new_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
user = update.effective_user
# Проверка прав доступа (для админ команд)
if user.id not in settings.admin_telegram_ids:
await update.message.reply_text("❌ Нет доступа")
return
# Логика команды
await update.message.reply_text("✅ Команда выполнена")
# В main.py
application.add_handler(CommandHandler("new_command", new_command))
Добавление новых callback обработчиков
# В callbacks.py
async def handle_new_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
query = update.callback_query
await query.answer()
# Логика обработки
await query.edit_message_text("✅ Обработано")
# В main.py
application.add_handler(CallbackQueryHandler(
callbacks.handle_new_callback,
pattern=r"^new_pattern_\d+quot;
))
Работа с базой данных
# Создание новой модели
class NewModel(Base):
__tablename__ = "new_table"
id = Column(Integer, primary_key=True)
name = Column(String(255))
created_at = Column(DateTime, default=datetime.utcnow)
# CRUD операции
def create_new_record(db: Session, data: dict):
record = NewModel(**data)
db.add(record)
db.commit()
db.refresh(record)
return record
Тестирование
# Установка тестовых зависимостей pip install pytest pytest-asyncio # Запуск тестов pytest tests/ # Тестирование конкретного модуля pytest tests/test_business_logic.py -v
Стиль кода
- PEP 8 для форматирования
- Type hints для всех функций
- Docstrings для публичных методов
- Логирование для всех важных операций
async def example_function(user_id: int, message: str) -> bool:
"""
Пример функции с правильным оформлением.
Args:
user_id: ID пользователя
message: Текст сообщения
Returns:
True если успешно, False иначе
"""
logger.info(f"📨 Обработка сообщения для пользователя {user_id}")
try:
# Логика функции
return True
except Exception as e:
logger.error(f"❌ Ошибка в example_function: {e}")
return False
📞 Поддержка
Контакты
Полезные ссылки
- Python Telegram Bot Documentation
- Google Sheets API Documentation
- APScheduler Documentation
- SQLAlchemy Documentation
📄 Лицензия
MIT License - подробности в файле LICENSE
🔄 История изменений
v1.0.0 (2025-06-26)
- ✅ Первый релиз
- 🤖 Основной функционал бота
- 📊 Интеграция с Google Sheets
- ⏰ Система планирования задач
- 🔍 Система логирования
June 26, 11:28
0 views
0 reactions
0 reposts