Dropbox Bot: Support sharing, searching and primary file operations.
This commit is contained in:
parent
47c6bbe787
commit
5b0a444ab8
41
zulip_bots/zulip_bots/bots/dropbox_share/doc.md
Normal file
41
zulip_bots/zulip_bots/bots/dropbox_share/doc.md
Normal file
|
@ -0,0 +1,41 @@
|
|||
# Dropbox Bot
|
||||
|
||||
This bot links your [dropbox](https://www.dropbox.com) account to [zulip](https://chat.zulip.org).
|
||||
|
||||
## Usage
|
||||
|
||||
- Create a dropbox app from [here](https://www.dropbox.com/developers/apps).
|
||||
- Click the `generate` button under the **Generate access token** section.
|
||||
- Copy the Access Token and paste it in a file named `dropbox_share.conf` as shown:
|
||||
```
|
||||
[dropbox_share]
|
||||
ACCESS_TOKEN=<your_access_token>
|
||||
```
|
||||
- Follow the instructions as described in [here](https://zulipchat.com/api/running-bots#running-a-bot).
|
||||
- Run the bot: `zulip-run-bot dropbox_share -b <Path/to/dropbox_share.conf> -c <Path/to/zuliprc>`
|
||||
|
||||
Use this bot with any of the following commands:
|
||||
|
||||
- `@dropbox mkdir` : Create a folder
|
||||
- `@dropbox ls` : List contents of a folder
|
||||
- `@dropbox write` : Save text to a file
|
||||
- `@dropbox rm` : Remove a file/folder
|
||||
- `@dropbox help` : See help text
|
||||
- `@dropbox read`: Read contents of a file
|
||||
- `@dropbox share`: Get a shareable link for a file/folder
|
||||
- `@dropbox search`: Search for matching file/folder names
|
||||
|
||||
where `dropbox` may be the name of the bot you registered in the zulip system.
|
||||
|
||||
### Usage examples
|
||||
|
||||
- `dropbox ls -` Shows files/folders in the root folder.
|
||||
- `dropbox mkdir foo` - Make folder named foo.
|
||||
- `dropbox ls foo/boo` - Shows the files/folders in foo/boo folder.
|
||||
- `dropbox write test hello world` - Write "hello world" to the file 'test'.
|
||||
- `dropbox rm test` - Remove the file/folder test.
|
||||
- `dropbox read foo` - Read the contents of file/folder foo.
|
||||
- `dropbox share foo` - Get shareable link for the file/folder foo.
|
||||
- `dropbox search boo` - Search for boo in root folder and get at max 20 results.
|
||||
- `dropbox search boo --mr 10` - Search for boo and get at max 10 results.
|
||||
- `dropbox search boo --fd foo` - Search for boo in folder foo.
|
|
@ -0,0 +1,2 @@
|
|||
[dropbox_share]
|
||||
ACCESS_TOKEN=<your_access_token>
|
227
zulip_bots/zulip_bots/bots/dropbox_share/dropbox_share.py
Normal file
227
zulip_bots/zulip_bots/bots/dropbox_share/dropbox_share.py
Normal file
|
@ -0,0 +1,227 @@
|
|||
from dropbox.dropbox import Dropbox
|
||||
from typing import Any, Dict, List, Tuple
|
||||
import re
|
||||
|
||||
URL = "[{name}](https://www.dropbox.com/home{path})"
|
||||
|
||||
class DropboxHandler(object):
|
||||
'''
|
||||
This bot allows you to easily share, search and upload files
|
||||
between zulip and your dropbox account.
|
||||
'''
|
||||
|
||||
def initialize(self, bot_handler: Any) -> None:
|
||||
self.config_info = bot_handler.get_config_info('dropbox_share')
|
||||
self.ACCESS_TOKEN = self.config_info.get('access_token')
|
||||
self.client = Dropbox(self.ACCESS_TOKEN)
|
||||
|
||||
def usage(self) -> str:
|
||||
return get_help()
|
||||
|
||||
def handle_message(self, message: Dict[str, str], bot_handler: Any) -> None:
|
||||
command = message['content']
|
||||
if command == "":
|
||||
command = "help"
|
||||
msg = dbx_command(self.client, command)
|
||||
bot_handler.send_reply(message, msg)
|
||||
|
||||
def get_help() -> str:
|
||||
return '''
|
||||
Example commands:
|
||||
|
||||
```
|
||||
@mention-bot usage: see usage examples
|
||||
@mention-bot mkdir: create a folder
|
||||
@mention-bot ls: list a folder
|
||||
@mention-bot write: write text
|
||||
@mention-bot rm: remove a file or folder
|
||||
@mention-bot read: read a file
|
||||
@mention-bot search: search a file/folder
|
||||
@mention-bot share: get a shareable link for the file/folder
|
||||
```
|
||||
'''
|
||||
|
||||
def get_usage_examples() -> str:
|
||||
return '''
|
||||
Usage:
|
||||
```
|
||||
@dropbox ls - Shows files/folders in the root folder.
|
||||
@dropbox mkdir foo - Make folder named foo.
|
||||
@dropbox ls foo/boo - Shows the files/folders in foo/boo folder.
|
||||
@dropbox write test hello world - Write "hello world" to the file 'test'.
|
||||
@dropbox rm test - Remove the file/folder test.
|
||||
@dropbox read foo - Read the contents of file/folder foo.
|
||||
@dropbox share foo - Get shareable link for the file/folder foo.
|
||||
@dropbox search boo - Search for boo in root folder and get at max 20 results.
|
||||
@dropbox search boo --mr 10 - Search for boo and get at max 10 results.
|
||||
@dropbox search boo --fd foo - Search for boo in folder foo.
|
||||
```
|
||||
'''
|
||||
|
||||
REGEXES = dict(
|
||||
command='(ls|mkdir|read|rm|write|search|usage|help)',
|
||||
path='(\S+)',
|
||||
optional_path='(\S*)',
|
||||
some_text='(.+?)',
|
||||
folder='?(?:--fd (\S+))?',
|
||||
max_results='?(?:--mr (\d+))?'
|
||||
)
|
||||
|
||||
def get_commands() -> Dict[str, Tuple[Any, List[str]]]:
|
||||
return {
|
||||
'help': (dbx_help, ['command']),
|
||||
'ls': (dbx_ls, ['optional_path']),
|
||||
'mkdir': (dbx_mkdir, ['path']),
|
||||
'rm': (dbx_rm, ['path']),
|
||||
'write': (dbx_write, ['path', 'some_text']),
|
||||
'read': (dbx_read, ['path']),
|
||||
'search': (dbx_search, ['some_text', 'folder', 'max_results']),
|
||||
'share': (dbx_share, ['path']),
|
||||
'usage': (dbx_usage, []),
|
||||
}
|
||||
|
||||
def dbx_command(client: Any, cmd: str) -> str:
|
||||
cmd = cmd.strip()
|
||||
if cmd == 'help':
|
||||
return get_help()
|
||||
cmd_name = cmd.split()[0]
|
||||
cmd_args = cmd[len(cmd_name):].strip()
|
||||
commands = get_commands()
|
||||
if cmd_name not in commands:
|
||||
return 'ERROR: unrecognized command\n' + get_help()
|
||||
f, arg_names = commands[cmd_name]
|
||||
partial_regexes = [REGEXES[a] for a in arg_names]
|
||||
regex = ' '.join(partial_regexes)
|
||||
regex += '$'
|
||||
m = re.match(regex, cmd_args)
|
||||
if m:
|
||||
return f(client, *m.groups())
|
||||
elif cmd_name == 'help':
|
||||
return get_help()
|
||||
else:
|
||||
return 'ERROR: ' + syntax_help(cmd_name)
|
||||
|
||||
def syntax_help(cmd_name: str) -> str:
|
||||
commands = get_commands()
|
||||
f, arg_names = commands[cmd_name]
|
||||
arg_syntax = ' '.join('<' + a + '>' for a in arg_names)
|
||||
if arg_syntax:
|
||||
cmd = cmd_name + ' ' + arg_syntax
|
||||
else:
|
||||
cmd = cmd_name
|
||||
return 'syntax: {}'.format(cmd)
|
||||
|
||||
def dbx_help(client: Any, cmd_name: str) -> str:
|
||||
return syntax_help(cmd_name)
|
||||
|
||||
def dbx_usage(client: Any) -> str:
|
||||
return get_usage_examples()
|
||||
|
||||
def dbx_mkdir(client: Any, fn: str) -> str:
|
||||
fn = '/' + fn # foo/boo -> /foo/boo
|
||||
try:
|
||||
result = client.files_create_folder(fn)
|
||||
msg = "CREATED FOLDER: " + URL.format(name=result.name, path=result.path_lower)
|
||||
except Exception:
|
||||
msg = "Please provide a correct folder path and name.\n"\
|
||||
"Usage: `mkdir <foldername>` to create a folder."
|
||||
|
||||
return msg
|
||||
|
||||
def dbx_ls(client: Any, fn: str) -> str:
|
||||
if fn != '':
|
||||
fn = '/' + fn
|
||||
|
||||
try:
|
||||
result = client.files_list_folder(fn)
|
||||
files_list = [] # type: List[str]
|
||||
for meta in result.entries:
|
||||
files_list += [" - " + URL.format(name=meta.name, path=meta.path_lower)]
|
||||
|
||||
msg = '\n'.join(files_list)
|
||||
if msg is '':
|
||||
msg = '`No files available`'
|
||||
|
||||
except Exception:
|
||||
msg = "Please provide a correct folder path\n"\
|
||||
"Usage: `ls <foldername>` to list folders in directory\n"\
|
||||
"or simply `ls` for listing folders in the root directory"
|
||||
|
||||
return msg
|
||||
|
||||
def dbx_rm(client: Any, fn: str) -> str:
|
||||
fn = '/' + fn
|
||||
|
||||
try:
|
||||
result = client.files_delete(fn)
|
||||
msg = "DELETED File/Folder : " + URL.format(name=result.name, path=result.path_lower)
|
||||
except Exception:
|
||||
msg = "Please provide a correct folder path and name.\n"\
|
||||
"Usage: `rm <foldername>` to delete a folder in root directory."
|
||||
return msg
|
||||
|
||||
def dbx_write(client: Any, fn: str, content: str) -> str:
|
||||
fn = '/' + fn
|
||||
|
||||
try:
|
||||
result = client.files_upload(content.encode(), fn)
|
||||
msg = "Written to file: " + URL.format(name=result.name, path=result.path_lower)
|
||||
except Exception:
|
||||
msg = "Incorrect file path or file already exists.\n"\
|
||||
"Usage: `write <filename> CONTENT`"
|
||||
|
||||
return msg
|
||||
|
||||
def dbx_read(client: Any, fn: str) -> str:
|
||||
fn = '/' + fn
|
||||
|
||||
try:
|
||||
result = client.files_download(fn)
|
||||
msg = "**{}** :\n{}".format(result[0].name, result[1].text)
|
||||
except Exception:
|
||||
msg = "Please provide a correct file path\n"\
|
||||
"Usage: `read <filename>` to read content of a file"
|
||||
|
||||
return msg
|
||||
|
||||
def dbx_search(client: Any, query: str, folder: str, max_results: str) -> str:
|
||||
if folder is None:
|
||||
folder = ''
|
||||
else:
|
||||
folder = '/' + folder
|
||||
if max_results is None:
|
||||
max_results = '20'
|
||||
try:
|
||||
result = client.files_search(folder, query, max_results=int(max_results))
|
||||
msg_list = []
|
||||
count = 0
|
||||
for entry in result.matches:
|
||||
file_info = entry.metadata
|
||||
count += 1
|
||||
msg_list += [" - " + URL.format(name=file_info.name, path=file_info.path_lower)]
|
||||
msg = '\n'.join(msg_list)
|
||||
|
||||
except Exception:
|
||||
msg = "Usage: `search <foldername> query --mr 10 --fd <folderName>`\n"\
|
||||
"Note:`--mr <int>` is optional and is used to specify maximun results.\n"\
|
||||
" `--fd <folderName>` to search in specific folder."
|
||||
|
||||
if msg == '':
|
||||
msg = "No files/folders found matching your query.\n"\
|
||||
"For file name searching, the last token is used for prefix matching"\
|
||||
" (i.e. “bat c” matches “bat cave” but not “batman car”)."
|
||||
|
||||
return msg
|
||||
|
||||
def dbx_share(client: Any, fn: str):
|
||||
fn = '/' + fn
|
||||
try:
|
||||
result = client.sharing_create_shared_link(fn)
|
||||
msg = result.url
|
||||
except Exception:
|
||||
msg = "Please provide a correct file name.\n"\
|
||||
"Usage: `share <filename>`"
|
||||
|
||||
return msg
|
||||
|
||||
handler_class = DropboxHandler
|
|
@ -0,0 +1 @@
|
|||
dropbox
|
129
zulip_bots/zulip_bots/bots/dropbox_share/test_dropbox_share.py
Normal file
129
zulip_bots/zulip_bots/bots/dropbox_share/test_dropbox_share.py
Normal file
|
@ -0,0 +1,129 @@
|
|||
from zulip_bots.test_lib import BotTestCase
|
||||
from typing import List
|
||||
from unittest.mock import patch
|
||||
|
||||
from zulip_bots.bots.dropbox_share.test_util import (
|
||||
MockFileMetadata,
|
||||
MockListFolderResult,
|
||||
MockSearchMatch,
|
||||
MockSearchResult,
|
||||
MockPathLinkMetadata,
|
||||
MockHttpResponse
|
||||
)
|
||||
|
||||
def get_files_list(*args, **kwargs):
|
||||
return MockListFolderResult(
|
||||
entries = [
|
||||
MockFileMetadata('foo', '/foo'),
|
||||
MockFileMetadata('boo', '/boo')
|
||||
],
|
||||
has_more = False
|
||||
)
|
||||
|
||||
def create_file(*args, **kwargs):
|
||||
return MockFileMetadata('foo', '/foo')
|
||||
|
||||
def download_file(*args, **kwargs):
|
||||
return [MockFileMetadata('foo', '/foo'), MockHttpResponse('boo')]
|
||||
|
||||
def search_files(*args, **kwargs):
|
||||
return MockSearchResult([
|
||||
MockSearchMatch(
|
||||
MockFileMetadata('foo', '/foo')
|
||||
),
|
||||
MockSearchMatch(
|
||||
MockFileMetadata('fooboo', '/fooboo')
|
||||
)
|
||||
])
|
||||
|
||||
def get_shared_link(*args, **kwargs):
|
||||
return MockPathLinkMetadata('http://www.foo.com/boo')
|
||||
|
||||
def get_help() -> str:
|
||||
return '''
|
||||
Example commands:
|
||||
|
||||
```
|
||||
@mention-bot usage: see usage examples
|
||||
@mention-bot mkdir: create a folder
|
||||
@mention-bot ls: list a folder
|
||||
@mention-bot write: write text
|
||||
@mention-bot rm: remove a file or folder
|
||||
@mention-bot read: read a file
|
||||
@mention-bot search: search a file/folder
|
||||
@mention-bot share: get a shareable link for the file/folder
|
||||
```
|
||||
'''
|
||||
|
||||
class TestDropboxBot(BotTestCase):
|
||||
bot_name = "dropbox_share"
|
||||
config_info = {"access_token": "1234567890"}
|
||||
|
||||
def test_bot_responds_to_empty_message(self):
|
||||
with self.mock_config_info(self.config_info):
|
||||
self.verify_reply('', get_help())
|
||||
self.verify_reply('help', get_help())
|
||||
|
||||
def test_dbx_ls(self):
|
||||
bot_response = " - [foo](https://www.dropbox.com/home/foo)\n"\
|
||||
" - [boo](https://www.dropbox.com/home/boo)"
|
||||
with patch('dropbox.dropbox.Dropbox.files_list_folder', side_effect=get_files_list), \
|
||||
self.mock_config_info(self.config_info):
|
||||
self.verify_reply("ls", bot_response)
|
||||
|
||||
def test_dbx_mkdir(self):
|
||||
bot_response = "CREATED FOLDER: [foo](https://www.dropbox.com/home/foo)"
|
||||
with patch('dropbox.dropbox.Dropbox.files_create_folder', side_effect=create_file), \
|
||||
self.mock_config_info(self.config_info):
|
||||
self.verify_reply('mkdir foo', bot_response)
|
||||
|
||||
def test_dbx_rm(self):
|
||||
bot_response = "DELETED File/Folder : [foo](https://www.dropbox.com/home/foo)"
|
||||
with patch('dropbox.dropbox.Dropbox.files_delete', side_effect=create_file), \
|
||||
self.mock_config_info(self.config_info):
|
||||
self.verify_reply('rm foo', bot_response)
|
||||
|
||||
def test_dbx_write(self):
|
||||
bot_response = "Written to file: [foo](https://www.dropbox.com/home/foo)"
|
||||
with patch('dropbox.dropbox.Dropbox.files_upload', side_effect=create_file), \
|
||||
self.mock_config_info(self.config_info):
|
||||
self.verify_reply('write foo boo', bot_response)
|
||||
|
||||
def test_dbx_read(self):
|
||||
bot_response = "**foo** :\nboo"
|
||||
with patch('dropbox.dropbox.Dropbox.files_download', side_effect=download_file), \
|
||||
self.mock_config_info(self.config_info):
|
||||
self.verify_reply('read foo', bot_response)
|
||||
|
||||
def test_dbx_search(self):
|
||||
bot_response = " - [foo](https://www.dropbox.com/home/foo)\n"\
|
||||
" - [fooboo](https://www.dropbox.com/home/fooboo)"
|
||||
with patch('dropbox.dropbox.Dropbox.files_search', side_effect=search_files), \
|
||||
self.mock_config_info(self.config_info):
|
||||
self.verify_reply('search foo', bot_response)
|
||||
|
||||
def test_dbx_share(self):
|
||||
bot_response = 'http://www.foo.com/boo'
|
||||
with patch('dropbox.dropbox.Dropbox.sharing_create_shared_link', side_effect=get_shared_link), \
|
||||
self.mock_config_info(self.config_info):
|
||||
self.verify_reply('share boo', bot_response)
|
||||
|
||||
def test_invalid_commands(self):
|
||||
ls_error_response = "ERROR: syntax: ls <optional_path>"
|
||||
mkdir_error_response = "ERROR: syntax: mkdir <path>"
|
||||
rm_error_response = "ERROR: syntax: rm <path>"
|
||||
write_error_response = "ERROR: syntax: write <path> <some_text>"
|
||||
search_error_response = "ERROR: syntax: search <path> <some_text> <max_results>"
|
||||
share_error_response = "ERROR: syntax: share <path>"
|
||||
|
||||
with self.mock_config_info(self.config_info):
|
||||
# ls
|
||||
self.verify_reply("ls foo boo", ls_error_response)
|
||||
# mkdir
|
||||
self.verify_reply("mkdir foo boo", mkdir_error_response)
|
||||
# rm
|
||||
self.verify_reply("rm foo boo", rm_error_response)
|
||||
# write
|
||||
self.verify_reply("write foo", write_error_response)
|
||||
# share
|
||||
self.verify_reply("share foo boo", share_error_response)
|
27
zulip_bots/zulip_bots/bots/dropbox_share/test_util.py
Normal file
27
zulip_bots/zulip_bots/bots/dropbox_share/test_util.py
Normal file
|
@ -0,0 +1,27 @@
|
|||
from typing import List
|
||||
|
||||
class MockFileMetadata:
|
||||
def __init__(self, name: str, path_lower: str):
|
||||
self.name = name
|
||||
self.path_lower = path_lower
|
||||
|
||||
class MockListFolderResult:
|
||||
def __init__(self, entries: str, has_more: str):
|
||||
self.entries = entries
|
||||
self.has_more = has_more
|
||||
|
||||
class MockSearchMatch:
|
||||
def __init__(self, metadata: List[MockFileMetadata]):
|
||||
self.metadata = metadata
|
||||
|
||||
class MockSearchResult:
|
||||
def __init__(self, matches: List[MockSearchMatch]):
|
||||
self.matches = matches
|
||||
|
||||
class MockPathLinkMetadata:
|
||||
def __init__(self, url: str):
|
||||
self.url = url
|
||||
|
||||
class MockHttpResponse:
|
||||
def __init__(self, text: str):
|
||||
self.text = text
|
Loading…
Reference in a new issue