King-of-the-all-Cookies/mikopbxbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

3

Browse Source
This commit is contained in:
King-of-the-all-Cookies
2026-06-24 17:09:24 +03:00
parent 83e3f8bffe
commit 90a4d02b79
5 changed files with 128 additions and 257 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.env.example
View File

@@ -1,7 +1,12 @@
# MikoPBX configuration
MIKOPBX_HOST=http://your-mikopbx-ip-or-host:8081
# === РЕКОМЕНДУЕТСЯ: Bearer Token (самый стабильный способ) ===
MIKOPBX_API_TOKEN=твой_токен_сюда
# Fallback (если токен не указан)
MIKOPBX_ADMIN_LOGIN=admin
MIKOPBX_ADMIN_PASSWORD=your-admin-password
MIKOPBX_ADMIN_PASSWORD=твой_пароль
# Telegram Bot
TELEGRAM_BOT_TOKEN=your-telegram-bot-token-here

149
README.md
View File

@@ -1,25 +1,23 @@
# MikoPBX Public SIP Telephony Bot
Telegram-бот для публичной SIP-телефонии на базе MikoPBX.
Telegram-бот для публичной SIP-телефонии на MikoPBX с полной синхронизацией.
## Основные возможности
## Авторизация в MikoPBX
### Для обычных пользователей
- Регистрация **строго 7-значного** номера
- **Один номер на одного пользователя** (ограничение)
- Выбор пароля: сгенерировать или ввести свой
- Получение **полных параметров подключения** (домен, сервер, порт, транспорт, outbound proxy, STUN и т.д.)
- Просмотр своих аккаунтов
Бот поддерживает **два способа** авторизации (приоритет — Bearer Token):
### Для администраторов
- Полный список всех аккаунтов
- Смена номера / пароля
- Удаление аккаунтов
- **Привязка уже существующих номеров из MikoPBX** к Telegram ID пользователя
1. **Bearer Token** (рекомендуется) — самый стабильный
2. **Basic Auth** (fallback)
### Синхронизация с MikoPBX
- Полная двусторонняя синхронизация (создание, изменение номера, изменение пароля, удаление)
- Возможность привязывать уже существующие в MikoPBX аккаунты
---
## Как создать Bearer Token в MikoPBX
1. Зайди в веб-интерфейс MikoPBX
2. Перейди в **Система → Права сотрудников**
3. Создай пользователя или открой существующего
4. В разделе **REST API** / **API Keys** создай токен
5. Скопируй токен и вставь в `.env`
---
@@ -28,125 +26,46 @@ Telegram-бот для публичной SIP-телефонии на базе M
```env
# MikoPBX
MIKOPBX_HOST=http://192.168.1.50:8081
MIKOPBX_ADMIN_LOGIN=admin
MIKOPBX_ADMIN_PASSWORD=твой_пароль
# === Bearer Token (рекомендуется) ===
MIKOPBX_API_TOKEN=1bbb7b1357e8bdb0a933f054aedf3aa1f947f7237f3895a05fa29198a253f935
# Telegram
TELEGRAM_BOT_TOKEN=твой_токен
ADMIN_TELEGRAM_IDS=123456789,987654321
# SIP параметры (показываются пользователям)
# SIP параметры
DEFAULT_SIP_SERVER=sip.tvoyserver.ru
DEFAULT_SIP_PORT=5060
DEFAULT_SIP_DOMAIN=sip.tvoyserver.ru
DEFAULT_SIP_TRANSPORT=UDP
DEFAULT_SIP_OUTBOUND_PROXY=
DEFAULT_SIP_STUN=stun.l.google.com:19302
```
**Все параметры подключения настраиваются в `.env`** и автоматически подставляются в сообщение пользователю.
---
## Как работает регистрация номера
1. Пользователь нажимает **"Register new number"**
2. Вводит **ровно 7 цифр**
3. Вводит имя (опционально)
4. Выбирает способ получения пароля:
- Сгенерировать случайный
- Ввести свой пароль
5. Бот создаёт аккаунт в **MikoPBX** с указанным паролем
6. Выдаёт **полные параметры подключения**
**Ограничения:**
- Номер строго 7 цифр
- У одного Telegram-пользователя может быть только **один** номер
---
## Привязка существующих номеров MikoPBX (Админ)
Это одна из ключевых фич:
1. Админ → **Admin panel** → **Link existing MikoPBX number**
2. Вводит существующий 7-значный номер из MikoPBX
3. Вводит Telegram ID пользователя
4. Номер привязывается к пользователю
После этого пользователь может:
- Видеть свой номер в разделе "My accounts"
- Получать параметры подключения (если пароль был заранее известен)
**Полезно**, когда номера уже созданы в MikoPBX вручную.
---
## Полный список кнопок и команд
### Главное меню (для всех)
- `Register new number` — регистрация нового номера
- `My accounts` — мои аккаунты
- `Admin panel` — только для администраторов
### Админ-панель
- `All registered accounts` — список всех аккаунтов
- `Link existing MikoPBX number` — привязка существующего номера
- При просмотре аккаунта:
- `Change number`
- `Change password`
- `Delete account`
### Команды
- `/start` — главное меню
- `/cancel` — отмена текущего действия
---
## Пример сообщения с параметрами подключения
```
Your SIP account has been created!
Number: 1001001
Secret (password): MySuperPass2025
--- Full connection parameters ---
Domain: sip.tvoyserver.ru
Server / Outbound Proxy: sip.tvoyserver.ru
Port: 5060
Transport: UDP
Username / Auth ID: 1001001
Password: MySuperPass2025
Outbound Proxy:
STUN Server: stun.l.google.com:19302
Recommended clients: Zoiper, MicroSIP, Linphone, Bria
Use these credentials in your SIP client.
DEFAULT_SIP_STUN=
```
---
## Запуск и обновление
## Запуск
```bash
# Первый запуск
docker compose up -d --build
# Просмотр логов
docker compose logs -f mikopbx-bot
# Пересборка после изменений
docker compose up -d --build
```
---
## Важные замечания
## Основные возможности
- Номера строго 7 цифр (валидация на стороне бота)
- Один Telegram ID = один номер
- Все изменения (номер/пароль/удаление) выполняются и в MikoPBX, и в локальной БД
- Привязка существующих номеров позволяет работать с уже созданными в MikoPBX аккаунтами без пересоздания
- Регистрация **ровно 7-значных** номеров
- Один номер на пользователя
- Выбор своего пароля при регистрации
- Полные параметры подключения (домен, outbound proxy, STUN и т.д.)
- **Привязка уже существующих номеров MikoPBX**
- Полная синхронизация (создание, смена номера/пароля, удаление)
Бот полностью готов к эксплуатации.
---
## Логи
```bash
docker compose logs -f mikopbx-bot
```

