74 lines
2.6 KiB
Python
74 lines
2.6 KiB
Python
|
import requests
|
||
|
|
import logging
|
||
|
|
import json
|
||
|
|
from datetime import datetime, timedelta
|
||
|
|
from deepmerge import always_merger
|
||
|
|
from pprint import pprint
|
||
|
|
|
||
|
|
BASE = 'https://api.dropbox.com/2'
|
||
|
|
|
||
|
|
class Dropbox:
|
||
|
|
def __init__(self, token, proxy=None):
|
||
|
|
self.token = token
|
||
|
|
if proxy:
|
||
|
|
self.proxies = {'http': "socks5://{}".format(proxy)}
|
||
|
|
else:
|
||
|
|
self.proxies = {}
|
||
|
|
info = self.get_info()
|
||
|
|
if info:
|
||
|
|
self.name = info["name"]
|
||
|
|
self.team_id = info["team_id"]
|
||
|
|
self.num_users = info["num_provisioned_users"]
|
||
|
|
else:
|
||
|
|
logging.error("Wrong token")
|
||
|
|
|
||
|
|
|
||
|
|
def call(self, endpoint, args=None, member=None, nested=False, cursor=True, base=BASE, params=None):
|
||
|
|
if params:
|
||
|
|
params = {"arg": json.dumps(params)}
|
||
|
|
r = requests.post('{}/{}'.format(base, endpoint), headers={"Authorization": "Bearer {}".format(self.token), "Dropbox-API-Select-User": member}, json=args, proxies=self.proxies, params=params)
|
||
|
|
if r.status_code == 200:
|
||
|
|
if 'json' not in r.headers['content-type']:
|
||
|
|
return r.content
|
||
|
|
data = r.json()
|
||
|
|
result = data
|
||
|
|
while nested == False and "has_more" in data and data["has_more"] == True and cursor == True:
|
||
|
|
args = {"cursor": data["cursor"]}
|
||
|
|
data = self.call("{}/continue".format(endpoint), args, member, True)
|
||
|
|
result = always_merger.merge(result, data)
|
||
|
|
return result
|
||
|
|
else:
|
||
|
|
logging.debug(r.text)
|
||
|
|
print(r.text)
|
||
|
|
return False
|
||
|
|
|
||
|
|
def get_info(self):
|
||
|
|
return self.call("team/get_info")
|
||
|
|
|
||
|
|
def members_list(self):
|
||
|
|
return self.call("team/members/list")
|
||
|
|
|
||
|
|
def list_member_devices(self, team_member_id):
|
||
|
|
return self.call("team/devices/list_member_devices", {"team_member_id": team_member_id, "include_web_sessions": True, "include_desktop_clients": True, "include_mobile_clients": True})
|
||
|
|
|
||
|
|
def get_current_account(self, team_member_id):
|
||
|
|
return self.call("users/get_current_account", None, team_member_id)
|
||
|
|
|
||
|
|
def list_folder(self, team_member_id, path, recursive=False):
|
||
|
|
return self.call("files/list_folder", {"path": path, "recursive": recursive}, team_member_id)
|
||
|
|
|
||
|
|
def list_namespaces(self):
|
||
|
|
return self.call("team/namespaces/list")
|
||
|
|
|
||
|
|
def get_events(self, team_member_id):
|
||
|
|
end = datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ")
|
||
|
|
start = (datetime.now() - timedelta(days=10)).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||
|
|
return self.call("team_log/get_events", {"limit": 100, "account_id": team_member_id, "time": {"start_time": start, "end_time": end}})
|
||
|
|
|
||
|
|
def download(self, team_member_id, path):
|
||
|
|
return self.call("files/download", params={"path": path}, member=team_member_id, base="https://content.dropboxapi.com/2")
|
||
|
|
|
||
|
|
def get_metadata(self, team_member_id, path):
|
||
|
|
return self.call("files/get_metadata", {"path": path}, member=team_member_id)
|
||
|
|
|