platypush/platypush/plugins/dropbox.py

import base64
import os

from platypush.plugins import Plugin, action


class DropboxPlugin(Plugin):
    """
    Plugin to manage a Dropbox account and its files and folders.

    Requires:

        * **dropbox** (``pip install dropbox``)
    """

    def __init__(self, access_token, **kwargs):
        """
        :param access_token: Dropbox API access token. You can get yours by creating an app on
            https://dropbox.com/developers/apps
        :type access_token: str
        """

        super().__init__(**kwargs)
        self.access_token = access_token
        self._dbx = None

    def _get_instance(self):
        """
        :rtype: :class:`dropbox.Dropbox`
        """
        # noinspection PyPackageRequirements
        import dropbox

        if not self._dbx:
            self._dbx = dropbox.Dropbox(self.access_token)

        return self._dbx

    @action
    def get_account(self, account_id=None):
        """
        Get the information about a linked Dropbox account

        :param account_id: account_id. If none is specified then it will retrieve the current user's account id
        :type account_id: str

        :return: dict with the following attributes:
            account_id, name, email, email_verified, disabled, profile_photo_url, team_member_id
        """

        dbx = self._get_instance()

        if not account_id:
            acc = dbx.users_get_current_account()
        else:
            acc = dbx.users_get_account(account_id)

        return {
            'account_id': acc.account_id,
            'name': acc.name.display_name,
            'email': acc.email,
            'email_verified': acc.email_verified,
            'disabled': acc.disabled,
            'profile_photo_url': acc.profile_photo_url,
            'team_member_id': acc.team_member_id,
        }

    @action
    def get_usage(self):
        """
        Get the amount of allocated and used remote space

        :return: dict
        """

        dbx = self._get_instance()
        usage = dbx.users_get_space_usage()

        return {
            'used': usage.used,
            'allocated': usage.allocation.get_individual().allocated,
        }

    @action
    def list(self, folder=''):
        """
        Returns the files in a folder

        :param folder: Folder name (default: root)
        :type folder: str

        :return: dict
        """

        from dropbox.files import FileMetadata

        dbx = self._get_instance()
        files = dbx.files_list_folder(folder).entries
        entries = []

        for item in files:
            entry = {
                attr: getattr(item, attr)
                for attr in ['id', 'name', 'path_display', 'path_lower',
                             'parent_shared_folder_id', 'property_groups']
            }

            if item.sharing_info:
                entry['sharing_info'] = {
                    attr: getattr(item.sharing_info, attr)
                    for attr in ['no_access', 'parent_shared_folder_id', 'read_only',
                                 'shared_folder_id', 'traverse_only']
                }
            else:
                entry['sharing_info'] = {}

            if isinstance(item, FileMetadata):
                entry['client_modified'] = item.client_modified.isoformat()
                entry['server_modified'] = item.server_modified.isoformat()

                for attr in ['content_hash', 'has_explicit_shared_members', 'is_downloadable', 'rev', 'size']:
                    if hasattr(item, attr):
                        entry[attr] = getattr(item, attr)

            entries.append(entry)

        return entries

    @action
    def copy(self, from_path: str, to_path: str, allow_shared_folder=True, autorename=False,
             allow_ownership_transfer=False):
        """
        Copy a file or folder to a different location in the user's Dropbox. If the source path is a folder all
        its contents will be copied.

        :param from_path: Source path
        :param to_path: Destination path
        :param bool allow_shared_folder: If true, :meth:`files_copy` will copy
            contents in shared folder, otherwise
            ``RelocationError.cant_copy_shared_folder`` will be returned if
            ``from_path`` contains shared folder. This field is always true for
            :meth:`files_move`.
        :param bool autorename: If there's a conflict, have the Dropbox server
            try to autorename the file to avoid the conflict.
        :param bool allow_ownership_transfer: Allow moves by owner even if it
            would result in an ownership transfer for the content being moved.
            This does not apply to copies.
        """

        dbx = self._get_instance()
        dbx.files_copy_v2(from_path, to_path, allow_shared_folder=allow_shared_folder,
                          autorename=autorename, allow_ownership_transfer=allow_ownership_transfer)

    @action
    def move(self, from_path: str, to_path: str, allow_shared_folder=True, autorename=False,
             allow_ownership_transfer=False):
        """
        Move a file or folder to a different location in the user's Dropbox. If the source path is a folder all its
        contents will be moved.

        :param from_path: Source path
        :param to_path: Destination path
        :param bool allow_shared_folder: If true, :meth:`files_copy` will copy
            contents in shared folder, otherwise
            ``RelocationError.cant_copy_shared_folder`` will be returned if
            ``from_path`` contains shared folder. This field is always true for
            :meth:`files_move`.
        :param bool autorename: If there's a conflict, have the Dropbox server
            try to autorename the file to avoid the conflict.
        :param bool allow_ownership_transfer: Allow moves by owner even if it
            would result in an ownership transfer for the content being moved.
            This does not apply to copies.
        """

        dbx = self._get_instance()
        dbx.files_move_v2(from_path, to_path, allow_shared_folder=allow_shared_folder,
                          autorename=autorename, allow_ownership_transfer=allow_ownership_transfer)

    @action
    def delete(self, path: str):
        """
        Delete the file or folder at a given path. If the path is a folder, all its contents will be deleted too

        :param str path: Path to be removed
        """

        dbx = self._get_instance()
        dbx.files_delete_v2(path)

    @action
    def restore(self, path: str, rev: str):
        """
        Restore a specific revision of a file to the given path.

        :param str path: Path to be removed
        :param str rev: Revision ID to be restored
        """

        dbx = self._get_instance()
        dbx.files_restore(path, rev)

    @action
    def mkdir(self, path: str):
        """
        Create a folder at a given path.

        :param str path: Folder path
        """

        dbx = self._get_instance()
        dbx.files_create_folder_v2(path)

    @action
    def save(self, path: str, url: str):
        """
        Save a specified URL into a file in user's Dropbox. If the given path already exists, the file will be renamed
        to avoid the conflict (e.g. myfile (1).txt).

        :param path: Dropbox destination path
        :param url: URL to download
        """

        dbx = self._get_instance()
        dbx.files_save_url(path, url)

    def _file_download(self, path: str, download_path=None):
        dbx = self._get_instance()
        metadata, response = dbx.files_download(path)
        ret = self._parse_metadata(metadata)
        ret['encoding'] = response.encoding or response.apparent_encoding

        if download_path:
            if os.path.isdir(download_path):
                download_path = os.path.join(download_path, metadata.name)

            with open(download_path, 'wb') as f:
                f.write(response.content)
            ret['file'] = download_path
        else:
            if ret['encoding'] in ('ascii', 'unicode', 'utf-8'):
                ret['content'] = response.text
            else:
                ret['content'] = base64.encodebytes(response.content).decode().strip()

        return ret

    def _file_download_zip(self, path: str, download_path=None):
        dbx = self._get_instance()
        result, response = dbx.files_download_zip(path)
        ret = self._parse_metadata(result.metadata)

        if download_path:
            if os.path.isdir(download_path):
                download_path = os.path.join(download_path, result.metadata.name + '.zip')

            with open(download_path, 'wb') as f:
                f.write(response.content)
            ret['file'] = download_path
        else:
            ret['content'] = base64.encodebytes(response.content).decode().strip()

        return ret

    # noinspection PyShadowingBuiltins
    @action
    def download(self, path: str, download_path=None, zip=False):
        """
        Download a file or a zipped directory from a user's Dropbox.

        :param str path: Dropbox destination path
        :param str download_path: Destination path on the local machine (optional)
        :param bool zip: If set then the content will be downloaded in zip format (default: False)
        :rtype: dict
        :return: A dictionary with keys: ``('id', 'name', 'parent_shared_folder_id', 'path', 'size', 'encoding',
            'content_hash')``. If download_path is set 'file' is also returned. Otherwise 'content' will be returned.
            If it's a text file then 'content' will contain its string representation, otherwise its base64-encoded
            representation.
        """

        from dropbox.files import FolderMetadata

        if download_path:
            download_path = os.path.abspath(os.path.expanduser(download_path))

        dbx = self._get_instance()

        metadata = dbx.files_get_metadata(path)
        if isinstance(metadata, FolderMetadata):
            zip = True

        if zip:
            return self._file_download_zip(path, download_path)

        return self._file_download(path, download_path)

    @action
    def get_metadata(self, path: str):
        """
        Get the metadata of the specified path

        :param str path: Path to the resource
        :rtype dict:
        """

        dbx = self._get_instance()
        metadata = dbx.files_get_metadata(path)
        return self._parse_metadata(metadata)

    @staticmethod
    def _parse_metadata(metadata):
        from dropbox.files import FileMetadata, FolderMetadata

        ret = {
            'name': metadata.name,
            'parent_shared_folder_id': metadata.parent_shared_folder_id,
            'path': metadata.path_display,
        }

        if isinstance(metadata, FileMetadata):
            ret['type'] = 'file'
            ret['id'] = metadata.id
            ret['size'] = metadata.size
            ret['content_hash'] = metadata.content_hash
            ret['rev'] = metadata.rev
        elif isinstance(metadata, FolderMetadata):
            ret['type'] = 'folder'
            ret['id'] = metadata.id
            ret['shared_folder_id'] = metadata.shared_folder_id

        return ret

    @action
    def search(self, query: str, path='', start=0, max_results=100, content=False):
        """
        Searches for files and folders. Note: Recent changes may not immediately
        be reflected in search results due to a short delay in indexing.

        :param str path: The path in the user's Dropbox to search. Should probably be a folder.
        :param str query: The string to search for. The search string is split
            on spaces into multiple tokens. For file name searching, the last
            token is used for prefix matching (i.e. "bat c" matches "bat cave"
            but not "batman car").
        :param long start: The starting index within the search results (used for paging).
        :param long max_results: The maximum number of search results to return.
        :param content: Search also in files content (default: False)
        :rtype dict:
        :return: Dictionary with the following fields: ``('matches', 'start')``.
        """

        from dropbox.files import SearchMode

        dbx = self._get_instance()
        response = dbx.files_search(query=query, path=path, start=start, max_results=max_results,
                                    mode=SearchMode.filename_and_content if content else SearchMode.filename)

        results = [self._parse_metadata(match.metadata) for match in response.matches]

        return {
            'results': results,
            'start': response.start,
        }

    @action
    def upload(self, file=None, text=None, path='/', overwrite=False, autorename=False):
        """
        Create a new file with the contents provided in the request. Do not use this to upload a file larger than 150 MB

        :param str file: File to be uploaded
        :param str text: Text content to be uploaded
        :param str path: Path in the user's Dropbox to save the file.
        :param bool overwrite: If set, in case of conflict the file will be overwritten (default: append content)
        :param bool autorename: If there's a conflict, as determined by
            ``mode``, have the Dropbox server try to autorename the file to
            avoid conflict.
        :rtype dict:
        :return: Dictionary with the metadata of the uploaded file
        """

        from dropbox.files import WriteMode, FolderMetadata
        from dropbox.exceptions import ApiError

        dbx = self._get_instance()

        if file:
            file = os.path.abspath(os.path.expanduser(file))
            try:
                metadata = dbx.files_get_metadata(path)
                if isinstance(metadata, FolderMetadata):
                    path = '/'.join([path, os.path.basename(file)])
            except ApiError:
                pass

            with open(file, 'rb') as f:
                content = f.read()
        elif text:
            content = text.encode()
        else:
            raise SyntaxError('Please specify either a file or text to be uploaded')

        metadata = dbx.files_upload(content, path, autorename=autorename,
                                    mode=WriteMode.overwrite if overwrite else WriteMode.add)

        return self._parse_metadata(metadata)


# vim:sw=4:ts=4:et: