Dropbox authentication and upload

pi-to-dropbox.py

#!/bin/python3

# Copied from https://bit.ly/4aZvn5B

###############################################################################
#        Script uploading files to Dropbox and receiving Dropbox links
#        -------------------------------------------------------------
# Every target file/folder is passed as argument to the script.
# Two preparation steps are needed:
#   1. Register a Dropbox application and put the application key as value of
#      APPLICATION_KEY global variable.
#   2. Register the used REDIRECT_URI in the application just created in
#      previous step. With host 'localhost' and port 8080, the redirect URL
#      that has to be registered is "http://localhost:8080/", for instance.
# Next, just make it executable (if needed), using (on Mac and Linux):
#   $ chmod a+x dropbox_file
# ... and put it in a place visible for execution in your system (somewhere in
# folders pointed by $PATH environment variable). On first run you will be
# invited to link your script to your Dropbox account. When getting links to
# work correct local Dropbox application and this script have to be link to the
# same account!
# For more help type:
#   $ dropbox_file --help
# Author: Здравко
#   www.dropboxforum.com/t5/user/viewprofilepage/user-id/422790
###############################################################################

from dropbox import Dropbox
from dropbox.exceptions import ApiError
from dropbox.files import WriteMode
import json
from pathlib import Path
from datetime import datetime
from os import sep
from sys import exit
import logging
from platformdirs import user_config_path
import click

# Place to save current configuration
CONFIG_JSON=user_config_path('dropbox_file') / 'cred.json'

# Take a look on your application in https://www.dropbox.com/developers/apps
APPLICATION_KEY='PUT YOUR KEY HERE'

URI_HOST='localhost'
URI_PORT=8080
# URI should be registered in the application redirect URIs list!!!
REDIRECT_URI=f"http://{URI_HOST}:{URI_PORT}/"
HOST_PORT=(URI_HOST,URI_PORT)

success_response = (
  "End of authentication flow.  You're going to do your work!")
cancel_response = "🤷 You have denied your application's work. "
error_response = " You got an error: "

class ApplicationConfig:
  def __init__(self, conf_path=CONFIG_JSON):
    self.conf_path=conf_path
    self.conf=None
    self.client=None
    self.access_token=None
    self.access_token_expiresat=None
    self.refresh_token=None
    if self.conf_path.is_file():
      try:
        with self.conf_path.open() as fconf:
          self.conf=json.load(fconf)
        self.access_token = self.conf['access_token']
        self.access_token_expiresat = datetime.fromtimestamp(
          self.conf['access_token_expiresat'])
        self.refresh_token = self.conf['refresh_token']
      except Exception:
        self.conf_path.unlink(True)
        self.conf=None
    else:
      conf_path.parent.mkdir(exist_ok=True)
  def __del__(self):
    "Checks for something changed (new access token) and dumps it when there is"
    if (self.client is not None and
        self.client._oauth2_access_token_expiration >
        self.access_token_expiresat):
      self.conf['access_token'] = self.client._oauth2_access_token
      self.conf['access_token_expiresat'] = (
        self.client._oauth2_access_token_expiration.timestamp())
      self.conf['refresh_token'] = self.client._oauth2_refresh_token
      with self.conf_path.open(mode='w') as fconf:
        json.dump(self.conf, fconf)
  def getClient(self):
    "Gets Dropbox client object. Performs OAuth flow if needed."
    if self.conf is None:
      self.client=None
      import webbrowser
      from dropbox import DropboxOAuth2Flow
      from dropbox.oauth import NotApprovedException
      from http.server import HTTPServer, BaseHTTPRequestHandler
      dbxAuth=DropboxOAuth2Flow(APPLICATION_KEY, REDIRECT_URI, {},
        'dropbox-auth-csrf-token', token_access_type='offline', use_pkce=True)
      webbrowser.open(dbxAuth.start())
      conf=None
      conf_path = self.conf_path
      class Handler(BaseHTTPRequestHandler):
        response_success = success_response.encode()
        response_cancel = cancel_response.encode()
        response_error = error_response.encode()
        def do_GET(self):
          nonlocal dbxAuth, conf
          from urllib.parse import urlparse, parse_qs
          query = parse_qs(urlparse(self.path).query)
          for r in query.keys():
            query[r] = query[r][0]
          self.send_response(200)
          self.send_header("content-type", "text/plain;charset=UTF-8")
          try:
            oauthRes = dbxAuth.finish(query)
            conf={'access_token': oauthRes.access_token,
                  'access_token_expiresat': oauthRes.expires_at.timestamp(),
                  'refresh_token': oauthRes.refresh_token}
            with conf_path.open(mode='w') as fconf:
              json.dump(conf, fconf)
          except NotApprovedException:
            conf={}
            self.send_header("content-length",
                             f"{len(Handler.response_cancel)}")
            self.end_headers()
            self.wfile.write(Handler.response_cancel)
            self.wfile.flush()
            return
          except Exception as e:
            conf={}
            r = Handler.response_error + str(e).encode()
            self.send_header("content-length", f"{len(r)}")
            self.end_headers()
            self.wfile.write(r)
            self.wfile.flush()
            return
          self.send_header("content-length",
                           f"{len(Handler.response_success)}")
          self.end_headers()
          self.wfile.write(Handler.response_success)
          self.wfile.flush()
      httpd=HTTPServer((URI_HOST, URI_PORT), Handler)
      while conf is None:
        httpd.handle_request()
      httpd.server_close()
      del httpd
      if 'refresh_token' not in conf:
        raise RuntimeError("Cannot process because missing authentication")
      self.conf = conf
      self.access_token = self.conf['access_token']
      self.access_token_expiresat = datetime.fromtimestamp(
        self.conf['access_token_expiresat'])
      self.refresh_token = self.conf['refresh_token']
    # Makes sure there is cached client object.
    if self.client is None:
      self.client=Dropbox(self.access_token,
                    oauth2_refresh_token=self.refresh_token,
                    oauth2_access_token_expiration=self.access_token_expiresat,
                    app_key=APPLICATION_KEY)
    return self.client

