__init__.py 2.6 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273
  1. import requests
  2. import logging
  3. import json
  4. from datetime import datetime, timedelta
  5. from deepmerge import always_merger
  6. from pprint import pprint
  7. BASE = 'https://api.dropbox.com/2'
  8. class Dropbox:
  9. def __init__(self, token, proxy=None):
  10. self.token = token
  11. if proxy:
  12. self.proxies = {'http': "socks5://{}".format(proxy)}
  13. else:
  14. self.proxies = {}
  15. info = self.get_info()
  16. if info:
  17. self.name = info["name"]
  18. self.team_id = info["team_id"]
  19. self.num_users = info["num_provisioned_users"]
  20. else:
  21. logging.error("Wrong token")
  22. def call(self, endpoint, args=None, member=None, nested=False, cursor=True, base=BASE, params=None):
  23. if params:
  24. params = {"arg": json.dumps(params)}
  25. r = requests.post('{}/{}'.format(base, endpoint), headers={"Authorization": "Bearer {}".format(self.token), "Dropbox-API-Select-User": member}, json=args, proxies=self.proxies, params=params)
  26. if r.status_code == 200:
  27. if 'json' not in r.headers['content-type']:
  28. return r.content
  29. data = r.json()
  30. result = data
  31. while nested == False and "has_more" in data and data["has_more"] == True and cursor == True:
  32. args = {"cursor": data["cursor"]}
  33. data = self.call("{}/continue".format(endpoint), args, member, True)
  34. result = always_merger.merge(result, data)
  35. return result
  36. else:
  37. logging.debug(r.text)
  38. print(r.text)
  39. return False
  40. def get_info(self):
  41. return self.call("team/get_info")
  42. def members_list(self):
  43. return self.call("team/members/list")
  44. def list_member_devices(self, team_member_id):
  45. 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})
  46. def get_current_account(self, team_member_id):
  47. return self.call("users/get_current_account", None, team_member_id)
  48. def list_folder(self, team_member_id, path, recursive=False):
  49. return self.call("files/list_folder", {"path": path, "recursive": recursive}, team_member_id)
  50. def list_namespaces(self):
  51. return self.call("team/namespaces/list")
  52. def get_events(self, team_member_id):
  53. end = datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ")
  54. start = (datetime.now() - timedelta(days=10)).strftime("%Y-%m-%dT%H:%M:%SZ")
  55. return self.call("team_log/get_events", {"limit": 100, "account_id": team_member_id, "time": {"start_time": start, "end_time": end}})
  56. def download(self, team_member_id, path):
  57. return self.call("files/download", params={"path": path}, member=team_member_id, base="https://content.dropboxapi.com/2")
  58. def get_metadata(self, team_member_id, path):
  59. return self.call("files/get_metadata", {"path": path}, member=team_member_id)