Skip to main content

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


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

ifa_nextcloud-0.1.4.tar.gz (18.2 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

ifa_nextcloud-0.1.4-py3-none-any.whl (19.9 kB view details)

Uploaded Python 3

File details

Details for the file ifa_nextcloud-0.1.4.tar.gz.

File metadata

  • Download URL: ifa_nextcloud-0.1.4.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

Hashes for ifa_nextcloud-0.1.4.tar.gz
Algorithm Hash digest
SHA256 467a0011b296c689d468dd6a2d5947a8239be2687d5bb508b6a70df56bbfcc29
MD5 19dd1f6a5b42f18aa40299a4f413acbb
BLAKE2b-256 89ed0adb7c77905296dd26f4a8fe69747cba2421e8adf840499f0ebc314455ed

See more details on using hashes here.

Provenance

The following attestation bundles were made for ifa_nextcloud-0.1.4.tar.gz:

Publisher: publish-to-pypi.yml on 1FaKe110/nextcloud

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

File details

Details for the file ifa_nextcloud-0.1.4-py3-none-any.whl.

File metadata

  • Download URL: ifa_nextcloud-0.1.4-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

Hashes for ifa_nextcloud-0.1.4-py3-none-any.whl
Algorithm Hash digest
SHA256 70a88f1aebdfb4f41d13feaa50e5ac49b124f9b31393b80153171b3f77a41c51
MD5 21c6f849b87c10a7d0b6ab0f94d526ca
BLAKE2b-256 3abc7cad403e4c4242b13342ecf915791d4c84d7fb8157e050db20175688cfc2

See more details on using hashes here.

Provenance

The following attestation bundles were made for ifa_nextcloud-0.1.4-py3-none-any.whl:

Publisher: publish-to-pypi.yml on 1FaKe110/nextcloud

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page