Python библиотека для создания ботов в Nextcloud Talk с API, похожим на python-telegram-bot
Project description
Nextcloud Talk Bot
Python библиотека для создания ботов в Nextcloud Talk с API, похожим на python-telegram-bot.
Возможности
- 🚀 Простой API — интерфейс, похожий на python-telegram-bot
- 📝 Отправка сообщений с поддержкой Markdown
- 📎 Отправка файлов (документы, фото, любые вложения)
- 🔄 Автоматическое управление членством в комнатах
- 💬 Поддержка threading и ответов на сообщения
- 🔌 Автоматическое определение версии API (v1-v4)
- 🛡️ Автоматическая переавторизация при ошибках
- 📡 Polling режим для получения сообщений
Требования
- Python 3.7+
- Nextcloud 33.x и выше (с установленным приложением Talk)
- Аккаунт бота в Nextcloud (рекомендуется использовать токен приложения)
Установка
Из исходного кода
git clone https://github.com/your-username/nextcloud-talk-bot.git
cd nextcloud-talk-bot
pip install -r requirements.txt
Использование как библиотеки
Скопируйте папку core/ в ваш проект или установите как модуль:
# Просто скопируйте файлы в ваш проект
from nextcloudbot import Bot
Быстрый старт
Минимальный пример
from nextcloudbot import Bot
# Создаем бота
bot = Bot(
host="https://nextcloud.example.com",
user="bot_user",
password="your-app-token", # или пароль
default_room="room_token_here"
)
# Отправляем сообщение
bot.send_message("Hello, World!")
# Запускаем получение сообщений
bot.run_polling()
Бот с обработкой команд
from nextcloudbot import Bot
bot = Bot(
host="https://nextcloud.example.com",
user="my_bot",
password="app-xxxxx",
default_room="room_token"
)
# Обработчик команды /start
@bot.command("start")
def start_command(update, context):
update.message.reply_text("👋 Привет! Я бот для Nextcloud Talk!")
# Обработчик команды /help
@bot.command("help")
def help_command(update, context):
help_text = """
🤖 **Доступные команды:**
/start - Приветствие
/help - Эта справка
/echo <текст> - Повторить сообщение
"""
update.message.reply_text(help_text)
# Обработчик команды /echo
@bot.command("echo")
def echo_command(update, context):
# Получаем текст после команды
text = update.message.text.replace("/echo", "").strip()
if text:
update.message.reply_text(f"🔊 {text}")
else:
update.message.reply_text("❌ Напишите что-нибудь после /echo")
# Обработчик всех текстовых сообщений
@bot.message_handler
def handle_message(update, context):
message = update.message
user = message.from_user
print(f"Получено сообщение от {user.full_name}: {message.text}")
message.reply_text(f"Получил: {message.text}")
# Запускаем бота
if __name__ == "__main__":
bot.run_polling()
Отправка сообщений
Текстовые сообщения
# Простой текст
bot.send_message("Hello, World!")
# Markdown форматирование
bot.send_message("**Жирный текст** и *курсив*")
# Ответ на сообщение
bot.send_message(
chat_id="room_token", # optional
text="Это ответ на ваше сообщение",
reply_to_message_id=12345
)
Отправка файлов
# Отправить файл с подписью
bot.send_message(
chat_id="room_token", # optional
text="Смотрите документ",
file_path="/path/to/document.pdf"
)
# Отправить фото
bot.send_file("/path/to/photo.jpg", caption="Красивый закат")
# Или через метод send_photo
bot.send_photo(
"room_token", # optional
"/path/to/photo.jpg",
caption="Красивый закат"
)
# Отправить документ
bot.send_document(
"room_token", # optional
"/path/to/report.pdf",
caption="Ежемесячный отчет"
)
# Отправить файл из памяти (например, сгенерированный)
import io
# Создаем файл в памяти
file_content = b"Hello, this is a text file!"
bot.send_message(
chat_id="room_token", # optional
text="Сгенерированный файл",
file_content=file_content,
file_name="hello.txt",
mime_type="text/plain"
)
# Отправить файл из BytesIO
from io import BytesIO
buffer = BytesIO()
buffer.write(b"Some data")
buffer.seek(0)
bot.send_document(
"room_token", # optional
buffer,
caption="Данные из буфера"
)
Работа с сообщениями
Получение сообщений через polling
# Запускаем получение сообщений
bot.run_polling(chat_id="room_token", poll_interval=2)
# Параметры:
# - chat_id: токен комнаты (если не указан, используется default_room)
# - poll_interval: интервал опроса в секундах (по умолчанию 2)
Объект Message
При получении сообщения вы получаете объект Message со следующими атрибутами:
@bot.message_handler
def handle_message(update, context):
message = update.message
# Основные атрибуты
print(f"ID: {message.message_id}")
print(f"Текст: {message.text}")
print(f"От: {message.from_user.full_name}")
print(f"Время: {message.date}")
# Ответить на сообщение
message.reply_text("Получил ваше сообщение!")
# Или использовать метод reply (алиас)
message.reply("Тоже самое")
Управление комнатами
Присоединение к комнате
# Автоматическое присоединение при запуске
bot = Bot(
host="https://nextcloud.example.com",
user="bot_user",
password="password",
default_room="room_token",
auto_join_room=True # По умолчанию True
)
# Ручное присоединение
bot.join_room("room_token", password="room_password")
Получение информации о комнатах
# Список всех доступных комнат
rooms = bot.get_rooms()
for room in rooms:
print(f"Комната: {room.get('name')} ({room.get('token')})")
print(f"Участников: {room.get('participantCount')}")
# Информация о конкретной комнате
room_info = bot.get_room_info("room_token")
print(f"Название: {room_info.get('name')}")
print(f"Тип: {room_info.get('type')}")
# Список участников
participants = bot.get_room_participants("room_token")
for p in participants:
print(f"Участник: {p.get('actorDisplayName')}")
Диагностика
Проверка подключения
# Проверка статуса сессии
status = bot.check_session_status()
if status['authenticated']:
print(f"✅ Подключен как: {status['user']}")
else:
print("❌ Ошибка аутентификации")
# Диагностика доступа к комнате
diagnostic = bot.diagnose_room_access("room_token")
print(json.dumps(diagnostic, indent=2, ensure_ascii=False))
Логирование
Библиотека использует loguru для логирования. Вы можете настроить уровень логирования:
from loguru import logger
# Установка уровня логирования
logger.remove() # Удаляем стандартный обработчик
logger.add(sys.stderr, level="INFO") # Только INFO и выше
# или
logger.add(sys.stderr, level="DEBUG") # Подробное логирование
Конфигурация
Параметры Bot
| Параметр | Тип | По умолчанию | Описание |
|---|---|---|---|
host |
str | - | URL Nextcloud сервера |
user |
str | - | Имя пользователя бота |
password |
str | - | Пароль или токен приложения |
default_room |
str | None | Токен комнаты по умолчанию |
read_all_chat |
bool | False | Читать всю историю (c новых к старым) или только новые сообщения |
auto_join_room |
bool | True | Автоматически присоединяться к комнатам |
Примеры
Простой эхо-бот
from core.nextcloud import Bot
bot = Bot(
host="https://nextcloud.example.com",
user="echo_bot",
password="app-xxxxx",
default_room="room_token"
)
@bot.message_handler
def echo(update, context):
message = update.message
if message.text:
message.reply_text(f"🔊 Эхо: {message.text}")
else:
message.reply_text("Напишите текст, я его повторю!")
bot.run_polling()
Бот для отправки уведомлений
import time
from core.nextcloud import Bot
class NotificationBot:
def __init__(self, host, user, password, room_token):
self.bot = Bot(
host=host,
user=user,
password=password,
default_room=room_token
)
self.room = room_token
def send_notification(self, title, message, priority="info"):
"""Отправить уведомление в чат"""
emoji = {
"info": "ℹ️",
"success": "✅",
"warning": "⚠️",
"error": "❌"
}.get(priority, "📢")
text = f"{emoji} **{title}**\n{message}"
self.bot.send_message(self.room, text)
def send_file_notification(self, title, file_path, message=None):
"""Отправить уведомление с файлом"""
full_text = f"📎 **{title}**"
if message:
full_text += f"\n{message}"
self.bot.send_message(
chat_id=self.room,
text=full_text,
file_path=file_path
)
# Использование
notifier = NotificationBot(
host="https://nextcloud.example.com",
user="notifier",
password="app-xxxxx",
room_token="room_token"
)
notifier.send_notification("Система", "Резервное копирование завершено", "success")
notifier.send_file_notification("Отчет", "/tmp/report.pdf", "Ежемесячный отчет готов")
Бот с обработкой команд
from core.nextcloud import Bot
from datetime import datetime
bot = Bot(
host="https://nextcloud.example.com",
user="helper_bot",
password="app-xxxxx",
default_room="room_token"
)
# Хранилище данных пользователей
user_data = {}
@bot.command("start")
def cmd_start(update, context):
user = update.message.from_user
update.message.reply_text(
f"👋 Привет, {user.first_name}!\n"
"Я бот-помощник. Доступные команды:\n"
"/time - текущее время\n"
"/ping - проверить работу\n"
"/note <текст> - сохранить заметку\n"
"/mynote - показать заметку"
)
@bot.command("time")
def cmd_time(update, context):
now = datetime.now().strftime("%d.%m.%Y %H:%M:%S")
update.message.reply_text(f"🕐 Текущее время: {now}")
@bot.command("ping")
def cmd_ping(update, context):
update.message.reply_text("🏓 Pong!")
@bot.command("note")
def cmd_note(update, context):
user_id = update.message.from_user.id
text = update.message.text.replace("/note", "").strip()
if text:
user_data[user_id] = text
update.message.reply_text("✅ Заметка сохранена!")
else:
update.message.reply_text("❌ Напишите текст заметки после /note")
@bot.command("mynote")
def cmd_mynote(update, context):
user_id = update.message.from_user.id
note = user_data.get(user_id)
if note:
update.message.reply_text(f"📝 Ваша заметка:\n{note}")
else:
update.message.reply_text("📭 У вас нет сохраненных заметок")
@bot.message_handler
def handle_unknown(update, context):
text = update.message.text
if text:
update.message.reply_text(
f"Неизвестная команда. Напишите /help для списка команд."
)
if __name__ == "__main__":
bot.run_polling()
API Reference
Bot
Методы отправки
| Метод | Описание |
|---|---|
send_message(chat_id, text, ...) |
Универсальный метод отправки (текст/файлы) |
send_file(chat_id, file_path, caption, ...) |
Отправить файл |
send_photo(chat_id, photo, caption, ...) |
Отправить фото |
send_document(chat_id, document, caption, ...) |
Отправить документ |
Управление комнатами
| Метод | Описание |
|---|---|
join_room(chat_id, password) |
Присоединиться к комнате |
get_rooms() |
Получить список всех комнат |
get_room_info(chat_id) |
Получить информацию о комнате |
get_room_participants(chat_id) |
Получить список участников |
Диагностика
| Метод | Описание |
|---|---|
check_session_status() |
Проверить статус сессии |
diagnose_room_access(chat_id) |
Диагностика доступа к комнате |
Message
| Атрибут/Метод | Описание |
|---|---|
message_id |
ID сообщения |
text |
Текст сообщения |
from_user |
Объект User (отправитель) |
chat |
Объект Chat (комната) |
date |
Время отправки |
reply_to_message |
Ответ на сообщение (если есть) |
reply_text(text) |
Ответить на сообщение |
reply(text) |
Алиас для reply_text |
Устранение неполадок
Ошибка аутентификации
# Проверьте подключение
status = bot.check_session_status()
if not status['authenticated']:
print(f"Ошибка: {status.get('error')}")
print("Проверьте логин и пароль/токен")
Комната не найдена
# Получите список доступных комнат
rooms = bot.get_rooms()
print("Доступные комнаты:")
for room in rooms:
print(f" - {room.get('name')} (токен: {room.get('token')})")
Проблемы с отправкой файлов
# Используйте диагностику
result = bot.diagnose_room_access("room_token")
print(json.dumps(result, indent=2))
# Проверьте права на запись в WebDAV
# Убедитесь, что у бота есть права на загрузку файлов
Лицензия
MIT License
Вклад в проект
Буду рад вашим pull requests и issue!
Ссылки
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file ifa_nextcloud-0.1.2.tar.gz.
File metadata
- Download URL: ifa_nextcloud-0.1.2.tar.gz
- Upload date:
- Size: 18.2 kB
- Tags: Source
- Uploaded using Trusted Publishing? Yes
- Uploaded via: twine/6.1.0 CPython/3.13.7
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
7d114b6e23c4b9d859aaf0c16eadc68db859fa61d603eb236fb761ef516686fc
|
|
| MD5 |
63daad809756cd2240625d833abe8a54
|
|
| BLAKE2b-256 |
a6f59c0fba9a3f5ca164039a5feb4383177d161537d5262e683ecf38778ea2f7
|
Provenance
The following attestation bundles were made for ifa_nextcloud-0.1.2.tar.gz:
Publisher:
publish-to-pypi.yml on 1FaKe110/nextcloud
-
Statement:
-
Statement type:
https://in-toto.io/Statement/v1 -
Predicate type:
https://docs.pypi.org/attestations/publish/v1 -
Subject name:
ifa_nextcloud-0.1.2.tar.gz -
Subject digest:
7d114b6e23c4b9d859aaf0c16eadc68db859fa61d603eb236fb761ef516686fc - Sigstore transparency entry: 1215846560
- Sigstore integration time:
-
Permalink:
1FaKe110/nextcloud@21e3a3bb23513b088825e7e81d1830718a99c906 -
Branch / Tag:
refs/tags/0.1.3 - Owner: https://github.com/1FaKe110
-
Access:
public
-
Token Issuer:
https://token.actions.githubusercontent.com -
Runner Environment:
github-hosted -
Publication workflow:
publish-to-pypi.yml@21e3a3bb23513b088825e7e81d1830718a99c906 -
Trigger Event:
push
-
Statement type:
File details
Details for the file ifa_nextcloud-0.1.2-py3-none-any.whl.
File metadata
- Download URL: ifa_nextcloud-0.1.2-py3-none-any.whl
- Upload date:
- Size: 19.9 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? Yes
- Uploaded via: twine/6.1.0 CPython/3.13.7
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
be972b0af17119d42f83c68b117604efe02cd5124b0ce816b34ac8da95728e4f
|
|
| MD5 |
bf269a2ccc81a561f78da0db6aa05b44
|
|
| BLAKE2b-256 |
d9f73bb289cb8f95742ccd8eadbb2dd2e7407be4879b29b7ba08c52168fdd753
|
Provenance
The following attestation bundles were made for ifa_nextcloud-0.1.2-py3-none-any.whl:
Publisher:
publish-to-pypi.yml on 1FaKe110/nextcloud
-
Statement:
-
Statement type:
https://in-toto.io/Statement/v1 -
Predicate type:
https://docs.pypi.org/attestations/publish/v1 -
Subject name:
ifa_nextcloud-0.1.2-py3-none-any.whl -
Subject digest:
be972b0af17119d42f83c68b117604efe02cd5124b0ce816b34ac8da95728e4f - Sigstore transparency entry: 1215846821
- Sigstore integration time:
-
Permalink:
1FaKe110/nextcloud@21e3a3bb23513b088825e7e81d1830718a99c906 -
Branch / Tag:
refs/tags/0.1.3 - Owner: https://github.com/1FaKe110
-
Access:
public
-
Token Issuer:
https://token.actions.githubusercontent.com -
Runner Environment:
github-hosted -
Publication workflow:
publish-to-pypi.yml@21e3a3bb23513b088825e7e81d1830718a99c906 -
Trigger Event:
push
-
Statement type: