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 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 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 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.2.0.tar.gz (30.7 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.2.0-py3-none-any.whl (39.5 kB view details)

Uploaded Python 3

File details

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

File metadata

  • Download URL: ifa_nextcloud-0.2.0.tar.gz
  • Upload date:
  • Size: 30.7 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/6.1.0 CPython/3.13.12

File hashes

Hashes for ifa_nextcloud-0.2.0.tar.gz
Algorithm Hash digest
SHA256 c6d4f72f96e050c888ec37b87bb855653b8f522ba604c57df95a8e3e0fc58a3b
MD5 332a069e5f961520f3f28bd522371867
BLAKE2b-256 05967de6f6d3262b6a947935559a3fa2e3421da07dddd891d0d9148fd68ad082

See more details on using hashes here.

Provenance

The following attestation bundles were made for ifa_nextcloud-0.2.0.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.2.0-py3-none-any.whl.

File metadata

  • Download URL: ifa_nextcloud-0.2.0-py3-none-any.whl
  • Upload date:
  • Size: 39.5 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/6.1.0 CPython/3.13.12

File hashes

Hashes for ifa_nextcloud-0.2.0-py3-none-any.whl
Algorithm Hash digest
SHA256 c82eb57f692788b62e91681cae37d6f91e9c9bc671cbf248fe324fdbfebf590f
MD5 89d65e7d18c8772863daf10976751425
BLAKE2b-256 d8f5f7edb45f40ecc44d2418af111bafc69fea055ec3b6b1a326b94c2e5f74c3

See more details on using hashes here.

Provenance

The following attestation bundles were made for ifa_nextcloud-0.2.0-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