Psion-fanatic/platypush
1
0
Fork 0
forked from platypush/platypush
Code Pull requests Activity

Added NextCloud integration [closes #149]

Browse source
This commit is contained in:
Fabio Manganiello 2020-09-10 11:10:26 +02:00
parent 0af326fa11
commit c0f7cc0782
12 changed files with 1049 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
docs/source/backends.rst
View file

@ -41,6 +41,7 @@ Backends
platypush/backend/music.mopidy.rst
platypush/backend/music.mpd.rst
platypush/backend/music.snapcast.rst
platypush/backend/nextcloud.rst
platypush/backend/nfc.rst
platypush/backend/nodered.rst
platypush/backend/ping.rst

3
docs/source/events.rst
View file

@ -28,6 +28,7 @@ Events
platypush/events/http.hook.rst
platypush/events/http.ota.booking.rst
platypush/events/http.rss.rst
platypush/events/inotify.rst
platypush/events/joystick.rst
platypush/events/kafka.rst
platypush/events/light.rst
@ -38,8 +39,8 @@ Events
platypush/events/mqtt.rst
platypush/events/music.rst
platypush/events/music.snapcast.rst
platypush/events/nextcloud.rst
platypush/events/nfc.rst
platypush/events/path.rst
platypush/events/ping.rst
platypush/events/pushbullet.rst
platypush/events/qrcode.rst

5
docs/source/platypush/backend/nextcloud.rst Normal file
View file

@ -0,0 +1,5 @@
``platypush.backend.nextcloud``
===============================
.. automodule:: platypush.backend.nextcloud
:members:

5
docs/source/platypush/events/inotify.rst Normal file
View file

@ -0,0 +1,5 @@
``platypush.message.event.inotify``
===================================
.. automodule:: platypush.message.event.inotify
:members:

5
docs/source/platypush/events/nextcloud.rst Normal file
View file

@ -0,0 +1,5 @@
``platypush.message.event.nextcloud``
=====================================
.. automodule:: platypush.message.event.nextcloud
:members:

5
docs/source/platypush/plugins/nextcloud.rst Normal file
View file

@ -0,0 +1,5 @@
``platypush.plugins.nextcloud``
===============================
.. automodule:: platypush.plugins.nextcloud
:members:

1
docs/source/plugins.rst
View file

@ -93,6 +93,7 @@ Plugins
platypush/plugins/music.rst
platypush/plugins/music.mpd.rst
platypush/plugins/music.snapcast.rst
platypush/plugins/nextcloud.rst
platypush/plugins/nmap.rst
platypush/plugins/otp.rst
platypush/plugins/pihole.rst

156
platypush/backend/nextcloud.py Normal file
View file

@ -0,0 +1,156 @@
from typing import Optional
from platypush.backend import Backend
from platypush.context import get_plugin
from platypush.message.event.nextcloud import NextCloudActivityEvent
from platypush.plugins.nextcloud import NextcloudPlugin
from platypush.plugins.variable import VariablePlugin
class NextcloudBackend(Backend):
"""
This backend triggers events when new activities occur on a NextCloud instance.
Triggers:
- :class:`platypush.message.event.nextcloud.NextCloudActivityEvent` when new activity occurs on the instance.
The field ``activity_type`` identifies the activity type (e.g. ``file_created``, ``file_deleted``,
``file_changed``). Example in the case of the creation of new files:
.. code-block:: json
{
"activity_id": 387,
"app": "files",
"activity_type": "file_created",
"user": "your-user",
"subject": "You created InstantUpload/Camera/IMG_0100.jpg, InstantUpload/Camera/IMG_0101.jpg and InstantUpload/Camera/IMG_0102.jpg",
"subject_rich": [
"You created {file3}, {file2} and {file1}",
{
"file1": {
"type": "file",
"id": "41994",
"name": "IMG_0100.jpg",
"path": "InstantUpload/Camera/IMG_0100.jpg",
"link": "https://your-domain/nextcloud/index.php/f/41994"
},
"file2": {
"type": "file",
"id": "42005",
"name": "IMG_0101.jpg",
"path": "InstantUpload/Camera/IMG_0102.jpg",
"link": "https://your-domain/nextcloud/index.php/f/42005"
},
"file3": {
"type": "file",
"id": "42014",
"name": "IMG_0102.jpg",
"path": "InstantUpload/Camera/IMG_0102.jpg",
"link": "https://your-domain/nextcloud/index.php/f/42014"
}
}
],
"message": "",
"message_rich": [
"",
[]
],
"object_type": "files",
"object_id": 41994,
"object_name": "/InstantUpload/Camera/IMG_0102.jpg",
"objects": {
"42014": "/InstantUpload/Camera/IMG_0100.jpg",
"42005": "/InstantUpload/Camera/IMG_0101.jpg",
"41994": "/InstantUpload/Camera/IMG_0102.jpg"
},
"link": "https://your-domain/nextcloud/index.php/apps/files/?dir=/InstantUpload/Camera",
"icon": "https://your-domain/nextcloud/apps/files/img/add-color.svg",
"datetime": "2020-09-07T17:04:29+00:00"
}
"""
_LAST_ACTIVITY_VARNAME = '_NEXTCLOUD_LAST_ACTIVITY_ID'
def __init__(self, url: Optional[str] = None, username: Optional[str] = None, password: Optional[str] = None,
object_type: Optional[str] = None, object_id: Optional[int] = None,
poll_seconds: Optional[float] = 60., **kwargs):
"""
:param url: NextCloud instance URL (default: same as the :class:`platypush.plugins.nextcloud.NextCloudPlugin`).
:param username: NextCloud username (default: same as the :class:`platypush.plugins.nextcloud.NextCloudPlugin`).
:param password: NextCloud password (default: same as the :class:`platypush.plugins.nextcloud.NextCloudPlugin`).
:param object_type: If set, only filter events on this type of object.
:param object_id: If set, only filter events on this object ID.
:param poll_seconds: How often the backend should poll the instance (default: one minute).
"""
super().__init__(**kwargs)
self.url: Optional[str] = None
self.username: Optional[str] = None
self.password: Optional[str] = None
self.object_type = object_type
self.object_id = object_id
self.poll_seconds = poll_seconds
self._last_seen_id = None
try:
plugin: Optional[NextcloudPlugin] = get_plugin('nextcloud')
if plugin:
self.url = plugin.conf.url
self.username = plugin.conf.username
self.password = plugin.conf.password
except Exception as e:
self.logger.info('NextCloud plugin not configured: {}'.format(str(e)))
self.url = url if url else self.url
self.username = username if username else self.username
self.password = password if password else self.password
assert self.url and self.username and self.password, \
'No configuration provided neither for the NextCloud plugin nor the backend'
@property
def last_seen_id(self) -> Optional[int]:
if self._last_seen_id is None:
variables: VariablePlugin = get_plugin('variable')
last_seen_id = variables.get(self._LAST_ACTIVITY_VARNAME).output.get(self._LAST_ACTIVITY_VARNAME)
self._last_seen_id = last_seen_id
return self._last_seen_id
@last_seen_id.setter
def last_seen_id(self, value: Optional[int]):
variables: VariablePlugin = get_plugin('variable')
variables.set(**{self._LAST_ACTIVITY_VARNAME: value})
self._last_seen_id = value
@staticmethod
def _activity_to_event(activity: dict) -> NextCloudActivityEvent:
return NextCloudActivityEvent(activity_type=activity.pop('type'), **activity)
def loop(self):
last_seen_id = int(self.last_seen_id)
new_last_seen_id = int(last_seen_id)
plugin: NextcloudPlugin = get_plugin('nextcloud')
# noinspection PyUnresolvedReferences
activities = plugin.get_activities(sort='desc', url=self.url, username=self.username, password=self.password,
object_type=self.object_type, object_id=self.object_id).output
events = []
for activity in activities:
if last_seen_id and activity['activity_id'] <= last_seen_id:
break
events.append(self._activity_to_event(activity))
if not new_last_seen_id or activity['activity_id'] > new_last_seen_id:
new_last_seen_id = int(activity['activity_id'])
for evt in events[::-1]:
self.bus.post(evt)
if new_last_seen_id and last_seen_id != new_last_seen_id:
self.last_seen_id = new_last_seen_id
# vim:sw=4:ts=4:et:

9
platypush/message/event/nextcloud.py Normal file
View file

@ -0,0 +1,9 @@
from platypush.message.event import Event
class NextCloudActivityEvent(Event):
def __init__(self, activity_id: int, activity_type: str, *args, **kwargs):
super().__init__(*args, activity_id=activity_id, activity_type=activity_type, **kwargs)
# vim:sw=4:ts=4:et:

853
platypush/plugins/nextcloud.py Normal file
View file

@ -0,0 +1,853 @@
import os
from dataclasses import dataclass
from datetime import datetime
from enum import IntEnum
from typing import Optional, List, Union, Dict
from platypush.plugins import Plugin, action
@dataclass
class ClientConfig:
url: str
username: str
password: str
def to_dict(self):
return {
'url': self.url,
'username': self.username,
'password': self.password,
}
class ShareType(IntEnum):
USER = 0
GROUP = 1
PUBLIC_LINK = 3
EMAIL = 4
FEDERATED_CLOUD_SHARE = 6
CIRCLE = 7
TALK_CONVERSATION = 10
class Permission(IntEnum):
READ = 1
UPDATE = 2
CREATE = 4
DELETE = 8
SHARE = 16
ALL = 31
class NextcloudPlugin(Plugin):
"""
Plugin to interact with a NextCloud instance.
Requires:
* **nextcloud-API** (``pip install git+https://github.com/EnterpriseyIntranet/nextcloud-API.git``)
"""
def __init__(self, url: Optional[str] = None, username: Optional[str] = None, password: Optional[str] = None,
**kwargs):
"""
:param url: URL to the index of your default NextCloud instance.
:param username: Default NextCloud username.
:param password: Default NextCloud password.
"""
super().__init__(**kwargs)
self.conf = ClientConfig(url=url, username=username, password=password)
self._client = self._get_client(**self.conf.to_dict())
def _get_client(self, url: Optional[str] = None, username: Optional[str] = None, password: Optional[str] = None,
raise_on_empty: bool = False):
from nextcloud import NextCloud
if not url:
if not self.conf.url:
if raise_on_empty:
raise AssertionError('No url/username/password provided')
return None
return NextCloud(endpoint=self.conf.url, user=self.conf.username, password=self.conf.password,
json_output=True)
return NextCloud(endpoint=url, user=username, password=password, json_output=True)
@staticmethod
def _get_permissions(permissions: Optional[List[str]]) -> int:
int_perm = 0
for perm in (permissions or []):
perm = perm.upper()
assert hasattr(Permission, perm), 'Unknown permissions type: {}. Supported permissions: {}'.format(
perm, [p.name.lower() for p in Permission])
if perm == 'ALL':
int_perm = Permission.ALL.value
break
int_perm += getattr(Permission, perm).value
return int_perm
@staticmethod
def _get_share_type(share_type: str) -> int:
share_type = share_type.upper()
assert hasattr(ShareType, share_type), 'Unknown share type: {}. Supported share types: {}'.format(
share_type, [s.name.lower() for s in ShareType])
return getattr(ShareType, share_type).value
def _execute(self, server_args: dict, method: str, *args, **kwargs):
client = self._get_client(**server_args)
assert hasattr(client, method), 'No such NextCloud method: {}'.format(method)
response = getattr(client, method)(*args, **kwargs)
if response is None:
return
assert response.is_ok, 'Error on {method}({args}{sep}{kwargs}): {error}'.format(
method=method,
args=', '.join(args),
sep=', ' if args and kwargs else '',
kwargs=', '.join(['{}={}'.format(k, v) for k, v in kwargs.items()]),
error=response.meta.get('message', '[No message]') if hasattr(response, 'meta') else response.raw.reason)
return response.data
@action
def get_activities(self, since: Optional[id] = None, limit: Optional[int] = None, object_type: Optional[str] = None,
object_id: Optional[int] = None, sort: str = 'desc', **server_args) -> List[str]:
"""
Get the list of recent activities on an instance.
:param since: Only return the activities that have occurred since the specified ID.
:param limit: Maximum number of activities to be returned (default: ``None``).
:param object_type: Filter by object type.
:param object_id: Only get the activities related to a specific ``object_id``.
:param sort: Sort mode, ``asc`` for ascending, ``desc`` for descending (default: ``desc``).
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
:return: The list of selected activities.
"""
return self._execute(server_args, 'get_activities', since=since, limit=limit, object_type=object_type,
object_id=object_id,
sort=sort)
@action
def get_apps(self, **server_args) -> List[str]:
"""
Get the list of apps installed on a NextCloud instance.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
:return: The list of installed apps as strings.
"""
return self._execute(server_args, 'get_apps').get('apps', [])
@action
def enable_app(self, app_id: Union[str, int], **server_args):
"""
Enable an app.
:param app_id: App ID.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
self._execute(server_args, 'enable_app', app_id)
@action
def disable_app(self, app_id: Union[str, int], **server_args):
"""
Disable an app.
:param app_id: App ID.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
self._execute(server_args, 'disable_app', app_id)
@action
def get_app(self, app_id: Union[str, int], **server_args) -> dict:
"""
Provides information about an application.
:param app_id: App ID.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
return self._execute(server_args, 'get_app', app_id)
@action
def get_capabilities(self, **server_args) -> dict:
"""
Returns the capabilities of the server.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
return self._execute(server_args, 'get_capabilities')
@action
def add_group(self, group_id: Union[str], **server_args):
"""
Create a new group.
:param group_id: New group unique ID.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
self._execute(server_args, 'add_group', group_id)
@action
def delete_group(self, group_id: Union[str], **server_args):
"""
Delete a group.
:param group_id: Group ID.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
self._execute(server_args, 'delete_group', group_id)
@action
def get_group(self, group_id: Union[str], **server_args) -> dict:
"""
Get the information of a group.
:param group_id: Group ID.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
return self._execute(server_args, 'get_group', group_id)
@action
def get_groups(self, search: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None,
**server_args) -> List[str]:
"""
Search for groups.
:param search: Search for groups matching the specified substring.
:param limit: Maximum number of returned entries.
:param offset: Start offset.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
return self._execute(server_args, 'get_groups', search=search, limit=limit, offset=offset).get('groups', [])
@action
def create_group_folder(self, name: str, **server_args):
"""
Create a new group folder.
:param name: Name/path of the folder.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
self._execute(server_args, 'create_group_folder', name)
@action
def delete_group_folder(self, folder_id: Union[int, str], **server_args):
"""
Delete a new group folder.
:param folder_id: Folder ID.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
self._execute(server_args, 'delete_group_folder', folder_id)
@action
def get_group_folder(self, folder_id: Union[int, str], **server_args) -> dict:
"""
Get a new group folder.
:param folder_id: Folder ID.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
return self._execute(server_args, 'get_group_folder', folder_id)
@action
def get_group_folders(self, **server_args) -> list:
"""
Get the list new group folder.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
return self._execute(server_args, 'get_group_folders')
@action
def rename_group_folder(self, folder_id: Union[int, str], new_name: str, **server_args):
"""
Rename a group folder.
:param folder_id: Folder ID.
:param new_name: New folder name.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
self._execute(server_args, 'rename_group_folder', folder_id, new_name)
@action
def grant_access_to_group_folder(self, folder_id: Union[int, str], group_id: str, **server_args):
"""
Grant access to a group folder to a given group.
:param folder_id: Folder ID.
:param group_id: Group ID.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
self._execute(server_args, 'grant_access_to_group_folder', folder_id, group_id)
@action
def revoke_access_to_group_folder(self, folder_id: Union[int, str], group_id: str, **server_args):
"""
Revoke access to a group folder to a given group.
:param folder_id: Folder ID.
:param group_id: Group ID.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
self._execute(server_args, 'revoke_access_to_group_folder', folder_id, group_id)
@action
def set_group_folder_quota(self, folder_id: Union[int, str], quota: Optional[int], **server_args):
"""
Set the quota of a group folder.
:param folder_id: Folder ID.
:param quota: Quota in bytes - set None for unlimited.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
self._execute(server_args, 'set_quota_of_group_folder', folder_id, quota if quota is not None else -3)
@action
def set_group_folder_permissions(self, folder_id: Union[int, str], group_id: str, permissions: List[str],
**server_args):
"""
Set the permissions on a folder for a group.
:param folder_id: Folder ID.
:param group_id: Group ID.
:param permissions: New permissions, as a list including any of the following:
- ``read``
- ``update``
- ``create``
- ``delete``
- ``share``
- ``all``
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
self._execute(server_args, 'set_permissions_to_group_folder', folder_id, group_id,
self._get_permissions(permissions))
@action
def get_notifications(self, **server_args) -> list:
"""
Get the list of notifications for the logged user.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
return self._execute(server_args, 'get_notifications')
@action
def delete_notifications(self, **server_args):
"""
Delete all notifications for the logged user.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
self._execute(server_args, 'delete_all_notifications')
@action
def get_notification(self, notification_id: int, **server_args) -> Union[dict, str]:
"""
Get the content of a notification.
:param notification_id: Notification ID.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
return self._execute(server_args, 'get_notification', notification_id)
@action
def delete_notification(self, notification_id: int, **server_args):
"""
Delete a notification.
:param notification_id: Notification ID.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
self._execute(server_args, 'delete_notification', notification_id)
@action
def create_share(self, path: str, share_type: str, share_with: Optional[str] = None, public_upload: bool = False,
password: Optional[str] = None, permissions: Optional[List[str]] = None, **server_args) -> dict:
"""
Share a file/folder with a user/group or a public link.
:param path: Path to the resource to be shared.
:param share_type: Share type. Supported values:
- ``user``
- ``group``
- ``public_link``
- ``email``
- ``federated_cloud_share``
- ``circle``
- ``talk_conversation``
:param share_with: User/group ID, email or conversation ID the resource should be shared with.
:param public_upload: Whether public upload to the shared folder is allowed (default: False).
:param password: Optional password to protect the share.
:param permissions: Share permissions, as a list including any of the following (default: ``read``):
- ``read``
- ``update``
- ``create``
- ``delete``
- ``share``
- ``all``
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
:return: The details of the newly created share. Example:
.. code-block:: json
{
"id": "4",
"share_type": 3,
"uid_owner": "your_uid",
"displayname_owner": "Your Name",
"permissions": 17,
"can_edit": true,
"can_delete": true,
"stime": 1599691325,
"parent": null,
"expiration": null,
"token": "AbCdEfG0123456789",
"uid_file_owner": "your_uid",
"note": "",
"label": "",
"displayname_file_owner": "Your Name",
"path": "/Shared Path",
"item_type": "folder",
"mimetype": "httpd/unix-directory",
"storage_id": "home::your-uid",
"storage": 2,
"item_source": 13960,
"file_source": 13960,
"file_parent": 6,
"file_target": "/Shared Path",
"share_with": null,
"share_with_displayname": "(Shared link)",
"password": null,
"send_password_by_talk": false,
"url": "https://your-domain/nextcloud/index.php/s/AbCdEfG0123456789",
"mail_send": 1,
"hide_download": 0
}
"""
share_type = self._get_share_type(share_type)
permissions = self._get_permissions(permissions or ['read'])
return self._execute(server_args, 'create_share', path, share_type=share_type, share_with=share_with,
public_upload=public_upload,
password=password, permissions=permissions)
@action
def get_shares(self, **server_args) -> List[dict]:
"""
Get the list of shares available on the server.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
:return: List of available shares. Example:
.. code-block:: json
[
{
"id": "4",
"share_type": 3,
"uid_owner": "your_uid",
"displayname_owner": "Your Name",
"permissions": 17,
"can_edit": true,
"can_delete": true,
"stime": 1599691325,
"parent": null,
"expiration": null,
"token": "AbCdEfG0123456789",
"uid_file_owner": "your_uid",
"note": "",
"label": "",
"displayname_file_owner": "Your Name",
"path": "/Shared Path",
"item_type": "folder",
"mimetype": "httpd/unix-directory",
"storage_id": "home::your-uid",
"storage": 2,
"item_source": 13960,
"file_source": 13960,
"file_parent": 6,
"file_target": "/Shared Path",
"share_with": null,
"share_with_displayname": "(Shared link)",
"password": null,
"send_password_by_talk": false,
"url": "https://your-domain/nextcloud/index.php/s/AbCdEfG0123456789",
"mail_send": 1,
"hide_download": 0
}
]
"""
return self._execute(server_args, 'get_shares')
@action
def delete_share(self, share_id: int, **server_args):
"""
Remove the shared state of a resource.
:param share_id: Share ID.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
self._execute(server_args, 'delete_share', str(share_id))
@action
def get_share(self, share_id: int, **server_args) -> dict:
"""
Get the information of a shared resource.
:param share_id: Share ID.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
return self._execute(server_args, 'get_share_info', str(share_id))
@action
def update_share(self, share_id: int, public_upload: Optional[bool] = None, password: Optional[str] = None,
permissions: Optional[List[str]] = None, expire_date: Optional[str] = None, **server_args):
"""
Update the permissions of a shared resource.
:param share_id: Share ID.
:param public_upload: Whether public upload to the shared folder is allowed (default: False).
:param password: Optional password to protect the share.
:param permissions: Share permissions, as a list including any of the following (default: ``read``):
- ``read``
- ``update``
- ``create``
- ``delete``
- ``share``
- ``all``
:param expire_date: Share expiration date, in the format ``YYYY-MM-DD``.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
if permissions:
permissions = self._get_permissions(permissions)
self._execute(server_args, 'update_share', share_id, public_upload=public_upload, password=password,
permissions=permissions, expire_date=expire_date)
@action
def create_user(self, user_id: str, password: str, **server_args):
"""
Create a user.
:param user_id: User ID/name.
:param password: User password
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
self._execute(server_args, 'add_user', user_id, password)
@action
def edit_user(self, user_id: str, properties: Dict[str, str], **server_args):
"""
Update a set of properties of a user.
:param user_id: User ID/name.
:param properties: Key-value pair of user attributes to be edited.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
for k, v in properties.items():
self._execute(server_args, 'edit_user', user_id, k, v)
@action
def get_user(self, user_id: str, **server_args) -> dict:
"""
Get the details of a user.
:param user_id: User ID/name.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
:return: User details. Example:
.. code-block:: json
{
"enabled": true,
"storageLocation": "/mnt/hd/nextcloud/user",
"id": "user",
"lastLogin": 1599693750000,
"backend": "Database",
"subadmin": [],
"quota": {
"free": 6869434515456,
"used": 1836924441,
"total": 6871271439897,
"relative": 0.03,
"quota": -3
},
"email": "info@yourdomain.com",
"displayname": "Your Name",
"phone": "+1234567890",
"address": "",
"website": "https://yourdomain.com",
"twitter": "@You",
"groups": [
"admin"
],
"language": "en",
"locale": "",
"backendCapabilities": {
"setDisplayName": true,
"setPassword": true
}
}
"""
return self._execute(server_args, 'get_user', user_id)
@action
def get_users(self, search: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None,
**server_args) -> List[str]:
"""
Get the list of users matching some search criteria.
:param search: Return users matching the provided string.
:param limit: Maximum number of results to be returned (default: no limit).
:param offset: Search results offset (default: None).
:return: List of the matched user IDs.
"""
return self._execute(server_args, 'get_users', search=search, limit=limit, offset=offset)
@action
def delete_user(self, user_id: str, **server_args):
"""
Delete a user.
:param user_id: User ID/name.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
self._execute(server_args, 'delete_user', user_id)
@action
def enable_user(self, user_id: str, **server_args):
"""
Enable a user.
:param user_id: User ID/name.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
self._execute(server_args, 'enable_user', user_id)
@action
def disable_user(self, user_id: str, **server_args):
"""
Disable a user.
:param user_id: User ID/name.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
self._execute(server_args, 'disable_user', user_id)
@action
def add_to_group(self, user_id: str, group_id: str, **server_args):
"""
Add a user to a group.
:param user_id: User ID/name.
:param group_id: Group ID.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
self._execute(server_args, 'add_to_group', user_id, group_id)
@action
def remove_from_group(self, user_id: str, group_id: str, **server_args):
"""
Remove a user from a group.
:param user_id: User ID/name.
:param group_id: Group ID.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
self._execute(server_args, 'remove_from_group', user_id, group_id)
@action
def create_subadmin(self, user_id: str, group_id: str, **server_args):
"""
Add a user as a subadmin for a group.
:param user_id: User ID/name.
:param group_id: Group ID.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
self._execute(server_args, 'create_subadmin', user_id, group_id)
@action
def remove_subadmin(self, user_id: str, group_id: str, **server_args):
"""
Remove a user as a subadmin from a group.
:param user_id: User ID/name.
:param group_id: Group ID.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
self._execute(server_args, 'remove_subadmin', user_id, group_id)
@action
def get_subadmin_groups(self, user_id: str, **server_args) -> List[str]:
"""
Get the groups where a given user is subadmin.
:param user_id: User ID/name.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
:return: List of matched groups as strings.
"""
return self._execute(server_args, 'get_subadmin_groups', user_id)
@action
def create_folder(self, path: str, user_id: Optional[str] = None, **server_args):
"""
Create a folder.
:param path: Path to the folder.
:param user_id: User ID associated to the folder (default: same as the configured user).
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
user_id = user_id or server_args.get('username', self.conf.username)
self._execute(server_args, 'create_folder', user_id, path)
@action
def delete_path(self, path: str, user_id: Optional[str] = None, **server_args):
"""
Delete a file or folder.
:param path: Path to the resource.
:param user_id: User ID associated to the resource (default: same as the configured user).
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
user_id = user_id or server_args.get('username', self.conf.username)
self._execute(server_args, 'delete_path', user_id, path)
@action
def upload_file(self, remote_path: str, local_path: Optional[str] = None, content: Optional[str] = None,
user_id: Optional[str] = None, timestamp: Optional[Union[datetime, int, str]] = None, **server_args):
"""
Upload a file.
:param remote_path: Path to the remote resource.
:param local_path: If set, identifies the path to the local file to be uploaded.
:param content: If set, create a new file with this content instead of uploading an existing file.
:param user_id: User ID associated to the resource (default: same as the configured user).
:param timestamp: File timestamp. If not specified it will be retrieved from the file info or set to ``now``
if ``content`` is specified.
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
user_id = user_id or server_args.get('username', self.conf.username)
if isinstance(timestamp, str):
timestamp = datetime.fromisoformat(timestamp)
if isinstance(timestamp, datetime):
timestamp = int(timestamp.timestamp())
assert (local_path or content) and not (local_path and content), 'Please specify either local_path or content'
if local_path:
method = 'upload_file'
local_path = os.path.abspath(os.path.expanduser(local_path))
else:
method = 'upload_file_contents'
return self._execute(server_args, method, user_id, local_path or content, remote_path, timestamp=timestamp)
@action
def download_file(self, remote_path: str, local_path: str, user_id: Optional[str] = None, **server_args):
"""
Download a file.
:param remote_path: Path to the remote resource.
:param local_path: Path to the local folder.
:param user_id: User ID associated to the resource (default: same as the configured user).
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
user_id = user_id or server_args.get('username', self.conf.username)
local_path = os.path.abspath(os.path.expanduser(local_path))
cur_dir = os.getcwd()
try:
os.chdir(local_path)
return self._execute(server_args, 'download_file', user_id, remote_path)
finally:
os.chdir(cur_dir)
@action
def list(self, path: str, user_id: Optional[str] = None, depth: int = 1, all_properties: bool = False,
**server_args) -> List[dict]:
"""
List the content of a folder on the NextCloud instance.
:param path: Remote path.
:param user_id: User ID associated to the resource (default: same as the configured user).
:param depth: Search depth (default: 1).
:param all_properties: Return all the file properties available (default: ``False``).
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
user_id = user_id or server_args.get('username', self.conf.username)
return self._execute(server_args, 'list_folders', user_id, path, depth=depth, all_properties=all_properties)
@action
def list_favorites(self, path: Optional[str] = None, user_id: Optional[str] = None, **server_args) -> List[dict]:
"""
List the favorite items for a user.
:param path: Return only the favorites under this path.
:param user_id: User ID associated to the resource (default: same as the configured user).
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
user_id = user_id or server_args.get('username', self.conf.username)
return self._execute(server_args, 'list_folders', user_id, path)
@action
def mark_favorite(self, path: Optional[str] = None, user_id: Optional[str] = None, **server_args):
"""
Add a path to a user's favorites.
:param path: Resource path.
:param user_id: User ID associated to the resource (default: same as the configured user).
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
user_id = user_id or server_args.get('username', self.conf.username)
self._execute(server_args, 'set_favorites', user_id, path)
@action
def copy(self, path: str, destination: str, user_id: Optional[str] = None, overwrite: bool = False, **server_args):
"""
Copy a resource to another path.
:param path: Resource path.
:param destination: Destination path.
:param user_id: User ID associated to the resource (default: same as the configured user).
:param overwrite: Set to ``True`` if you want to overwrite any existing file (default: ``False``).
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
user_id = user_id or server_args.get('username', self.conf.username)
self._execute(server_args, 'copy_path', user_id, path, destination, overwrite=overwrite)
@action
def move(self, path: str, destination: str, user_id: Optional[str] = None, overwrite: bool = False, **server_args):
"""
Move a resource to another path.
:param path: Resource path.
:param destination: Destination path.
:param user_id: User ID associated to the resource (default: same as the configured user).
:param overwrite: Set to ``True`` if you want to overwrite any existing file (default: ``False``).
:param server_args: Override the default server settings (see :meth:`._get_client` arguments).
"""
user_id = user_id or server_args.get('username', self.conf.username)
self._execute(server_args, 'move_path', user_id, path, destination, overwrite=overwrite)
# vim:sw=4:ts=4:et:

5
requirements.txt
View file

@ -146,7 +146,7 @@ pyScss
# Support for NFC tags
# nfcpy >= 1.0
# ndef
# ndeflib
# Support for enviropHAT
# envirophat
@ -295,3 +295,6 @@ croniter
# Support for IMAP mail integration
# imapclient
# Support for NextCloud integration
# git+https://github.com/EnterpriseyIntranet/nextcloud-API.git

4
setup.py
View file

@ -236,7 +236,7 @@ setup(
# Support for mpv player plugin
'mpv': ['python-mpv'],
# Support for NFC tags
'nfc': ['nfcpy>=1.0', 'ndef'],
'nfc': ['nfcpy>=1.0', 'ndeflib'],
# Support for enviropHAT
'envirophat': ['envirophat'],
# Support for GPS
@ -332,5 +332,7 @@ setup(
'lcd': ['RPi.GPIO', 'RPLCD'],
# Support for IMAP mail integration
'imap': ['imapclient'],
# Support for NextCloud integration
'nextcloud': ['nextcloud-API @ git+https://github.com/EnterpriseyIntranet/nextcloud-API.git'],
},
)