Spaces:
Runtime error
Runtime error
| import datetime,logging | |
| import logging.config | |
| import time | |
| import dropbox | |
| from dropbox.files import WriteMode | |
| from dropbox.exceptions import ApiError, AuthError | |
| import sys,os,yaml | |
| import baseInfra.firebase_handler as fbh | |
| # Load the config file | |
| with open('logging_config.yaml', 'rt') as f: | |
| config = yaml.safe_load(f.read()) | |
| logging.config.dictConfig(config) | |
| TOKEN=fbh.fb_get("d2_accesstoken") | |
| APP_KEY=os.environ['DROPBOX_APP_KEY'] | |
| APP_SECRET=os.environ['DROPBOX_APP_SECRET'] | |
| REFRESH_TOKEN=fbh.fb_get("d2_refreshtoken") | |
| #os.environ['DROP_DIR2']="C:/dockers/chroma/chroma1/" | |
| #os.environ['APP_PATH']="/" | |
| #print("token::",TOKEN) | |
| with dropbox.Dropbox(oauth2_access_token=TOKEN,app_key=APP_KEY,app_secret=APP_SECRET,oauth2_refresh_token=REFRESH_TOKEN) as dbx: #,app_key=APP_KEY,app_secret=APP_SECRET,oauth2_refresh_token=REFRESH_TOKEN) as dbx: | |
| # Check that the access token is valid | |
| try: | |
| dbx.users_get_current_account() | |
| if (TOKEN != dbx._oauth2_access_token): | |
| fbh.fb_update("d2_accesstoken",dbx._oauth2_access_token) | |
| TOKEN=dbx._oauth2_access_token | |
| print("dropbox connection ok,",dbx._oauth2_access_token) | |
| print(dbx._oauth2_refresh_token) | |
| except AuthError: | |
| try: | |
| dbx.check_and_refresh_access_token() | |
| fbh.fb_update("d2_accesstoken",dbx._oauth2_access_token) | |
| print("dropbox connection refreshed and updated",dbx._oauth2_access_token) | |
| print(dbx._oauth2_refresh_token) | |
| except Exception: | |
| sys.exit("ERROR: Invalid access token; try re-generating an " | |
| "access token from the app console on the web.") | |
| def normalizeFilename(filename): | |
| while '//' in filename: | |
| filename = filename.replace('//', '/') | |
| return filename | |
| def getDropboxFilename(localFilename): | |
| """ localFilename is $DROP_DIR2/<subpath>/<filename>""" | |
| """ dropboxFilename is $APP_PATH/<subpath>/<filename""" | |
| #if not localFilename.startswith(os.environ['DROP_DIR2']): | |
| # localFilename=os.environ['DROP_DIR2']+localFilename | |
| localFilename=normalizeFilename(localFilename) | |
| return normalizeFilename(localFilename.replace(os.environ['DROP_DIR2'],"/",1).replace("/",os.environ['APP_PATH'],1)) | |
| def getLocalFilename(dropboxFilename): | |
| """ localFilename is $DROP_DIR2/<subpath>/<filename>""" | |
| """ dropboxFilename is $APP_PATH/<subpath>/<filename""" | |
| #if not dropboxFilename.startswith(os.environ['APP_PATH']): | |
| # dropboxFilename=os.environ['APP_PATH']+dropboxFilename | |
| dropboxFilename=normalizeFilename(dropboxFilename) | |
| return normalizeFilename(dropboxFilename.replace(os.environ['APP_PATH'],"/",1).replace("/",os.environ['DROP_DIR2'],1)) | |
| async def backupFile(localFilename): | |
| """Upload a file. | |
| Return the request response, or None in case of error. | |
| This will also create directory on dropbox if needed | |
| """ | |
| global TOKEN | |
| if not localFilename.startswith(os.environ['DROP_DIR2']): | |
| localFilename=os.environ['DROP_DIR2']+localFilename | |
| localFilename=normalizeFilename(localFilename) | |
| dropboxFilename=getDropboxFilename(localFilename) | |
| print("backing file ",localFilename," to ",dropboxFilename) | |
| mode = dropbox.files.WriteMode.overwrite | |
| mtime = os.path.getmtime(localFilename) | |
| with open(localFilename, 'rb') as f: | |
| data = f.read() | |
| try: | |
| res = dbx.files_upload( | |
| data, dropboxFilename, mode, | |
| client_modified=datetime.datetime(*time.gmtime(mtime)[:6]), | |
| mute=True) | |
| if (TOKEN != dbx._oauth2_access_token): | |
| fbh.fb_update("d2_accesstoken",dbx._oauth2_access_token) | |
| TOKEN=dbx._oauth2_access_token | |
| print(dbx._oauth2_refresh_token) | |
| except dropbox.exceptions.ApiError as err: | |
| print('*** API error', err) | |
| return None | |
| print('uploaded as', res.name.encode('utf8')) | |
| return res | |
| def restoreFile(dropboxFilename): | |
| """Download a file. | |
| Return the bytes of the file, or None if it doesn't exist. | |
| Will create dir+subdirs if possible | |
| """ | |
| global TOKEN | |
| dropboxFilename=normalizeFilename(dropboxFilename) | |
| localFilename=getLocalFilename(dropboxFilename) | |
| print("restoring file ",localFilename," from ",dropboxFilename) | |
| try: | |
| md, res = dbx.files_download(dropboxFilename) | |
| if (TOKEN != dbx._oauth2_access_token): | |
| fbh.fb_update("d2_accesstoken",dbx._oauth2_access_token) | |
| TOKEN=dbx._oauth2_access_token | |
| print(dbx._oauth2_refresh_token) | |
| except dropbox.exceptions.HttpError as err: | |
| print('*** HTTP error', err) | |
| return None | |
| data = res.content | |
| print(len(data), 'bytes; md:', md) | |
| localdir=os.path.dirname(localFilename) | |
| if not os.path.exists(localdir): | |
| os.makedirs(localdir) | |
| with open(localFilename, 'wb') as f: | |
| f.write(data) | |
| return data | |
| async def backupFolder(localFolder): | |
| """ list all files in folder and subfolder and upload them""" | |
| print("backup folder called for ",localFolder) | |
| if not localFolder.startswith(os.environ['DROP_DIR2']): | |
| localFolder=os.environ['DROP_DIR2']+localFolder | |
| print("Local folder is ",localFolder) | |
| filenames=[] | |
| for (root,dirs,files) in os.walk(localFolder, topdown=True): | |
| print(root) | |
| for filename in files: | |
| filenames.append(root+"/"+filename) | |
| print(root+"/"+filename) | |
| await backupFile(root+"/"+filename) | |
| async def restoreFolder(dropboxFolder): | |
| """ list all files in dropbox folder and subfolders and restore them""" | |
| global TOKEN | |
| if not dropboxFolder.startswith(os.environ['APP_PATH']): | |
| dropboxFolder=os.environ['APP_PATH']+dropboxFolder | |
| try: | |
| res=dbx.files_list_folder(dropboxFolder) | |
| if (TOKEN != dbx._oauth2_access_token): | |
| fbh.fb_update("d2_accesstoken",dbx._oauth2_access_token) | |
| TOKEN=dbx._oauth2_access_token | |
| print(dbx._oauth2_refresh_token) | |
| except dropbox.exceptions.ApiError as err: | |
| print('Folder listing failed for', dropboxFolder, '-- assumed empty:', err) | |
| return | |
| except dropbox.exceptions.AuthError as err1: | |
| print('Folder listing failed for', dropboxFolder, '-- assumed empty:', err1) | |
| return | |
| for entry in res.entries: | |
| if (isinstance(entry, dropbox.files.FileMetadata)): | |
| restoreFile(entry.path_display) | |
| else: | |
| try: | |
| restoreFolder(entry.path_display) | |
| except Exception: | |
| print("Error restoring folder,",entry.path_display) | |
| print(entry.path_display) | |