class PathMapper:
  def __init__(self, local=True):
    if not local:
      self.dbx_path = ''
      return
    dbx_info = Path('~/.dropbox/info.json').expanduser()
    if not dbx_info.is_file():
      raise RuntimeError("Missing Dropbox application information")
    with dbx_info.open() as finfo:
      # Only personal accounts are supported by now - group accounts need
      # additional namespace handling (just changing 'personal' is not enough).
      # Somebody else may make some exercises. 
      self.dbx_path = json.load(finfo)['personal']['path']
  def __contains__(self, path):
    path = str(Path(path).expanduser().absolute())
    return ((len(path) == len(self.dbx_path) and path == self.dbx_path) or
            (len(path) > len(self.dbx_path) and path[len(self.dbx_path)] == sep
             and path[:len(self.dbx_path)] == self.dbx_path))
  def __getitem__(self, path):
    path = str(Path(path).expanduser().absolute())
    if ((len(path) == len(self.dbx_path) and path == self.dbx_path) or
        (len(path) > len(self.dbx_path) and path[len(self.dbx_path)] == sep
         and path[:len(self.dbx_path)] == self.dbx_path)):
      return path[len(self.dbx_path):]

pass_config = click.make_pass_decorator(ApplicationConfig, ensure=True)

@click.group()
@click.option('-v', '--verbose', is_flag=True, help="Toggle verbose mode.")
def main(verbose):
  """Perform file actions on files in Dropbox account. Type:
  
  $ dropbox_file COMMAND --help
  
  ... where COMMAND is one of listed below, for command specific options.
  """
  if verbose:
    logging.basicConfig(level=logging.DEBUG)

@main.command()
@click.option('-l', '--local', is_flag=True, help="Try parce path(s) as local "
              "- residing in application's dir (Mac and Linux only).")
@click.argument('paths', type=click.Path(), nargs=-1)
@pass_config
def get_link(conf, paths, local):
  """Takes link(s) to particular file(s) in Dropbox account.
  
  By default paths are passed as rooted to Dropbox account home.
  
  Returns preview URLs (line per file/folder).
  """
  dbxPathMap = PathMapper(local)
  dbx = conf.getClient()
  
  for path in paths:
    if path not in dbxPathMap:
      logging.error(
        f"Passed path '{path}' is not part of the Dropbox driectory tree")
      continue
    dbx_path = dbxPathMap[path]
    if len(dbx_path) == 0:
      logging.error("Dropbox folder itself cannot be pointed by link")
      continue
    logging.debug(f"Processing {'local' if local else 'remote'} file '{path}'"
                  f" with Dropbox path '{dbx_path}'")
    try:
      metadata = dbx.sharing_create_shared_link_with_settings(dbx_path)
    except ApiError as e:
      er = e.error
      if not er.is_shared_link_already_exists():
        raise
      er = er.get_shared_link_already_exists()
      if not er.is_metadata():
        raise
      metadata = er.get_metadata()
    print(metadata.url)

@main.command()
# Can be improved to handle dirs in recursion, but... some other time.
@click.argument('src', type=click.Path(exists=True, dir_okay=False), nargs=-1)
@click.argument('dest', type=click.Path(), nargs=1)
@pass_config
def upload(conf, dest, src):
  """Uploads passed local file(s) to designated place in Dropbox account.
  
  The last argument is the designated place/folder in your Dropbox account.
  This place represent always a target folder. Can be uploaded one or more
  files represented by all arguments but the last one.
  
  Returns associated ID (line per file).
  """
  dbx = conf.getClient()
  dest = Path(dest)
  
  for path in src:
    path = Path(path)
    d = "".join('/'+d for d in dest.joinpath(path.name).parts if d != sep)
    logging.debug(f"Uploading local file '{path}' to Dropbox path '{d}'")
    # Handlig of upload sessions would be nice here for bigger files support
    # and batch upload, but... again some other time.
    with path.open('rb') as f:
      metadata = dbx.files_upload(f.read(), d, WriteMode('overwrite'))
    print(metadata.id)

if __name__ == "__main__":
  try:
    main()
  except Exception as e:
    logging.error(f"Unexpected error: {e}")
    exit(1)