AGorokhov/vk-sales-bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial clean history
CI/CD / Test (push) Failing after 4m32s
Details
CI/CD / Deploy (push) Has been skipped
Details

Browse Source
This commit is contained in:
User
2026-05-11 16:46:25 +03:00
commit bf89e671d8
33 changed files with 1333 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.dockerignore
+18
View File
@@ -0,0 +1,18 @@
.git
.gitignore
.gitea
.kilo
.vscode
README.md
__pycache__/
*.pyc
*.pyo
*.pyd
.pytest_cache/
.venv/
venv/
.env
logs/
data/
tests/
docs/
.env.example
+8
View File
@@ -0,0 +1,8 @@
VK_GROUP_ID=123456789
VK_TOKEN=your_vk_group_token_here
ADMIN_IDS=12345678
DATA_DIR=data
LEADS_FILE=data/leads.xlsx
BACKUP_DIR=data/backups
LOG_LEVEL=INFO
TIMEZONE=Europe/Moscow
.gitea/workflows/ci-cd.yml
+131
View File
@@ -0,0 +1,131 @@
name: CI/CD
on:
push:
pull_request:
workflow_dispatch:
jobs:
test:
name: Test
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: https://github.com/actions/checkout@v4
- name: Set up Python
uses: https://github.com/actions/setup-python@v5
with:
python-version: "3.11"
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt pytest
- name: Run tests
run: python -m pytest tests
- name: Check syntax
run: python -m compileall -q config core services utils main.py start.py tests
- name: Optional Docker build check
run: |
if command -v docker >/dev/null 2>&1 && docker info >/dev/null 2>&1; then
docker build -t vk-sales-bot:ci .
else
echo "Docker is not available on this Gitea runner, skipping local image build."
fi
deploy:
name: Deploy
runs-on: ubuntu-latest
needs:
- test
if: ${{ gitea.ref == 'refs/heads/main' && (gitea.event_name == 'push' || gitea.event_name == 'workflow_dispatch') }}
env:
DEPLOY_PATH: /opt/vk-sales-bot
SSH_HOST: infocyber.pro
SSH_KNOWN_HOSTS: ${{ secrets.SSH_KNOWN_HOSTS }}
SSH_PORT: "22"
SSH_PRIVATE_KEY: ${{ secrets.SSH_PRIVATE_KEY }}
SSH_USER: root
steps:
- name: Checkout
uses: https://github.com/actions/checkout@v4
- name: Validate deploy settings
run: |
set -eu
: "${SSH_PRIVATE_KEY:?Missing SSH_PRIVATE_KEY secret}"
echo "Deploy path: ${DEPLOY_PATH}"
- name: Install deploy tools
run: |
set -eu
if command -v rsync >/dev/null 2>&1 && command -v ssh >/dev/null 2>&1; then
exit 0
fi
if ! command -v apt-get >/dev/null 2>&1; then
echo "apt-get is not available on this runner. Install rsync and openssh-client manually."
exit 1
fi
SUDO=""
if command -v sudo >/dev/null 2>&1; then
SUDO="sudo"
fi
$SUDO apt-get update
$SUDO apt-get install -y openssh-client rsync
- name: Configure SSH
run: |
set -eu
mkdir -p "$HOME/.ssh"
chmod 700 "$HOME/.ssh"
printf '%s\n' "$SSH_PRIVATE_KEY" > "$HOME/.ssh/id_ed25519"
chmod 600 "$HOME/.ssh/id_ed25519"
SSH_PORT="${SSH_PORT:-22}"
if [ -n "${SSH_KNOWN_HOSTS:-}" ]; then
printf '%s\n' "$SSH_KNOWN_HOSTS" > "$HOME/.ssh/known_hosts"
else
ssh-keyscan -p "$SSH_PORT" "$SSH_HOST" > "$HOME/.ssh/known_hosts"
fi
chmod 644 "$HOME/.ssh/known_hosts"
- name: Prepare remote directory
run: |
set -eu
SSH_PORT="${SSH_PORT:-22}"
ssh -i "$HOME/.ssh/id_ed25519" -p "$SSH_PORT" "$SSH_USER@$SSH_HOST" \
"mkdir -p '$DEPLOY_PATH' '$DEPLOY_PATH/data' '$DEPLOY_PATH/data/backups' '$DEPLOY_PATH/logs' && test -w '$DEPLOY_PATH'"
- name: Sync project to server
run: |
set -eu
SSH_PORT="${SSH_PORT:-22}"
rsync -az --delete \
--exclude '.env' \
--exclude '.git/' \
--exclude '.gitea/' \
--exclude '.kilo/' \
--exclude '.vscode/' \
--exclude '.venv/' \
--exclude '__pycache__/' \
--exclude '*.pyc' \
--exclude 'data/' \
--exclude 'docs/' \
--exclude 'logs/' \
--exclude 'tests/' \
-e "ssh -i $HOME/.ssh/id_ed25519 -p $SSH_PORT" \
./ "$SSH_USER@$SSH_HOST:$DEPLOY_PATH/"
- name: Restart application
run: |
set -eu
SSH_PORT="${SSH_PORT:-22}"
ssh -i "$HOME/.ssh/id_ed25519" -p "$SSH_PORT" "$SSH_USER@$SSH_HOST" \
"cd '$DEPLOY_PATH' && test -f .env && docker compose up -d --build --remove-orphans && docker compose ps"
.gitignore
+49
View File
@@ -0,0 +1,49 @@
# Secrets
.env
.env.*
!.env.example
# Python
__pycache__/
*.py[cod]
*.pyc.*
*.pyd
*.pyo
*.so
.pytest_cache/
.mypy_cache/
.ruff_cache/
.hypothesis/
.tox/
.nox/
.coverage
.coverage.*
htmlcov/
# Packaging / build
build/
dist/
site/
.eggs/
*.egg
*.egg-info/
pip-wheel-metadata/
# Virtual environments
.venv/
venv/
env/
ENV/
# Runtime data
logs/
data/
# Editor / OS noise
.idea/
*.swp
*.swo
*.tmp
.DS_Store
Thumbs.db
Desktop.ini
.kilo/kilo.jsonc
+3
View File
@@ -0,0 +1,3 @@
{
"$schema": "https://app.kilo.ai/config.json"
}
.vscode/settings.json
Vendored
+3
View File
@@ -0,0 +1,3 @@
{
"python-envs.defaultEnvManager": "ms-python.python:system"
}
Dockerfile
+21
View File
@@ -0,0 +1,21 @@
FROM python:3.11-slim
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
PIP_DISABLE_PIP_VERSION_CHECK=1 \
PIP_PROGRESS_BAR=off \
OPENBLAS_NUM_THREADS=1 \
OMP_NUM_THREADS=1 \
MKL_NUM_THREADS=1 \
NUMEXPR_NUM_THREADS=1
WORKDIR /app
COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
RUN mkdir -p /app/data /app/logs
CMD ["python", "main.py"]
README.md
+160
View File
@@ -0,0 +1,160 @@
# VK Sales Bot
VK-бот для сбора заявок в личных сообщениях сообщества. Бот проводит пользователя по сценарию диалога, сохраняет лиды в Excel, пишет логи и делает резервные копии.
## Что умеет
- запрашивает согласие на обработку персональных данных;
- собирает ФИО, телефон и удобное время звонка;
- умеет сохранить данные родителя или опекуна;
- обрабатывает базовые возражения пользователя;
- сохраняет заявки в `data/leads.xlsx`;
- делает резервные копии в `data/backups/`;
- поддерживает административные команды в VK.
## Как работает сценарий
Диалог построен на FSM и проходит по шагам:
`согласие -> ФИО -> данные родителя (опционально) -> телефон -> удобное время -> подтверждение`
После подтверждения лид записывается в Excel-файл. При отказе или возражениях бот может сохранить заявку со статусом `postponed` или `rejected`.
## Требования
- Python 3.11+
- access token сообщества VK с правами `messages` и `groups`
- ID группы VK
## Быстрый старт
1. Создайте виртуальное окружение и установите зависимости:
```powershell
python -m venv .venv
.venv\Scripts\activate
pip install -r requirements.txt
```
2. Скопируйте пример конфигурации:
```powershell
Copy-Item .env.example .env
```
3. Заполните `.env`.
4. При желании проверьте токен:
```powershell
python start.py
```
5. Запустите бота:
```powershell
python main.py
```
## Переменные окружения
| Переменная | Описание |
| --- | --- |
| `VK_GROUP_ID` | ID сообщества VK |
| `VK_TOKEN` | токен группы VK |
| `ADMIN_IDS` | список ID администраторов через запятую |
| `DATA_DIR` | директория для рабочих данных |
| `LEADS_FILE` | путь к Excel-файлу с лидами |
| `BACKUP_DIR` | директория для резервных копий |
| `LOG_LEVEL` | уровень логирования |
| `TIMEZONE` | часовой пояс приложения |
Пример уже есть в [.env.example](./.env.example).
## Административные команды
Команды доступны только пользователям из `ADMIN_IDS`:
- `/status` — проверить, что бот работает;
- `/export` — создать копию файла с лидами;
- `/backup` — создать резервную копию вручную;
- `/stats` — показать статистику по лидам;
- `/reload` — перезагрузить тексты из `config/phrases.py` без перезапуска.
## Docker
Для запуска через Docker:
```powershell
docker compose up --build -d
```
Контейнер монтирует локальные каталоги:
- `./data -> /app/data`
- `./logs -> /app/logs`
## CI/CD
В репозитории добавлен workflow для Gitea Actions: [`.gitea/workflows/ci-cd.yml`](./.gitea/workflows/ci-cd.yml).
Что делает pipeline:
- на `push` и `pull_request` запускает тесты;
- если у runner есть доступ к Docker, дополнительно проверяет сборку образа;
- на `push` в `main` деплоит проект на сервер по `SSH` и выполняет `docker compose up -d --build`.
Что нужно настроить в Gitea:
1. Включить `Repository Actions` в настройках репозитория.
2. Поднять runner с меткой `ubuntu-latest`.
Если у runner другая метка, замените `runs-on` в workflow.
3. Добавить secrets репозитория:
- `SSH_PRIVATE_KEY` — приватный ключ для входа по SSH;
- `SSH_KNOWN_HOSTS` — публичный host key сервера, рекомендуется для безопасной проверки хоста.
Требования на сервере:
- установлен `docker` и `docker compose`;
- установлен `rsync`;
- проект деплоится в `/opt/vk-sales-bot`;
- в `/opt/vk-sales-bot` уже есть рабочий `.env`, потому что workflow его не перезаписывает;
- workflow подключается как `root` на `infocyber.pro`.
Важно:
- workflow уже привязан к `root@infocyber.pro` и пути `/opt/vk-sales-bot`;
- локальный alias для подключения к серверу у вас называется `ea2go`;
- бот не публикует порты в `compose.yaml`, поэтому сам по себе не должен конфликтовать с Gitea на том же сервере.
## Тесты
В репозитории есть базовые тесты для валидации и логики проекта.
```powershell
pip install pytest
pytest tests/
```
## Структура проекта
- [main.py](./main.py) — основная точка входа;
- [start.py](./start.py) — быстрая проверка токена VK;
- [config](./config) — настройки и текстовые фразы;
- [core](./core) — FSM, модели, валидация и экспорт;
- [services](./services) — интеграция с VK API;
- [utils](./utils) — логирование, middleware и бэкапы;
- [tests](./tests) — тесты;
- [docs](./docs) — дополнительные инструкции.
## Данные и логи
- лиды сохраняются в `data/leads.xlsx`;
- резервные копии создаются в `data/backups/`;
- логи пишутся в `logs/bot.log`.
## Документация
- [Руководство оператора](./docs/user_guide.md)
- [Документация разработчика](./docs/developer_guide.md)
compose.yaml
+10
View File
@@ -0,0 +1,10 @@
services:
bot:
build: .
container_name: vk-sales-bot
restart: unless-stopped
env_file:
- .env
volumes:
- ./data:/app/data
- ./logs:/app/logs
config/__init__.py
View File
config/phrases.py
+93
View File
@@ -0,0 +1,93 @@
# Все тексты бота вынесены сюда для быстрой правки
ASK_PARENT_CONSENT = (
"👪 Укажите, пожалуйста, контактные данные родителя или опекуна. "
"Вы можете предоставить их сейчас или пропустить этот шаг. "
"Напишите 'пропустить', чтобы пропустить, или укажите ФИО родителя (два слова на кириллице):"
)
ASK_PARENT_FIO = "Укажите ФИО родителя (Фамилия Имя):"
ASK_PARENT_PHONE = "Укажите телефон родителя в формате +7 (XXX) XXX-XX-XX:"
PARENT_FIO_INVALID = "❌ ФИО родителя должно состоять из двух слов (Фамилия Имя) кириллицей."
PARENT_PHONE_INVALID = "❌ Неверный формат телефона родителя. Пример: +7 (912) 345-67-89"
PARENT_DATA_SKIPPED = "✅ Данные родителя пропущены. Продолжаем оформление."
PARENT_DATA_SAVED = "✅ Данные родителя сохранены."
GREETING = (
"Здравствуйте! 👋 Меня зовут Алексей, я специалист отдела продаж.\n\n"
"Подскажите, как к вам обращаться? (Фамилия, Имя, Отчество)"
)
ASK_FI = "Пожалуйста, укажите вашу фамилию и имя (два слова на кириллице):"
FIO_INVALID = "❌ ФИО должно состоять из двух слов (Фамилия Имя) и содержать только буквы кириллицы. Попробуйте ещё раз:"
ASK_PHONE = "Отлично! Теперь укажите ваш номер телефона в формате +7 (XXX) XXX-XX-XX:"
PHONE_INVALID = "❌ Неверный формат. Нужно +7 и 10 цифр после кода страны. Пример: +7 (912) 345-67-89"
ASK_TIME = "Укажите удобное время по МСК для звонка (например: 'завтра после 18', 'в среду утром', 'сегодня в 15:00'):"
TIME_INVALID = "❌ Не понял время. Попробуйте сказать иначе: 'завтра в 10 утра', 'сегодня после обеда'."
CONSENT_TEXT = (
"🔐 Перед началом работы, пожалуйста, ознакомьтесь с условиями обработки персональных данных.\n\n"
"Мы собираем ваши ФИО, номер телефона и предпочтительное время звонка для связи с вами "
"по вопросам, связанным с услугами нашей компании. Ваши данные будут использоваться только "
"сотрудниками нашей компании и не будут переданы третьим лицам.\n\n"
"Дайте согласие на обработку персональных данных:"
)
ASK_CONSENT = "Пожалуйста, подтвердите согласие на обработку персональных данных (Да / Нет):"
CONSENT_YES = "✅ Спасибо за согласие! Продолжим оформление заявки."
CONSENT_NO = "❌ К сожалению, без согласия на обработку данных мы не можем продолжить. Спасибо за обращение! Хорошего дня."
BUTTON_CONSENT_YES = "✅ Да, согласен"
BUTTON_CONSENT_NO = "❌ Нет, не согласен"
CONSENT_YES_PHRASES = ["да", "yes", "согласен", "даю согласие", "хорошо", "ok"]
CONSENT_NO_PHRASES = [
"нет", "no", "не согласен", "отказываюсь", "не даю согласие",
"нет, не согласен", "не хочу давать согласие", "отказ"
]
CONFIRM_MESSAGE = (
"Проверьте данные:\n"
"ФИ: {fio}\n"
"Телефон: {phone}\n"
"Время звонка по МСК: {time}\n"
"{parent_info}"
"Всё верно? (Да / Нет)"
)
CONFIRM_YES = "Спасибо! Ваши данные сохранены. Менеджер свяжется с вами в указанное время. Хорошего дня! 🌟"
CONFIRM_NO = "Давайте начнём заново. " + GREETING
OBJECTION_NOT_NOW = (
"Понимаю, возможно сейчас неудобно. Я оставлю заявку в статусе 'отложено'. "
"Напишите 'начать заново', когда будете готовы продолжить."
)
OBJECTION_THINK = (
"Хорошо, подумайте. Я сохраню текущие данные. Для продолжения напишите 'продолжить'."
)
OBJECTION_NO_PHONE = (
"К сожалению, без номера телефона мы не сможем с вами связаться. "
"Если передумаете – напишите 'начать заново'."
)
# Кнопки
BUTTON_YES = "✅ Да"
BUTTON_NO = "❌ Нет"
BUTTON_RESTART = "🔄 Начать заново"
BUTTON_CONTINUE = "➡️ Продолжить"
BUTTON_REGISTER = "📝 Зарегистрироваться"
# Сообщения для группы
MSG_REGISTER_INSTRUCTIONS = (
"📝 Для регистрации нажмите кнопку ниже.\n\n"
"Если кнопка не работает, напишите мне в личные сообщения — "
"откройте мой профиль и отправьте любое сообщение.\n\n"
"👉 [Написать в ЛС](vk.com/im-redirect?target=st%3D{user_id})"
)
# Команды
CMD_START = "/start"
CMD_STATUS = "/status"
CMD_EXPORT = "/export"
CMD_BACKUP = "/backup"
CMD_STATS = "/stats"
CMD_RELOAD = "/reload"
config/settings.py
+25
View File
@@ -0,0 +1,25 @@
import os
from dotenv import load_dotenv
from pathlib import Path
load_dotenv()
BASE_DIR = Path(__file__).parent.parent
VK_GROUP_ID = int(os.getenv("VK_GROUP_ID"))
VK_TOKEN = os.getenv("VK_TOKEN")
ADMIN_IDS = [int(x.strip()) for x in os.getenv("ADMIN_IDS", "").split(",") if x.strip()]
DATA_DIR = Path(os.getenv("DATA_DIR", BASE_DIR / "data"))
LEADS_FILE = Path(os.getenv("LEADS_FILE", DATA_DIR / "leads.xlsx"))
BACKUP_DIR = Path(os.getenv("BACKUP_DIR", DATA_DIR / "backups"))
LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
TIMEZONE = os.getenv("TIMEZONE", "Europe/Moscow")
# Создаём директории
DATA_DIR.mkdir(exist_ok=True)
BACKUP_DIR.mkdir(exist_ok=True)
# Настройки валидации
PHONE_MASK = "+7 (___) ___-__-__"
PHONE_LENGTH = 11 # цифр после +7
core/__init__.py
View File
core/exceptions.py
+5
View File
@@ -0,0 +1,5 @@
class ValidationError(Exception):
pass
class ExportError(Exception):
pass
core/exporter.py
+76
View File
@@ -0,0 +1,76 @@
import pandas as pd
from pathlib import Path
from filelock import FileLock
from datetime import datetime
from loguru import logger
from core.models import LeadData
from config import settings
import shutil
class ExcelExporter:
def __init__(self, file_path: Path):
self.file_path = file_path
self.lock_path = file_path.with_suffix('.lock')
self._init_file()
def _init_file(self):
"""Создаёт файл с заголовками, если не существует"""
if not self.file_path.exists():
df = pd.DataFrame(columns=[
"timestamp", "user_id", "full_name", "phone",
"preferred_time", "status", "consent_given",
"parent_full_name", "parent_phone"
])
df.to_excel(self.file_path, index=False, engine='openpyxl')
def save_lead(self, lead: LeadData):
"""Атомарная дозапись с блокировкой"""
lock = FileLock(self.lock_path)
with lock:
try:
# Читаем существующий DataFrame
df = pd.read_excel(self.file_path, engine='openpyxl')
# Добавляем новую строку
new_row = pd.DataFrame([{
"timestamp": lead.timestamp,
"user_id": lead.user_id,
"full_name": lead.full_name,
"phone": lead.phone,
"preferred_time": lead.preferred_time,
"status": lead.status,
"consent_given": lead.consent_given,
"parent_full_name": lead.parent_full_name,
"parent_phone": lead.parent_phone
}])
df = pd.concat([df, new_row], ignore_index=True)
# Записываем обратно
df.to_excel(self.file_path, index=False, engine='openpyxl')
logger.info(f"Lead saved for user {lead.user_id}")
except Exception as e:
logger.error(f"Failed to save lead: {e}")
raise
def backup(self):
"""Создаёт копию файла с меткой времени"""
if self.file_path.exists():
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_path = settings.BACKUP_DIR / f"leads_backup_{timestamp}.xlsx"
shutil.copy2(self.file_path, backup_path)
logger.info(f"Backup created: {backup_path}")
def get_stats(self) -> dict:
"""Возвращает статистику по лидам"""
if not self.file_path.exists():
return {"total": 0, "today": 0, "statuses": {}}
df = pd.read_excel(self.file_path, engine='openpyxl')
total = len(df)
today = len(df[pd.to_datetime(df['timestamp']).dt.date == datetime.now().date()])
statuses = df['status'].value_counts().to_dict()
return {"total": total, "today": today, "statuses": statuses}
def export(self, target_path: Path = None):
"""Принудительный экспорт (копия)"""
if target_path is None:
target_path = settings.DATA_DIR / f"export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx"
shutil.copy2(self.file_path, target_path)
return target_path
core/fsm.py
+233
View File
@@ -0,0 +1,233 @@
from enum import StrEnum, auto
from typing import Dict, Optional
from loguru import logger
from core.models import LeadData
from core.validator import validate_fio, validate_phone, parse_time
from core.exporter import ExcelExporter
from config import phrases
class DialogState(StrEnum):
START = auto()
ASK_CONSENT = auto()
COLLECT_FI = auto() # ФИО пользователя
ASK_PARENT_DATA = auto() # Спрашиваем, хочет ли пользователь ввести данные родителей
COLLECT_PARENT_FIO = auto() # Ввод ФИО родителя
COLLECT_PARENT_PHONE = auto() # Ввод телефона родителя
COLLECT_PHONE = auto()
COLLECT_TIME = auto()
CONFIRM = auto()
FINISHED = auto()
OBJECTION_HANDLED = auto()
class DialogManager:
def __init__(self, exporter: ExcelExporter):
self.exporter = exporter
self.sessions: Dict[int, dict] = {}
# def _handle_age(self, user_id: int, text: str) -> str:
# result = validate_age(text)
# if not result.is_valid:
# return result.error_message + "\n" + phrases.ASK_AGE
# session = self.sessions[user_id]
# session["data"].age = result.value
# logger.info(f"User {user_id} age = {result.value}")
# if result.value < 18:
# session["state"] = DialogState.ASK_PARENT_DATA
# logger.info(f"State changed to ASK_PARENT_DATA for user {user_id}")
# return phrases.ASK_PARENT_CONSENT
# else:
# session["state"] = DialogState.COLLECT_PHONE
# logger.info(f"Age >=18, state changed to COLLECT_PHONE")
# return phrases.ASK_PHONE
def _handle_parent_consent(self, user_id: int, text: str) -> str:
"""Обрабатывает ответ на вопрос про данные родителей.
Если пользователь вводит 'пропустить' и т.п. — пропускает.
Иначе считает, что это и есть ФИО родителя, и переходит к телефону.
"""
session = self.sessions[user_id]
text_lower = text.lower().strip()
if text_lower in ("пропустить", "нет", "не надо", "skip", "не хочу"):
session["state"] = DialogState.COLLECT_PHONE
logger.info(f"Parent data skipped for user {user_id}")
return phrases.PARENT_DATA_SKIPPED + "\n" + phrases.ASK_PHONE
else:
# Считаем, что пользователь сразу ввёл ФИО родителя
result = validate_fio(text)
if not result.is_valid:
return phrases.PARENT_FIO_INVALID + "\n" + phrases.ASK_PARENT_CONSENT
session["data"].parent_full_name = result.value
session["state"] = DialogState.COLLECT_PARENT_PHONE
logger.info(f"Parent FIO saved for user {user_id}, state changed to COLLECT_PARENT_PHONE")
return phrases.ASK_PARENT_PHONE
def _handle_parent_fio(self, user_id: int, text: str) -> str:
result = validate_fio(text) # используем ту же валидацию, что и для ФИО пользователя
if not result.is_valid:
# Подменяем сообщение об ошибке для родителя
return phrases.PARENT_FIO_INVALID + "\n" + phrases.ASK_PARENT_FIO
session = self.sessions[user_id]
session["data"].parent_full_name = result.value
session["state"] = DialogState.COLLECT_PARENT_PHONE
return phrases.ASK_PARENT_PHONE
def _handle_parent_phone(self, user_id: int, text: str) -> str:
result = validate_phone(text) # используем существующий валидатор телефона
if not result.is_valid:
return phrases.PARENT_PHONE_INVALID + "\n" + phrases.ASK_PARENT_PHONE
session = self.sessions[user_id]
session["data"].parent_phone = result.value
session["state"] = DialogState.COLLECT_PHONE
return phrases.PARENT_DATA_SAVED + "\n" + phrases.ASK_PHONE
def handle_message(self, user_id: int, text: str) -> str:
"""Обрабатывает сообщение и возвращает ответ бота"""
logger.info(f"=== HANDLE: user {user_id}, text='{text}', current state={self.sessions.get(user_id, {}).get('state')} ===")
# Команда /start или "начать заново"
if text.lower() in (phrases.CMD_START, phrases.BUTTON_RESTART.lower(), "начать заново"):
return self._reset_dialog(user_id)
session = self.sessions.get(user_id)
if not session:
return self._start_dialog(user_id)
state = session["state"]
lead = session["data"]
# Обработка возражений на любом этапе
if self._is_objection(text):
return self._handle_objection(user_id, text)
# Продолжение после возражения "подумаю"
if state == DialogState.OBJECTION_HANDLED and text.lower() == phrases.BUTTON_CONTINUE.lower():
# Возвращаемся к предыдущему состоянию
prev_state = session.get("prev_state", DialogState.COLLECT_FI)
session["state"] = prev_state
# Не рекурсивный вызов, а продолжение основного потока
state = session["state"]
lead = session["data"]
# Обычные состояния
if state == DialogState.ASK_CONSENT:
return self._handle_consent(user_id, text)
if state == DialogState.COLLECT_FI:
return self._handle_fio(user_id, text)
elif state == DialogState.ASK_PARENT_DATA: # Спрашиваем про данные родителей
return self._handle_parent_consent(user_id, text)
elif state == DialogState.COLLECT_PARENT_FIO:
return self._handle_parent_fio(user_id, text)
elif state == DialogState.COLLECT_PARENT_PHONE:
return self._handle_parent_phone(user_id, text)
elif state == DialogState.COLLECT_PHONE:
return self._handle_phone(user_id, text)
elif state == DialogState.COLLECT_TIME:
return self._handle_time(user_id, text)
elif state == DialogState.CONFIRM:
return self._handle_confirm(user_id, text)
def _start_dialog(self, user_id: int) -> str:
self.sessions[user_id] = {
"state": DialogState.ASK_CONSENT, # ← было COLLECT_FI
"data": LeadData(user_id=user_id)
}
return phrases.ASK_CONSENT
def _reset_dialog(self, user_id: int) -> str:
if user_id in self.sessions:
del self.sessions[user_id]
return self._start_dialog(user_id)
def _handle_fio(self, user_id: int, text: str) -> str:
result = validate_fio(text)
if not result.is_valid:
return result.error_message + "\n" + phrases.ASK_FI
session = self.sessions[user_id]
session["data"].full_name = result.value
session["state"] = DialogState.ASK_PARENT_DATA
return phrases.ASK_PARENT_CONSENT
def _handle_phone(self, user_id: int, text: str) -> str:
result = validate_phone(text)
if not result.is_valid:
return result.error_message + "\n" + phrases.ASK_PHONE
session = self.sessions[user_id]
session["data"].phone = result.value
session["state"] = DialogState.COLLECT_TIME
return phrases.ASK_TIME
def _handle_time(self, user_id: int, text: str) -> str:
result = parse_time(text)
if not result.is_valid:
return result.error_message + "\n" + phrases.ASK_TIME
session = self.sessions[user_id]
session["data"].preferred_time = result.value
parent_info = ""
if session["data"].parent_full_name and session["data"].parent_phone:
parent_info = f"Родитель: {session['data'].parent_full_name}, тел: {session['data'].parent_phone}\n"
elif session["data"].parent_full_name:
parent_info = f"Родитель: {session['data'].parent_full_name} (телефон не указан)\n"
elif session["data"].parent_phone:
parent_info = f"Родитель: телефон {session['data'].parent_phone} (ФИО не указано)\n"
session["state"] = DialogState.CONFIRM
return phrases.CONFIRM_MESSAGE.format(
fio=session["data"].full_name,
phone=session["data"].phone,
time=session["data"].preferred_time,
parent_info=parent_info
)
def _handle_confirm(self, user_id: int, text: str) -> str:
if text.lower() in ("да", phrases.BUTTON_YES.lower()):
lead = self.sessions[user_id]["data"]
lead.status = "completed"
self.exporter.save_lead(lead)
self.sessions[user_id]["state"] = DialogState.FINISHED
return phrases.CONFIRM_YES
elif text.lower() in ("нет", phrases.BUTTON_NO.lower()):
return self._reset_dialog(user_id)
else:
return "Пожалуйста, ответьте 'Да' или 'Нет'."
def _is_objection(self, text: str) -> bool:
obj_phrases = ["не сейчас", "подумаю", "не хочу оставлять телефон", "не хочу говорить номер"]
return any(phrase in text.lower() for phrase in obj_phrases)
def _handle_objection(self, user_id: int, text: str) -> str:
session = self.sessions.get(user_id)
if not session:
return self._start_dialog(user_id)
lead = session["data"]
if "не сейчас" in text.lower():
lead.status = "postponed"
self.exporter.save_lead(lead)
self.sessions[user_id]["state"] = DialogState.FINISHED
return phrases.OBJECTION_NOT_NOW
elif "подумаю" in text.lower():
# Сохраняем текущее состояние, переходим в OBJECTION_HANDLED
session["prev_state"] = session["state"]
session["state"] = DialogState.OBJECTION_HANDLED
return phrases.OBJECTION_THINK
elif any(phrase in text.lower() for phrase in ["не хочу оставлять телефон", "не хочу говорить номер"]):
lead.status = "rejected"
self.exporter.save_lead(lead)
self.sessions[user_id]["state"] = DialogState.FINISHED
return phrases.OBJECTION_NO_PHONE
return self.handle_message(user_id, text) # fallback
def _handle_consent(self, user_id: int, text: str) -> str:
text_lower = text.lower().strip()
session = self.sessions.get(user_id)
if not session:
return self._start_dialog(user_id)
if any(phrase in text_lower for phrase in phrases.CONSENT_YES_PHRASES):
session["data"].consent_given = True
session["state"] = DialogState.COLLECT_FI # ← было COLLECT_FIO
return phrases.CONSENT_YES + "\n" + phrases.ASK_FI # ← было ASK_FIO
elif any(phrase in text_lower for phrase in phrases.CONSENT_NO_PHRASES):
del self.sessions[user_id]
return phrases.CONSENT_NO
else:
return "Пожалуйста, ответьте 'Да' (согласен) или 'Нет' (не согласен)."
core/models.py
+20
View File
@@ -0,0 +1,20 @@
from datetime import datetime
from pydantic import BaseModel, Field
from typing import Optional
class LeadData(BaseModel):
user_id: int
full_name: Optional[str] = None
phone: Optional[str] = None
preferred_time: Optional[str] = None
timestamp: datetime = Field(default_factory=datetime.now)
status: str = "new"
consent_given: bool = False
parent_full_name: Optional[str] = None
parent_phone: Optional[str] = None
class ValidationResult(BaseModel):
is_valid: bool
value: Optional[str] = None
error_message: Optional[str] = None
core/validator.py
+52
View File
@@ -0,0 +1,52 @@
from loguru import logger
import re
import dateparser
from datetime import datetime
from core.models import ValidationResult
from config import settings
def validate_fio(text: str) -> ValidationResult:
"""Проверка ФИ: 2 слова, кириллица, возможен дефис"""
text = text.strip()
words = text.split()
if len(words) != 2:
return ValidationResult(is_valid=False, error_message="Должно быть 2 слова")
pattern = r'^[А-ЯЁа-яё\-]+$'
for word in words:
if not re.match(pattern, word):
return ValidationResult(is_valid=False, error_message="Только буквы кириллицы и дефис")
return ValidationResult(is_valid=True, value=text)
def validate_phone(phone: str) -> ValidationResult:
"""Принимает +7 (XXX) XXX-XX-XX или просто 10 цифр после 7/8"""
cleaned = re.sub(r'\D', '', phone)
if len(cleaned) == 11 and cleaned.startswith('7'):
cleaned = '8' + cleaned[1:]
if len(cleaned) == 10:
cleaned = '8' + cleaned
if len(cleaned) == 11 and cleaned.startswith('8'):
formatted = f"+7 ({cleaned[1:4]}) {cleaned[4:7]}-{cleaned[7:9]}-{cleaned[9:]}"
return ValidationResult(is_valid=True, value=formatted)
return ValidationResult(is_valid=False, error_message="Некорректный номер")
def parse_time(text: str) -> ValidationResult:
"""Парсит естественно-языковое время, возвращает строку для отображения"""
text_lower = text.lower().strip()
if any(phrase in text_lower for phrase in ["любое время", "не важно", "в любое", "когда удобно"]):
return ValidationResult(is_valid=True, value="Любое удобное время")
try:
parsed = dateparser.parse(
text,
settings={'TIMEZONE': settings.TIMEZONE, 'RETURN_AS_TIMEZONE_AWARE': False}
)
if parsed and parsed > datetime.now():
formatted = parsed.strftime("%d.%m.%Y в %H:%M")
return ValidationResult(is_valid=True, value=formatted)
except Exception as e:
logger.warning(f"Failed to parse time '{text}': {e}")
return ValidationResult(is_valid=False, error_message="Не удалось распознать время. Укажите, например: 'завтра после 18', 'в среду утром' или 'любое время'.")
docs/developer_guide.md
+30
View File
@@ -0,0 +1,30 @@
# Документация разработчика
## Архитектура
Проект построен на **конечном автомате (FSM)**. Диалог управляется классом `DialogManager`.
Состояния: `COLLECT_FI` → `COLLECT_PHONE` → `COLLECT_TIME` → `CONFIRM` → `FINISHED`.
Поддерживаются возражения и перезапуск.
**Компоненты:**
- `core/fsm.py` – логика диалога.
- `core/validator.py` – валидация ФИО, телефона, парсинг времени.
- `core/exporter.py` – работа с Excel (атомарная запись, бэкапы).
- `services/vk_bot.py` – интеграция с VK API, обработка команд.
- `utils/middleware.py` – мидлвары (логирование, аналитика).
## Добавление нового поля в анкету
1. Добавить текст в `config/phrases.py`.
2. Расширить модель `LeadData` в `core/models.py`.
3. Добавить состояние в `DialogState` и методы в `DialogManager`.
4. Добавить валидатор в `core/validator.py`.
5. Обновить `ExcelExporter._init_file` и метод сохранения.
## Тестирование
Установите тестовые зависимости: `pip install pytest pytest-asyncio responses`.
Запуск: `pytest tests/`.
Пример теста для валидатора:
```python
def test_validate_fio():
from core.validator import validate_fio
assert validate_fio("Иванов Иван Иванович").is_valid
assert not validate_fio("John Doe").is_valid
docs/user_guide.md
+22
View File
@@ -0,0 +1,22 @@
# Руководство оператора "Сотрудник первой линии"
## Быстрый старт
1. Убедитесь, что установлен Python 3.11+ и все зависимости (`pip install -r requirements.txt`).
2. Скопируйте `.env.example` в `.env` и укажите токен группы VK.
3. Запустите бота: `python main.py`
4. Проверьте статус: напишите боту `/status` (только для администраторов, указанных в ADMIN_IDS).
## Управление
- `/status` – проверить, работает ли бот.
- `/export` – создать копию файла лидов с меткой времени.
- `/backup` – вручную создать резервную копию.
- `/stats` – показать статистику (всего лидов, за сегодня, по статусам).
- `/reload` – перезагрузить текстовые фразы без перезапуска.
## Просмотр данных
Файл с лидами находится в `data/leads.xlsx`. Резервные копии – в `data/backups/`.
## Решение проблем
- Бот не отвечает: проверьте токен и права доступа группы (нужны `messages` и `groups`).
- Ошибка записи в файл: убедитесь, что папка `data` существует и доступна для записи.
- Бот не видит сообщения: группа должна быть публичной или добавьте бота в беседу.
main.py
+28
View File
@@ -0,0 +1,28 @@
from utils.logger import logger
from core.exporter import ExcelExporter
from core.fsm import DialogManager
from services.vk_bot import VKBot
from config import settings
from utils.middleware import logging_middleware
def main():
"""Точка входа в приложение"""
logger.info("Initializing bot...")
exporter = ExcelExporter(settings.LEADS_FILE)
fsm = DialogManager(exporter)
bot = VKBot(fsm, exporter)
# Добавляем мидлвары (можно расширить)
bot.middlewares.add(logging_middleware)
try:
bot.run()
except KeyboardInterrupt:
logger.info("Bot stopped")
except Exception as e:
logger.exception(f"Fatal error: {e}")
if __name__ == "__main__":
main()
requirements.txt
+9
View File
@@ -0,0 +1,9 @@
vk_api==11.9.9
pandas==2.2.0
openpyxl==3.1.2
python-dotenv==1.0.0
loguru==0.7.2
pydantic==2.5.3
dateparser==1.2.0
filelock==3.13.1
schedule==1.2.0
services/__init__.py
View File
services/vk_bot.py
+169
View File
@@ -0,0 +1,169 @@
import vk_api
from vk_api.bot_longpoll import VkBotLongPoll, VkBotEventType
from vk_api.keyboard import VkKeyboard, VkKeyboardColor
from loguru import logger
from config import settings, phrases
from core.fsm import DialogManager
from core.exporter import ExcelExporter
from utils.backup import schedule_daily_backup
from utils.middleware import MiddlewareChain
from vk_api.exceptions import ApiError
import requests.exceptions
import re
class VKBot:
def __init__(self, fsm: DialogManager, exporter: ExcelExporter):
self.fsm = fsm
self.exporter = exporter
self.vk_session = vk_api.VkApi(token=settings.VK_TOKEN)
self.longpoll = VkBotLongPoll(self.vk_session, settings.VK_GROUP_ID)
self.vk = self.vk_session.get_api()
self.middlewares = MiddlewareChain()
def _is_direct_message(self, user_id: int, peer_id: int) -> bool:
return user_id == peer_id
def _send_message(self, peer_id: int, text: str, keyboard=None):
"""Отправка сообщения с возможной клавиатурой"""
try:
self.vk.messages.send(
peer_id=peer_id,
message=text,
random_id=0,
keyboard=keyboard.get_keyboard() if keyboard else None
)
except ApiError as e:
if e.code == 901:
logger.warning(f"Can't send to peer {peer_id}: {e}")
# Здесь можно отправить уведомление администратору или сохранить ID
else:
logger.error(f"VK API error {e.code} for peer {peer_id}: {e}")
raise
except Exception as e:
logger.error(f"Failed to send message to {peer_id}: {e}")
raise
def _get_keyboard_for_state(self, state):
"""Генерирует клавиатуру в зависимости от состояния"""
keyboard = VkKeyboard(one_time=False)
if state == "ASK_CONSENT":
keyboard.add_button(phrases.BUTTON_CONSENT_YES, color=VkKeyboardColor.POSITIVE)
keyboard.add_button(phrases.BUTTON_CONSENT_NO, color=VkKeyboardColor.NEGATIVE)
elif state in ("CONFIRM", "confirm"):
keyboard.add_button(phrases.BUTTON_YES, color=VkKeyboardColor.POSITIVE)
keyboard.add_button(phrases.BUTTON_NO, color=VkKeyboardColor.NEGATIVE)
elif state == "START" or state is None:
# Начальное состояние — кнопка "Зарегистрироваться"
keyboard.add_button(phrases.BUTTON_REGISTER, color=VkKeyboardColor.POSITIVE, payload='register_btn')
else:
keyboard.add_button(phrases.BUTTON_RESTART, color=VkKeyboardColor.SECONDARY)
return keyboard
def _handle_command(self, user_id: int, text: str) -> bool:
"""Обработка команд администрирования"""
if user_id not in settings.ADMIN_IDS:
return False
if text == phrases.CMD_STATUS:
self._send_message(user_id, "✅ Бот работает и принимает сообщения.")
return True
elif text == phrases.CMD_EXPORT:
path = self.exporter.export()
self._send_message(user_id, f"Экспорт создан: {path}")
return True
elif text == phrases.CMD_BACKUP:
self.exporter.backup()
self._send_message(user_id, "Резервная копия создана.")
return True
elif text == phrases.CMD_STATS:
stats = self.exporter.get_stats()
msg = (f"📊 Статистика:\nВсего лидов: {stats['total']}\n"
f"За сегодня: {stats['today']}\nСтатусы: {stats['statuses']}")
self._send_message(user_id, msg)
return True
elif text == phrases.CMD_RELOAD:
import importlib
import config.phrases
importlib.reload(config.phrases)
# Обновляем ссылку на phrases в текущем модуле
globals()['phrases'] = config.phrases
self._send_message(user_id, "Конфигурация перезагружена.")
return True
return False
def run(self):
logger.info("Bot started")
schedule_daily_backup(self.exporter)
while True:
try:
for event in self.longpoll.listen():
try:
if event.type == VkBotEventType.MESSAGE_NEW and event.message:
user_id = event.message.from_id
peer_id = event.message.peer_id
text = event.message.text
if not self._is_direct_message(user_id, peer_id):
logger.debug(f"Ignoring chat message from {user_id} in peer {peer_id}")
continue
# Очистка текста от VK-ссылок вида [club123456|текст] и [user123|текст]
text = re.sub(r'\[club\d+\|([^]]*)\]', r'\1', text)
text = re.sub(r'\[user\d+\|([^]]*)\]', r'\1', text)
text = re.sub(r'\[id\d+\|([^]]*)\]', r'\1', text)
logger.debug(f"Message from {user_id} in peer {peer_id}: {text}")
if not self.middlewares.process(user_id, text):
continue
if self._handle_command(user_id, text):
continue
response = self.fsm.handle_message(user_id, text)
logger.info(f"Response to user {user_id}: {response[:100]}")
state = self.fsm.sessions.get(user_id, {}).get("state", "")
keyboard = self._get_keyboard_for_state(state)
self._send_message(peer_id, response, keyboard)
# Обработка нажатия callback-кнопки "Зарегистрироваться"
elif event.type == VkBotEventType.MESSAGE_NEW_CALLBACK and event.message:
user_id = event.message.from_id
peer_id = event.message.peer_id
callback_text = event.message.get_text() if event.message.get_text() else ""
if not self._is_direct_message(user_id, peer_id):
logger.debug(f"Ignoring chat callback from {user_id} in peer {peer_id}")
continue
logger.debug(f"Callback from {user_id} in peer {peer_id}: {callback_text}")
if not self.middlewares.process(user_id, callback_text):
continue
# Проверяем, что нажата кнопка "Зарегистрироваться"
if event.message.get_payload() and 'register_btn' in str(event.message.get_payload()):
# Начинаем диалог заново
response = self.fsm._reset_dialog(user_id)
# Если сессии нет, значит диалог ещё не начат
if user_id not in self.fsm.sessions:
response = self.fsm._start_dialog(user_id)
state = self.fsm.sessions.get(user_id, {}).get("state", "START")
keyboard = self._get_keyboard_for_state(state)
self._send_message(peer_id, response, keyboard)
except ApiError as e:
logger.error(f"VK API error while processing event: {e}")
except Exception as e:
logger.error(f"Error processing event: {e}", exc_info=True)
except requests.exceptions.ConnectionError as e:
logger.warning(f"LongPoll connection lost: {e}. Reconnecting in 5 seconds...")
import time
time.sleep(5)
except Exception as e:
logger.error(f"Fatal error in longpoll loop: {e}", exc_info=True)
logger.warning("Reconnecting in 10 seconds...")
import time
time.sleep(10)
start.py
+27
View File
@@ -0,0 +1,27 @@
"""
Заглушка для проверки токена VK API.
Перед использованием:
1. Скопируйте .env.example в .env
2. Заполните VK_TOKEN и VK_GROUP_ID в .env
3. Запустите: python start.py
"""
import os
from vk_api import VkApi
from dotenv import load_dotenv
load_dotenv()
token = os.getenv("VK_TOKEN")
if not token:
print("❌ Ошибка: VK_TOKEN не найден в .env файле!")
print("Скопируйте .env.example в .env и заполните данные.")
exit(1)
vk_session = VkApi(token=token)
try:
perms = vk_session.get_api().account.getAppPermissions()
print(f"✅ Токен активен. Права: {perms}")
print("4096 = messages, 8192 = groups, 12288 = messages + groups")
except Exception as e:
print(f"❌ Ошибка проверки токена: {e}")
tests/__init__.py
View File
tests/test_exporter.py
View File
tests/test_fsm.py
View File
tests/test_validator.py
+24
View File
@@ -0,0 +1,24 @@
import pytest
from core.validator import validate_fio, validate_phone, parse_time
def test_fio_valid():
res = validate_fio("Иванов Иван Иванович")
assert res.is_valid
assert res.value == "Иванов Иван Иванович"
def test_fio_invalid():
res = validate_fio("Иванов Иван")
assert not res.is_valid
def test_phone_valid():
res = validate_phone("+7 (912) 345-67-89")
assert res.is_valid
assert "912" in res.value
def test_phone_clean():
res = validate_phone("89123456789")
assert res.is_valid
def test_parse_time():
res = parse_time("завтра в 15:00")
assert res.is_valid
utils/__init__.py
View File
utils/backup.py
+29
View File
@@ -0,0 +1,29 @@
import schedule
import threading
import time
from loguru import logger
def schedule_daily_backup(exporter):
"""Запускает ежесуточное резервное копирование в отдельном потоке"""
def job():
try:
exporter.backup()
logger.info("Daily backup executed successfully")
except Exception as e:
logger.error(f"Backup failed: {e}")
schedule.every().day.at("03:00").do(job)
def run_scheduler():
while True:
try:
schedule.run_pending()
except Exception as e:
logger.error(f"Scheduler error: {e}")
time.sleep(30) # Уменьшил интервал с 60 до 30 секунд для более точного срабатывания
thread = threading.Thread(target=run_scheduler, daemon=True)
thread.start()
logger.info("Backup scheduler started")
utils/logger.py
+17
View File
@@ -0,0 +1,17 @@
from loguru import logger
import sys
from config import settings
logger.remove()
logger.add(
sys.stdout,
format="<green>{time:HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan> - {message}",
level=settings.LOG_LEVEL
)
logger.add(
"logs/bot.log",
rotation="1 day",
retention="30 days",
format="{time} | {level} | {message}",
level="DEBUG"
)
utils/middleware.py
+71
View File
@@ -0,0 +1,71 @@
from loguru import logger
from typing import Dict, List
from datetime import datetime, timedelta
class MiddlewareChain:
"""Простая цепочка мидлваров (логирование, антиспам, аналитика)"""
def __init__(self):
self.middlewares: List[callable] = []
def add(self, middleware: callable):
self.middlewares.append(middleware)
def process(self, user_id: int, text: str) -> bool:
for mw in self.middlewares:
if not mw(user_id, text):
return False
return True
# ==================== Антиспам-фильтр ====================
class SpamFilter:
"""Простой антиспам-фильтр: блокирует пользователя, если он отправляет
более `max_messages` сообщений за `window_seconds` секунд."""
def __init__(self, max_messages: int = 5, window_seconds: int = 60):
self.max_messages = max_messages
self.window_seconds = window_seconds
# user_id -> [timestamp1, timestamp2, ...]
self._messages: Dict[int, List[float]] = {}
def is_spam(self, user_id: int) -> bool:
now = datetime.now().timestamp()
window_start = now - self.window_seconds
# Инициализируем список, если нет
if user_id not in self._messages:
self._messages[user_id] = []
# Удаляем старые записи за пределами окна
self._messages[user_id] = [
ts for ts in self._messages[user_id] if ts > window_start
]
# Проверяем, не превышен ли лимит
if len(self._messages[user_id]) >= self.max_messages:
return True
# Записываем текущее сообщение
self._messages[user_id].append(now)
return False
def reset(self, user_id: int):
"""Сброс счётчика для пользователя (например, при /start)"""
self._messages.pop(user_id, None)
# ==================== Логирование ====================
def logging_middleware(user_id: int, text: str) -> bool:
logger.info(f"Processed message from {user_id}: {text}")
return True
# ==================== Пример: мидлвар для аналитики ====================
def analytics_middleware(user_id: int, text: str) -> bool:
"""Заглушка для аналитики — можно расширить позже."""
logger.debug(f"Analytics: user {user_id} sent '{text}'")
return True