1
bot/config.py
View File

@@ -4,6 +4,7 @@ from dotenv import load_dotenv
load_dotenv()
MIKOPBX_HOST = os.getenv("MIKOPBX_HOST", "http://localhost:8081")
MIKOPBX_API_TOKEN = os.getenv("MIKOPBX_API_TOKEN") # Bearer Token (приоритет)
MIKOPBX_ADMIN_LOGIN = os.getenv("MIKOPBX_ADMIN_LOGIN", "admin")
MIKOPBX_ADMIN_PASSWORD = os.getenv("MIKOPBX_ADMIN_PASSWORD", "admin")

214
bot/services/mikopbx_service.py
View File

@@ -1,75 +1,72 @@
import aiohttp
import logging
from typing import Optional, Dict, Any
from config import MIKOPBX_HOST, MIKOPBX_ADMIN_LOGIN, MIKOPBX_ADMIN_PASSWORD
from config import (
MIKOPBX_HOST,
MIKOPBX_API_TOKEN,
MIKOPBX_ADMIN_LOGIN,
MIKOPBX_ADMIN_PASSWORD
)
logger = logging.getLogger(__name__)
class MikoPBXService:
def __init__(self):
self.host = MIKOPBX_HOST.rstrip('/')
self.api_token = MIKOPBX_API_TOKEN
self.admin_login = MIKOPBX_ADMIN_LOGIN
self.admin_password = MIKOPBX_ADMIN_PASSWORD
self.session = None
self.auth_cookies = None
async def _get_session(self):
if self.session is None:
self.session = aiohttp.ClientSession()
return self.session
async def login(self) -> bool:
"""Login to MikoPBX admin panel to get session cookie"""
session = await self._get_session()
try:
url = f"{self.host}/admin-cabinet/session/start"
data = {
"login": self.admin_login,
"password": self.admin_password
def _get_auth_headers(self) -> Dict[str, str]:
"""Возвращает заголовки авторизации (приоритет — Bearer Token)"""
headers = {
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
"X-Requested-With": "XMLHttpRequest"
}
async with session.post(url, data=data) as resp:
if resp.status == 200:
self.auth_cookies = resp.cookies
logger.info("Successfully logged into MikoPBX")
return True
else:
logger.error(f"Login failed: {resp.status}")
return False
except Exception as e:
logger.error(f"Login error: {e}")
return False
if self.api_token:
headers["Authorization"] = f"Bearer {self.api_token}"
logger.debug("Using Bearer Token authentication")
elif self.admin_login and self.admin_password:
# Fallback на Basic Auth
import base64
credentials = f"{self.admin_login}:{self.admin_password}"
encoded = base64.b64encode(credentials.encode()).decode()
headers["Authorization"] = f"Basic {encoded}"
logger.debug("Using Basic Auth (fallback)")
return headers
async def _make_request(self, method: str, endpoint: str, data: Optional[Dict] = None, params: Optional[Dict] = None):
"""Make authenticated request to MikoPBX API"""
"""Выполняет запрос к MikoPBX API"""
session = await self._get_session()
if not self.auth_cookies:
await self.login()
url = f"{self.host}{endpoint}"
headers = {
"X-Requested-With": "XMLHttpRequest",
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"
}
cookies = {}
if self.auth_cookies:
for key, cookie in self.auth_cookies.items():
cookies[key] = cookie.value
headers = self._get_auth_headers()
try:
if method.upper() == "GET":
async with session.get(url, params=params, cookies=cookies, headers=headers) as resp:
async with session.get(url, params=params, headers=headers) as resp:
if resp.status == 401:
logger.error("Unauthorized (401). Check your API token.")
return await resp.json()
else:
async with session.post(url, data=data, cookies=cookies, headers=headers) as resp:
async with session.post(url, data=data, headers=headers) as resp:
if resp.status == 401:
logger.error("Unauthorized (401). Check your API token.")
return await resp.json()
except Exception as e:
logger.error(f"API request error: {e}")
return {"result": False, "messages": [str(e)]}
async def get_extension_template(self) -> Optional[Dict]:
"""Get empty template for new extension"""
"""Получить шаблон для создания нового сотрудника"""
response = await self._make_request(
"GET",
"/pbxcore/api/extensions/getRecord",
@@ -77,27 +74,25 @@ class MikoPBXService:
)
if response.get("result"):
return response.get("data")
logger.error(f"Failed to get template: {response.get('messages')}")
return None
async def create_extension(self, number: str, secret: str = None, username: str = "", email: str = "") -> Dict[str, Any]:
"""Create new SIP extension with optional custom secret"""
"""Создать нового SIP пользователя"""
template = await self.get_extension_template()
if not template:
return {"success": False, "error": "Failed to get template"}
return {"success": False, "error": "Failed to get template from MikoPBX"}
# Fill template with our data
template["number"] = number
if username:
template["user_username"] = username
if email:
template["user_email"] = email
# Ensure SIP type and basic settings
template["type"] = "SIP"
template["show_in_phonebook"] = "1"
template["is_general_user_number"] = "1"
# Set custom secret if provided
if secret:
template["sip_secret"] = secret
@@ -121,82 +116,8 @@ class MikoPBXService:
"error": response.get("messages", ["Unknown error"])
}
async def update_extension(self, extension_id: str, data: Dict) -> Dict[str, Any]:
"""Update existing extension"""
# First get current record
current = await self._make_request(
"GET",
"/pbxcore/api/extensions/getRecord",
params={"id": extension_id}
)
if not current.get("result"):
return {"success": False, "error": "Extension not found"}
record = current.get("data", {})
record.update(data)
response = await self._make_request(
"POST",
"/pbxcore/api/extensions/saveRecord",
data=record
)
return {
"success": response.get("result", False),
"messages": response.get("messages", [])
}
async def change_extension_number(self, old_number: str, new_number: str) -> Dict[str, Any]:
"""Change extension number (delete old + create new with same secret)"""
# Get current record by number (we'll search via getList if available, else use template approach)
# For simplicity we get template and update the number
template = await self.get_extension_template()
if not template:
return {"success": False, "error": "Failed to get template"}
# We need the existing record. Since there's no direct "get by number" we use a workaround:
# Try to find by attempting getRecord with number as ID (MikoPBX sometimes accepts it)
current = await self._make_request(
"GET",
"/pbxcore/api/extensions/getRecord",
params={"id": old_number}
)
if current.get("result"):
record = current.get("data", {})
old_secret = record.get("sip_secret", "")
# Delete old extension first
delete_response = await self._make_request(
"POST",
"/pbxcore/api/extensions/deleteRecord",
data={"id": record.get("id", old_number)}
)
# Create new one with same settings but new number
record["number"] = new_number
record["id"] = "" # new record
record.pop("sip_uniqid", None)
create_response = await self._make_request(
"POST",
"/pbxcore/api/extensions/saveRecord",
data=record
)
if create_response.get("result"):
return {
"success": True,
"old_number": old_number,
"new_number": new_number,
"secret": old_secret
}
return {"success": False, "error": "Could not change number"}
async def set_extension_secret(self, number: str, new_secret: str) -> Dict[str, Any]:
"""Update secret for existing extension"""
"""Обновить пароль у существующего номера"""
current = await self._make_request(
"GET",
"/pbxcore/api/extensions/getRecord",
@@ -220,20 +141,53 @@ class MikoPBXService:
"messages": response.get("messages", [])
}
async def get_all_extensions(self) -> list:
"""Get list of all extensions (if supported by MikoPBX)"""
# Note: MikoPBX may have /pbxcore/api/extensions/getList endpoint
response = await self._make_request(
async def change_extension_number(self, old_number: str, new_number: str) -> Dict[str, Any]:
"""Сменить номер (удалить старый + создать новый)"""
current = await self._make_request(
"GET",
"/pbxcore/api/extensions/getList"
"/pbxcore/api/extensions/getRecord",
params={"id": old_number}
)
if response.get("result"):
return response.get("data", [])
return []
if not current.get("result"):
return {"success": False, "error": "Extension not found"}
record = current.get("data", {})
old_secret = record.get("sip_secret", "")
# Удаляем старый
delete_resp = await self._make_request(
"POST",
"/pbxcore/api/extensions/deleteRecord",
data={"id": record.get("id", old_number)}
)
if not delete_resp.get("result"):
return {"success": False, "error": "Failed to delete old extension"}
# Создаём новый
record["number"] = new_number
record["id"] = ""
record.pop("sip_uniqid", None)
create_resp = await self._make_request(
"POST",
"/pbxcore/api/extensions/saveRecord",
data=record
)
if create_resp.get("result"):
return {
"success": True,
"old_number": old_number,
"new_number": new_number,
"secret": old_secret
}
return {"success": False, "error": "Failed to create new extension"}
async def delete_extension(self, number: str) -> Dict[str, Any]:
"""Delete extension"""
"""Удалить номер"""
current = await self._make_request(
"GET",
"/pbxcore/api/extensions/getRecord",
@@ -257,12 +211,6 @@ class MikoPBXService:
"messages": response.get("messages", [])
}
async def get_extension_by_number(self, number: str) -> Optional[Dict]:
"""Try to find extension by number"""
# This is a simplified approach - in real scenario we would use getList
# For now return None, or implement getList
return None
async def close(self):
if self.session:
await self.session.close()

2
docker-compose.yml
View File

@@ -1,5 +1,3 @@
version: '3.8'
services:
mikopbx-bot:
build: ./bot