import os
from dataclasses import dataclass
from datetime import datetime
from enum import IntEnum
from typing import Optional, List, Union, Dict
from platypush.context import Variable
from platypush.message.event.nextcloud import NextCloudActivityEvent
from platypush.plugins import RunnablePlugin, action
[docs]
@dataclass
class ClientConfig:
"""
Configuration for the NextCloud client.
"""
url: str
username: str
password: str
def to_dict(self):
return {
'url': self.url,
'username': self.username,
'password': self.password,
}
[docs]
class ShareType(IntEnum):
"""
Enumerates the types of shares that can be created.
"""
USER = 0
GROUP = 1
PUBLIC_LINK = 3
EMAIL = 4
FEDERATED_CLOUD_SHARE = 6
CIRCLE = 7
TALK_CONVERSATION = 10
[docs]
class Permission(IntEnum):
"""
Enumerates the permissions that can be granted to a user/group on a
resource.
"""
READ = 1
UPDATE = 2
CREATE = 4
DELETE = 8
SHARE = 16
ALL = 31
_LAST_ACTIVITY_VARNAME = '_NEXTCLOUD_LAST_ACTIVITY_ID'
[docs]
class NextcloudPlugin(RunnablePlugin):
"""
Plugin to interact with a NextCloud instance.
"""
[docs]
def __init__(
self,
url: str,
username: str,
password: str,
poll_interval: Optional[float] = 30.0,
**kwargs,
):
"""
:param url: URL to the index of your default NextCloud instance.
:param username: Default NextCloud username.
:param password: Default NextCloud password. If the user has 2FA
enabled then you can generate an app password from the personal
settings page of your NextCloud instance.
:param poll_interval: How often the plugin should poll for new activity
events (default: 30 seconds).
"""
super().__init__(poll_interval=poll_interval, **kwargs)
self.conf = ClientConfig(url=url, username=username, password=password)
self._client = self._get_client(**self.conf.to_dict(), raise_on_empty=False)
self._last_seen_id = None
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,
)
return NextCloud(endpoint=url, user=username, password=password)
@property
def _last_seen_var(self) -> Variable:
return Variable(_LAST_ACTIVITY_VARNAME)
@property
def last_seen_id(self) -> int:
if self._last_seen_id is None:
self._last_seen_id = self._last_seen_var.get() or 0
return int(self._last_seen_id)
@last_seen_id.setter
def last_seen_id(self, value: int = 0):
self._last_seen_var.set(value)
self._last_seen_id = value
@staticmethod
def _activity_to_event(activity: dict) -> NextCloudActivityEvent:
activity_type = activity.pop('type')
dt = datetime.fromisoformat(activity.pop('datetime'))
return NextCloudActivityEvent(
activity_type=activity_type, datetime=dt, **activity
)
@staticmethod
def _get_permissions(permissions: Optional[List[str]]) -> int:
int_perm = 0
for perm in permissions or []:
perm = perm.upper()
assert hasattr(Permission, perm), (
f'Unknown permissions type: {perm}. Supported permissions: '
f'{[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), (
f'Unknown share type: {share_type}. Supported share types: '
f'{[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)
func = getattr(client, method, None)
assert func, f'No such NextCloud method: {method}'
response = func(*args, **kwargs)
assert response is not None, 'No response from NextCloud server'
assert response.is_ok, (
'Error on '
+ method
+ '('
+ ', '.join(args)
+ (', ' if args and kwargs else '')
+ ', '.join([f'{k}={v}' for k, v in kwargs.items()])
+ '): status='
+ str(response.status_code)
+ ', message='
+ getattr(response, 'meta', {}).get('message', '[No message]')
+ ', data='
+ str(response.json_data)
)
return response.json_data
def _get_activities(
self,
since: Optional[str] = None,
limit: Optional[int] = 25,
object_type: Optional[str] = None,
object_id: Optional[int] = None,
sort: str = 'desc',
**server_args,
) -> List[dict]:
return self._execute(
server_args,
'get_activities',
since=since,
limit=limit,
object_type=object_type,
object_id=object_id,
sort=sort,
)
[docs]
@action
def get_activities(
self,
since: Optional[str] = None,
limit: Optional[int] = 25,
object_type: Optional[str] = None,
object_id: Optional[int] = None,
sort: str = 'desc',
**server_args,
) -> List[dict]:
"""
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: 25).
: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._get_activities(
since=since,
limit=limit,
object_type=object_type,
object_id=object_id,
sort=sort,
**server_args,
)
[docs]
@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', [])
[docs]
@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)
[docs]
@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)
[docs]
@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)
[docs]
@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: Server capabilities. Example:
.. code-block:: json
{
"version": {
"major": 27,
"minor": 1,
"micro": 5,
"string": "27.1.5",
"edition": "",
"extendedSupport": false
},
"capabilities": {
"core": {
"pollinterval": 60,
"webdav-root": "remote.php/webdav",
},
"files": {
"bigfilechunking": true,
"blacklisted_files": [
".htaccess"
],
"comments": true,
"undelete": true,
"versioning": true,
"version_labeling": true,
"version_deletion": true
},
"activity": {
"apiv2": [
"filters",
"filters-api",
"previews",
"rich-strings"
]
},
"notifications": {
"ocs-endpoints": [
"list",
"get",
"delete",
"delete-all",
"icons",
"rich-strings",
"action-web",
"user-status",
"exists"
],
"push": [
"devices",
"object-data",
"delete"
],
"admin-notifications": [
"ocs",
"cli"
]
},
"theming": {
"name": "Nextcloud",
"url": "https://nextcloud.com",
"slogan": "a safe home for all your data",
"color": "#1F5C98",
"color-text": "#ffffff",
"color-element": "#1F5C98",
"color-element-bright": "#1F5C98",
"color-element-dark": "#1F5C98",
"logo": "https://cloud.example.com/core/img/logo/logo.svg?v=0",
"background": "#1F5C98",
"background-plain": true,
"background-default": true,
"logoheader": "https://cloud.example.com/core/img/logo/logo.svg?v=0",
"favicon": "https://cloud.example.com/core/img/logo/logo.svg?v=0"
},
"user_status": {
"enabled": true,
"restore": true,
"supports_emoji": true
},
}
}
"""
return self._execute(server_args, 'get_capabilities')
[docs]
@action
def add_group(self, group_id: Union[str, int], **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)
[docs]
@action
def delete_group(self, group_id: Union[str, int], **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)
[docs]
@action
def get_group(self, group_id: Union[str, int], **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)
[docs]
@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', [])
[docs]
@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)
[docs]
@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)
[docs]
@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)
[docs]
@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')
[docs]
@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)
[docs]
@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)
[docs]
@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)
[docs]
@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,
)
[docs]
@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),
)
[docs]
@action
def get_notifications(self, **server_args) -> List[dict]:
"""
Get the list of notifications for the logged user.
:param server_args: Override the default server settings (see
:meth:`._get_client` arguments).
:return: A list of notifications. Example:
.. code-block:: json
[
{
"notification_id": 1234,
"app": "updatenotification",
"user": "username",
"datetime": "2024-01-01T19:00:00+00:00",
"object_type": "side_menu",
"object_id": "1.2.3",
"subject": "Update for MyApp to version 1.2.3 is available.",
"message": "",
"link": "https://cloud.example.com/index.php/settings/apps/updates#myapp",
"subjectRich": "Update for {app} to version 1.2.3 is available.",
"subjectRichParameters": {
"app": {
"type": "app",
"id": "myapp",
"name": "MyApp"
}
},
"messageRich": "",
"messageRichParameters": [],
"icon": "https://cloud.example.com/apps/updatenotification/img/notification.svg",
"shouldNotify": true,
"actions": []
}
]
"""
return self._execute(server_args, 'get_notifications')
[docs]
@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')
[docs]
@action
def get_notification(self, notification_id: int, **server_args) -> dict:
"""
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: Notification details. Example:
.. code-block:: json
{
"notification_id": 1234,
"app": "updatenotification",
"user": "username",
"datetime": "2024-01-01T19:00:00+00:00",
"object_type": "side_menu",
"object_id": "1.2.3",
"subject": "Update for MyApp to version 1.2.3 is available.",
"message": "",
"link": "https://cloud.example.com/index.php/settings/apps/updates#myapp",
"subjectRich": "Update for {app} to version 1.2.3 is available.",
"subjectRichParameters": {
"app": {
"type": "app",
"id": "myapp",
"name": "MyApp"
}
},
"messageRich": "",
"messageRichParameters": [],
"icon": "https://cloud.example.com/apps/updatenotification/img/notification.svg",
"shouldNotify": true,
"actions": []
}
"""
return self._execute(server_args, 'get_notification', notification_id)
[docs]
@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)
[docs]
@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_id = self._get_share_type(share_type)
permissions_id = self._get_permissions(permissions or ['read'])
return self._execute(
server_args,
'create_share',
path,
share_type=share_type_id,
share_with=share_with,
public_upload=public_upload,
password=password,
permissions=permissions_id,
)
[docs]
@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')
[docs]
@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))
[docs]
@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: The details of the 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
}
"""
return self._execute(server_args, 'get_share_info', str(share_id))
[docs]
@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).
"""
perms = self._get_permissions(permissions) if permissions else None
self._execute(
server_args,
'update_share',
share_id,
public_upload=public_upload,
password=password,
permissions=perms,
expire_date=expire_date,
)
[docs]
@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)
[docs]
@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)
[docs]
@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)
[docs]
@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. 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_users', search=search, limit=limit, offset=offset
)
[docs]
@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)
[docs]
@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)
[docs]
@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)
[docs]
@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)
[docs]
@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)
[docs]
@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)
[docs]
@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)
[docs]
@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)
[docs]
@action
def create_folder(self, path: str, **server_args):
"""
Create a folder.
:param path: Path to the folder.
:param server_args: Override the default server settings (see
:meth:`._get_client` arguments).
"""
self._execute(server_args, 'create_folder', path)
[docs]
@action
def mkdir(self, path: str, **server_args):
"""
Alias for :meth:`.create_folder`.
:param path: Path to the folder.
:param server_args: Override the default server settings (see
:meth:`._get_client` arguments).
"""
self.create_folder(path, **server_args)
[docs]
@action
def delete(self, path: str, **server_args):
"""
Delete a file or folder.
:param path: Path to the resource.
:param server_args: Override the default server settings (see
:meth:`._get_client` arguments).
"""
self._execute(server_args, 'delete_path', path)
[docs]
@action
def upload_file(
self,
remote_path: str,
local_path: Optional[str] = None,
content: 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 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).
"""
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,
local_path or content,
remote_path,
timestamp=timestamp,
)
[docs]
@action
def download_file(
self,
remote_path: str,
local_path: str,
**server_args,
):
"""
Download a file.
:param remote_path: Path to the remote resource.
:param local_path: Path to the local folder.
:param server_args: Override the default server settings (see
:meth:`._get_client` arguments).
"""
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', remote_path)
finally:
os.chdir(cur_dir)
[docs]
@action
def list(
self,
path: str,
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 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).
"""
return self._execute(
server_args,
'list_folders',
path,
depth=depth,
all_properties=all_properties,
)
[docs]
@action
def list_favorites(self, path: Optional[str] = None, **server_args) -> List[dict]:
"""
List the favorite items for a user.
:param path: Return only the favorites under this path.
:param server_args: Override the default server settings (see
:meth:`._get_client` arguments).
"""
return self._execute(server_args, 'list_folders', path)
[docs]
@action
def mark_favorite(self, path: Optional[str] = None, **server_args):
"""
Add a path to a user's favorites.
:param path: Resource path.
:param server_args: Override the default server settings (see
:meth:`._get_client` arguments).
"""
self._execute(server_args, 'set_favorites', path)
[docs]
@action
def copy(
self,
path: str,
destination: str,
overwrite: bool = False,
**server_args,
):
"""
Copy a resource to another path.
:param path: Resource path.
:param destination: Destination path.
: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).
"""
self._execute(server_args, 'copy_path', path, destination, overwrite=overwrite)
[docs]
@action
def move(
self,
path: str,
destination: str,
overwrite: bool = False,
**server_args,
):
"""
Move a resource to another path.
:param path: Resource path.
:param destination: Destination path.
: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).
"""
self._execute(server_args, 'move_path', path, destination, overwrite=overwrite)
[docs]
def main(self):
while not self.should_stop():
last_seen_id = self.last_seen_id
new_last_seen_id = last_seen_id
activities = self._get_activities(
sort='desc',
url=self.conf.url,
username=self.conf.username,
password=self.conf.password,
limit=100,
)
events = []
for activity in activities:
activity_id = int(activity['activity_id'])
if last_seen_id and activity_id <= last_seen_id:
break
events.append(self._activity_to_event(activity))
if not new_last_seen_id or activity_id > new_last_seen_id:
new_last_seen_id = 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
self.wait_stop(self.poll_interval)
# vim:sw=4:ts=4:et:
