File: android-deltas-sync/common.py
""" ======================================================================= Code used by scripts run on both PC and phone. See _README.html for license, attribution, version, and docs. Dev notes: 1) This file is loaded by import, and expects callers to immediately run its startup(), passing in a config_{pc, phone} module which has global settings copied and used here. This is due to its Bash heritage; Py scripts here are more-or-less direct translations for portability. 2) In config_*.py files, set PauseSteps=n to avoid having to press Enter here, and set CheckFilenames=n to avoid name-fixer interactions here. 3) You can also change ZIP here to point to a newer installed ziptools, but don't generally need to; the version in Mergeall-source/ suffices. ======================================================================= """ import os, time, sys, shutil, subprocess, errno from os.path import join VERSION = 1.2 APPNAME = 'Android Deltas Sync' # renamed Oct22 [1.2] # Linux also means WSL on Windows, and Android; # Cygwin is Windows, but using its own Python RunningOnWindows = sys.platform.startswith('win') # may be run by Cygwin RunningOnMacOS = sys.platform.startswith('darwin') RunningOnLinux = sys.platform.startswith('linux') # includes WSL, Android RunningOnCygwin = sys.platform.startswith('cygwin') # Cygwin's own python RunningOnAndroid = any(key for key in os.environ if key.startswith('ANDROID_')) # common indent width indent = ' ' * 4 # catch control-c in input() and exit nicely builtin_input = input def input(prompt=''): try: return builtin_input(prompt + ' ') except KeyboardInterrupt: print('\nRun aborted by control-c at prompt.') shutdown(bad=True) import builtins builtins.input = input # default to version here everywhere #---------------------------------------------------------------------- # Setup: globals, logs dir, etc. #---------------------------------------------------------------------- def startup(config, self): """ ------------------------------------------------------------------- Common startup code, called with either the PC or phone config module, and the module containing this function. This copies the config module's names into this module's namespace, and adds two more names here--and all of these names become globals both here and in the client scripts. Hackish, maybe, but globals are denoted by all uppercase (mostly), and this is better than the original Bash translation, which ran this file's all-top-level code in-line in clients' namespaces to emulate a Bash 'source <file>': from config_pc import * mydir = os.path.dirname(__file__) exec(open(join(mydir, 'common.py')).read()) This worked until the post hook script did the same thing and broke outputs. Some globals may be avoided by passing args to utilities, but this module's namespace avoids extra args (~6 in some cases). ------------------------------------------------------------------- """ global ZIP, DATE # set here! print('%s %.1f' % (APPNAME, VERSION)) # copy pc-or-phone config settings to my namespace for name in dir(config): if not name.startswith('__'): setattr(self, name, getattr(config, name)) # check that main common vars are set before tweaking for config in ('LOGS', 'MALL', 'FROM', 'TO', 'STUFF'): try: getattr(self, config) except: print('Config "%s" is not set' % config) print('Exiting; please check the config file and rerun.') shutdown(bad=True) # expand user home (~, ~user) and env vars ($var, ${var}) in paths expander = lambda path: os.path.expandvars(os.path.expanduser(path)) for config in ('LOGS', 'MALL', 'FROM', 'TO'): setattr(self, config, expander(getattr(self, config))) # path checks in all scripts; STUFF is also checked in some scripts for config in ('MALL', 'FROM', 'TO'): setting = getattr(self, config) if not os.path.isdir(setting): print('The %s path does not exist: "%s"' % (config, setting)) print('Exiting; please check the config file and rerun.') shutdown(bad=True) # ensure logfiles folder in all scripts if not os.path.isdir(LOGS): if os.path.exists(LOGS): os.remove(LOGS) os.mkdir(LOGS) # fix Windows non-ASCII prints when 'script > file' (cygwin too?) if RunningOnWindows: os.environ['PYTHONIOENCODING'] = 'utf8' # subprocs inherit # ziptools source-code folder (included in Mergeall) ZIP = join(MALL, 'test', 'ziptools') # logfile date stamp now = time.localtime() DATE = time.strftime('%y-%m-%d', now) # e.g., 21-09-30 for Sep 30, 2021 (sortable) #---------------------------------------------------------------------- # Utilities #---------------------------------------------------------------------- def announce(steplabel): """ ------------------------------------------------------------------- A conditional start-of-step display: pause for enter if enabled. ------------------------------------------------------------------- """ print('\nSTART:', steplabel) if PauseSteps: try: builtin_input('Press enter/return to continue') except KeyboardInterrupt: print('\nRun aborted by control-c before step.') shutdown() def opener(message): """ ------------------------------------------------------------------- Display opening message, save run start time for use in closer(). Package name+version are shown in startup(), before error tests. ------------------------------------------------------------------- """ global RunStartTime RunStartTime = time.perf_counter() # relative seconds print('\n' + message) def closer(message): """ ------------------------------------------------------------------- Normal exit: show total elapsed time (all steps + user time), and pause for an Enter press to allow output to be read if a PC-side script was run on Windows by an icon click (see shutdown() docs). ------------------------------------------------------------------- """ global RunStartTime elapsed = time.perf_counter() - RunStartTime # float seconds h, m, s = (elapsed // 3600), ((elapsed % 3600) // 60), (elapsed % 60) print('\nFinished - total elapsed time (h:m:s) = %d:%d:%.0f' % (h, m, s)) print(message) shutdown() def shutdown(bad=False): """ ------------------------------------------------------------------- All exits: pause for an Enter press iff a PC-side script was run on Windows by an icon click, else the console window is closed before its output can be read. Then, end the script run now with shell status 1 if bad (errors), or 0 if not bad (good/normal). Caveat: as coded, this may require an Enter in some other Windows contexts besides icon clicks, unless streams are redirected. Its PROMPT test suffices to keep click windows up, and does not require an Enter in Command Prompt, Cygwin, or some IDEs (e.g., PyEdit). But it does require an Enter in both PowerShell and the IDLE IDE. The only way to do better seems sniffing parent processes, and this is too complex and brittle. There really should be a better way to detect this very common Windows use case by now... Update: Termux:Widget launches similarly erase the console window on script exit (and open a different Termux session). To support this, test a ForceKeyToClose config-file setting on phone. This is preset to False; Termux:Widget is nice, but likely rare in the wild. This config also works on PCs, but is not needed for Windows clicks. See also _etc/termux-widget-shims/ for wrappers that make this moot. ------------------------------------------------------------------- """ if (ForceKeyToClose or # Termux:Widget? (RunningOnWindows and # Windows + sys.stdout.isatty() and # console output + sys.stdin.isatty() and # console input + 'PROMPT' not in os.environ)): # clicked only? (maybe) builtin_input('Press enter/return to exit') sys.exit(1 if bad else 0) def offerFixnamesRun(loglabel, target): """ ------------------------------------------------------------------- On PC, ask to run the nonportable-filenames fixer, in both initial copies and syncs. This is nearly required on Linux and macOS for interoperability with Windows, and FAT32, exFAT, and BDR drives. But don't ask on Windows: Windows disallows nonportables globally, and Cygwin's munged nonportables register as the munges instead of the nonportables if it's using Windows Python. Cygwin's own Python does translates munges to nonportables, but also sets sys.platform to 'cygwin' so it's not RunningOnWindows. WSL munges too, but its sys.platform is 'linux'. See _etc/examples/windows*/x* for demos. For convenience, this scrapes the fixer's last two report lines as a summary, and pulls out the number-found at the end of the last. This is both brittle and error prone, and requires forging user input for the script's verify, but reduces output and interaction. Update [1.2]: this is now also run on the _phone_ by part 1 of the new export scripts, because it's not impossible that names have been created in Android app-private storage which will not work on Windows and some drives. This required adding CheckFilenames to the phone configs file too (it will loaded from there by the on-phone export script), as well as the target path parameter here (which is join(FROM, STUFF) on PC, and join(TO, STUFF) on phone). Auto-detect of app storage is tough; set CheckFilenames as desired. ------------------------------------------------------------------- """ if CheckFilenames and not RunningOnWindows: print('\n' 'It is recommended to run fix-nonportable-filenames.py before\n' 'propagating content from Unix to Windows, some Android\'s shared\n' 'storage, and proxy drives using the FAT32 or exFAT filesystems.') fixer = join(MALL, 'fix-nonportable-filenames.py') logto = join(LOGS, '%s--%s-fixnames-log.txt' % (DATE, loglabel)) userreply = input('Run the name-fixer script in report-only mode (y or n)?') if userreply == 'y': print('Running name fixer') lasttwo = runpy(fixer, target, '-', input=b'y\n', logpath=logto, tailing=(2, 'Name-fixer summary')) # end interaction now if no lines to change numfound = int(lasttwo[-1].split(':')[1].strip()) if numfound == 0: print('No names to change') return userreply = input('Display name-fixer\'s full output (y or n)?') if userreply == 'y': print(open(logto, 'r', encoding='utf8').read()) userreply = input('Run the name-fixer script to fix filenames (y or n)?') if userreply != 'y': print('Name-fixer script not run') else: print('Running name fixer') runpy(fixer, target, input=b'y\n', logpath=logto, tailing=(2, 'Name-fixer summary')) def verifyPCtoProxy(loglabelM, loglabelD): """ ------------------------------------------------------------------- On PC, verify that PC and proxy-drive content is the same. This is called for both initial copies and content syncs. It asks to run mergeall and diffall separately: the former is fast, but the latter is slow for larger content collections. Similar code in verify*part2 couldn't be factored into this. Update [1.2]: now also run from export part 2 (PC) post syncs. ------------------------------------------------------------------- """ if VerifyProxy: userreply = input('\nVerify PC copy to proxy copy with mergeall (y or n)?') if userreply == 'y': print('Running mergeall') logto = join(LOGS, '%s--%s-verify-mergeall-log.txt' % (DATE, loglabelM)) runpy(join(MALL, 'mergeall.py'), join(FROM, STUFF), join(TO, STUFF), '-report', '-skipcruft', '-quiet', logpath=logto, tailing=(9, 'Mergeall summary')) userreply = input('\nVerify PC copy to proxy copy with diffall (y or n)?') if userreply == 'y': print('Running diffall') logto = join(LOGS, '%s--%s-verify-diffall-log.txt' % (DATE, loglabelD)) runpy(join(MALL, 'diffall.py'), join(FROM, STUFF), join(TO, STUFF), '-skipcruft', '-quiet', logpath=logto, tailing=(7, 'Diffall summary')) # 6=none, 7=1 unique (__bkp__) def previewChanges(loglabel): """ ------------------------------------------------------------------- On PC, show PC~proxy deltas before saving or propagating, for sync runs only. Like offerFixnamesRun(), this routes output to a file and scrapes its final report, to reduce output in the console. ------------------------------------------------------------------- """ if PreviewSyncs: userreply = input('\nPreview changes in the source tree (y or n)?') if userreply == 'y': print('Running preview of changes to be synced') logto = join(LOGS, '%s--%s-preview-log.txt' % (DATE, loglabel)) runpy(join(MALL, 'mergeall.py'), join(FROM, STUFF), join(TO, STUFF), '-report', '-skipcruft', '-quiet', logpath=logto, tailing=(9, 'Preview summary')) userreply = input('Display preview\'s full output (y or n)?') if userreply == 'y': print(open(logto, 'r', encoding='utf8').read()) userreply = input('Continue with sync (y or n)?') if userreply != 'y': print('Sync aborted.') shutdown() def removeExistingOrStop(thedst, where): """ ------------------------------------------------------------------- In both initial-copy scripts, get the user's okay and then delete existing content trees on proxy or phone; shutdown if not approved. Also used in the phone-sync script before unzipping, in case a DELTAS folder is present from a prior-sync abort or other source. deltas.py always removes an existing DELTAS folder before starting. All calls to this precede a zip-extract, which requires an empty dir. Update [1.2]: all calls to this are also crucial, so required=True for the new rmtree_FWP(): if the user okays the delete, the user must rm on delete failures, else stop run to avoid content damage. At last count, there were 5 calls to this, and 6 to rmtree_FWP(). ------------------------------------------------------------------- """ if os.path.exists(thedst): print('Removing prior content on', where) userreply = input('Proceed with removal (y or n)?') if userreply != 'y': print('Run stopped for existing content.') shutdown() else: timefunc(lambda: rmtree_FWP(thedst, required=True)) print('Starting unzip') def moveZipToProxy(thezip, prxzip): """ ------------------------------------------------------------------- Run a copy+delete across filesystems. This is used by PC scripts for both the initial copy's full zip, and the sync's deltas zip. prxzip is already a file in driveRootPath(); run early for messages. This cannot use a simple move, because from/to devices differ. ------------------------------------------------------------------- """ try: if os.path.exists(prxzip): os.remove(prxzip) shutil.copy2(thezip, prxzip) os.remove(thezip) except Exception as E: print('Zip move failed - Python exception:', E) print('Shutting down; check permissions and space and try again.') shutdown(bad=True) def tail(filename, lines, message): """ ------------------------------------------------------------------- A portable 'tail -n lines filename': show the last 'lines' lines in file 'filename'. This is inefficient for large files as coded, but seek-and-scan is complex, and memory use is not a concern here. ------------------------------------------------------------------- """ print(message + ':') lastfew = open(filename, 'r', encoding='utf8').readlines()[-lines:] for line in lastfew: print(indent + line.rstrip()) return lastfew # tail lines for scraping def runpy(*cmdargs, logpath=None, input=None, showcmd=False, tailing=None): """ ------------------------------------------------------------------- Run a Python command line, optionally sending stdout to a file named in logpath, and providing stdin text from bytes input. *cmdargs is individual words in the command line to run; the first is the pathname of the Python script to run. This adds the host Python's path to the front of *cmdargs automatically. tailing can be used iff logpath is a filename. It is a tuple (lines, message) whose items are passed to the tail() function along with the logpath name. When used, the return value is the result of tail(); else, the return value is elapsed time. Used on both PC and phone, portable to Win, Mac, Linux, Android. Caveat: subprocess.run() requires Python 3.5+, but this is from 6 years ago (2015), and anything older would be rare to find today. Caveat: can't use showRuntimes=ShowRuntimes arg; not yet defined. Update: now always adds the '-u' switch of Python (not script) to force unbuffered stdout/stderr. Else, logfiles may be incomplete if inspected while a long-running operation is in progress (e.g., a shared-storage delete of a large __bkp__ folder on the phone). diffall has its own '-u' but it's used only for frozen apps/exes. [1.2] Check exit status of Python run and shutdown immediately if the run failed (status != 0). The Python exception will be printed to stderr (the console) too. This should never happen... except that it did, after a bad copy of a zipfile to app-private storage. ------------------------------------------------------------------- """ cmdargs = (sys.executable, '-u') + cmdargs if showcmd: print('Command:', cmdargs) start = time.perf_counter() if not logpath: res = subprocess.run(args=cmdargs, input=input) else: logfile = open(logpath, 'w', encoding='utf8') res = subprocess.run(args=cmdargs, stdout=logfile, input=input) logfile.close() if res.returncode != 0: print('Python run failed:', cmdargs) print('Exiting; please resolve and retry.') shutdown(bad=True) stop = time.perf_counter() if ShowRuntimes: elapsed = stop - start print(indent + 'Runtime: %dm, %.3fs' % (elapsed // 60, elapsed % 60)) if logpath and tailing: lines, message = tailing lastfew = tail(logpath, lines, message) return lastfew else: return elapsed def timefunc(func): """ ------------------------------------------------------------------- Time any function call; could be a decorator, but also for lambdas. ------------------------------------------------------------------- """ start = time.perf_counter() func() stop = time.perf_counter() if ShowRuntimes: elapsed = stop - start print(indent + 'Runtime: %dm, %.3fs' % (elapsed // 60, elapsed % 60)) # constants for next function FileLimit = 260 - 1 # 259 in Py, including 3 for drive, N for UNC DirLimit = FileLimit - 12 # 247 in Py, after reserving 12 for 8.3 name def FWP(pathname, force=False, limit=DirLimit, trace=False): r""" ------------------------------------------------------------------- [1.2] Fix too-long paths on Windows (only) by prefixing as needed to invoke APIs that support extended-length paths. force=True is used to fix paths preemptively where needed, regardless of length. Use for top-level paths passed to tools that walk folders - including the shutil.rmtree() call here. The fix is a direct add for local-drive paths: 'C:\folder...' => '\\?\C:\folder...' But requires a minor reformatting step for network-drive paths: '\\server\folder...' => '\\?\UNC\server\folder...' Either way, this prefix lifts the pathname length limit to 32k chars, with 255 chars generally allowed per path component. This is true even on PCs not configured to allow long paths. Paths are also made absolute before prefixing per Windows rules, so it's okay if pathname is relative as passed in; the absolute and prefixed form returned here will be sent to file-op calls. For additional docs trimmed here, see Mergeall's fixlongpaths.py, from which this was copied (learning-python.com/mergeall.html). ------------------------------------------------------------------- """ if not RunningOnWindows: # macOS, Linux, Android, etc.: no worries return pathname else: abspathname = os.path.abspath(pathname) # use abs len (see above) if len(abspathname) <= limit and not force: # rel path len is moot # Windows path within limits: ok return pathname else: # Windows path too long: fix it pathname = abspathname # to absolute, and / => \ extralenprefix = '\\\\?\\' # i.e., \\?\ (or r'\\?'+'\\') if not pathname.startswith('\\\\'): # i.e., \\ (or r'\\') # local drives: C:\ pathname = extralenprefix + pathname # C:\dir => \\?\C:\dir else: # network drives: \\... # \\dev => \\?\UNC\dev pathname = extralenprefix + 'UNC' + pathname[1:] if trace: print('Extended path =>', pathname[:60]) return pathname def rmtree_FWP(folderpath, required=False): """ ------------------------------------------------------------------- [1.2] Call this everywhere, instead of shutil.rmtree() directly. On Windows, the path must be prefixed with FWP() for longpaths _before_ shutil.rmtree(). Per ahead, onerror doesn't suffice. This is just for folder removals in this package; the nested Mergeall and ziptools do similar for all their file operations. With this new scheme, longpath items in folders on Windows will generally be deleted without messages. As a last resort, ask the user to remove the item if rmtreeworkaround() and its error handler fail, instead of reraising an exception that triggers a crash and confusing exception message. Verify if required==True. We can ignore final clean-up deletes, but not initial content or deltas deletes - these both could lead to data-integrity issues. Pass required=True for contexts that require a delete to continue. Deletes can fail for many reasons, including locks for files in use, intentional read-only permissions, or system bugs (it happens). ------------------------------------------------------------------- """ try: fixfolderpath = FWP(folderpath, force=True) shutil.rmtree(fixfolderpath, onerror=rmtreeworkaround) except Exception as E: print('System error - cannot delete folder "%s"' % folderpath) print('\nException info:', E, end='\n\n') if required: input('Please delete manually, and press enter/return here when finished...') if os.path.exists(FWP(folderpath)): print('Sorry - cannot continue with folder not empty; exiting.') shutdown(bad=True) else: print('Please delete manually at your discretion.') def rmtreeworkaround(function, fullpath, exc_info): """ ------------------------------------------------------------------- Catch and try to recover from failures in Python's shutil.rmtree() folder removal tool, which calls this function automatically on all system errors. Raising an exception here makes shutil.rmtree() raise too, and a return makes it continue with results that vary per function (e.g., os.rmdir can recover, but os.scandir cannot). This is adapted from Mergeall, which documents it in full. In short, deletions are not always atomic on Windows, which matters for folder contents; macOS may remove "._*' Apple-double files automatically with their main file before they are reached here; and Windows may similarly remove folders automatically with their associated files. Other failures may stem from permissions or in-use locks: propagate. Note: Windows auto-removal of folders with files might only occur in Explorer; a TBD, but if so, the test here is pointless but harmless. macOS auto-removal of Apple-double files is pervasive, and Windows pending-delete loops (and longpaths) have been spotted in the wild. ---- Update [1.2]: Windows auto-deletes The Windows auto-folder-deletion test has been disabled pending an unlikely use case. Per recent testing, the auto-delete of a folder with its similarly named file is not triggered by this system's file ops. It's also not just an Explorer thing, but a Windows concept known as "Connected Files" which can be disabled in the registry. To test: save any web page completely on Windows, such that both an HTML file and a similarly named folder with a "_files" suffix (usually) are saved. Then, deletion results vary by context: In Windows Explorer: Auto-deletes happen - deleting the file automatically deletes the folder, and deleting the folder automatically deletes the file. So it's not just folders: associated files can vanish too. In the shell (Command Prompt): Auto-deletes do NOT happen - deleting the file with "del" does not automatically delete the folder, and deleting the folder with "rmdir /S" does not automatically delete the file. In Python (py -3): Auto-deletes do NOT happen - deleting the file with "os.remove" (a.k.a. "os.unlink") does not automatically delete the folder, and deleting the folder with "shutil.rmtree" does not automatically delete the file. Hence, this is an "Explorer thing" in the sense that its code or APIs must recognize Connected Files. File ops used in the shell and Python, however, do not. The latter makes this case moot in this system. While files and/or folders might be deleted outside this system's code while a delete is in progress here, that's a different issue, and out of scope. Nit: this assumes that website saves are representative of Connected Files in general. Else, it's unclear how this system can handle both vanishing files and folders (errors are broad), but it very likely doesn't have to. See https://duckduckgo.com/?q=windows+Connected+Files" for more; since this system won't invoke the auto-deletes, though, it's irrelevant here. Microsoft also covers its older but typically proprietary behavior here: https://learn.microsoft.com/en-us/windows/win32/shell/manage#connected-files ---- Update [1.2]: code typo fixes Fixed two uncaught but rarely triggered typos: - RunningOnMac => RunningOnMacOS, which matters only on macOS when "._*" files are auto-deleted before reached here. - Import 'errno' else undefined, which matters only on Windows when folders trigger the pending-deletions loop here. ---- Update [1.2]: Windows longpaths support, take 2 This system now wraps folder paths in FWP() before every call to shutil.rmtree, to avoid Windows path-too-long errors. The FWP() prefixing call is a no-op outside Windows; see its code above. An initial coding (see the defunct "take 1" below) tried calling FWP() _here_ on errors only, but that doesn't work in all cases: for folders, Python's shutil.rmtree first tries os.scandir() and issues a call here if it fails on a too-long-path error, but then does nothing to recover but set a folder-contents list to empty. Hence, there's no way to adjust here that allows shutil.rmtree to continue, and the folder's removal simply fails on a not-empty state. Calling FWP() _before_ the call to shutil.rmtree--like Mergeall does--avoids the issue, because os.scandir() will succeed. Code here must still handle other failures, including Windows pending deletions and macOS AppleDouble auto-removals. While at it, the higher-level rmtree_FWP() removal function now also asks the user to delete manually if all else fails, instead of crashing or risking content corruption. All of which should be _exceedingly_ rare, but stuff happens. For examples of the console messages and interaction of the new scheme, as well as additional background info, see this file: _etc/examples/x-1.2-export-example/_etc/windows-fwp-fix-results.txt ---- Update [1.2]: Windows longpaths support, take 1 (defunct) Try fixing longpaths on Windows on any error. The FWP() function lifts Window's 260ish-character limit for path names, and is borrowed from Mergeall and ziptools. Both of those systems underly the deltas scripts, and already wrap all file calls in FWP(). Hence, most ops are already longpath safe, but this must be able to delete folders which ziptools makes. This is an issue here only for folder-delete calls, so it's okay to adjust in this error handler instead of at top-level calls. This is also not an issue if Windows users enable longpaths via the registry or Python installs, but this cannot be expected, and one crash has already been seen. Still, the fix here will only kick in on PCs that otherwise fail for long paths. ------------------------------------------------------------------- """ # initial attempt to fix with FWP() here removed: see docstr [1.2] # Windows only, directory deletes only if RunningOnWindows and function == os.rmdir: """ # assume removed by Windows, or other # moot per above - disable till use case arises [1.2] if exc_info[0] == FileNotFoundError: msg = '...ignored FileNotFoundError for Windows dir' print(msg, fullpath) return # folder deleted with file?: proceed """ # wait for pending deletes of contents timeout = 0.001 # nit: need to try iff ENOTEMPTY while timeout < 1.0: # 10 tries only, increasing delays print('...retrying rmdir: ', fullpath) # set off, but not just for pruning try: os.rmdir(fullpath) # rerun the failed delete (with FWP!) except os.error as exc: if exc.errno == errno.ENOENT: # no such file (not-empty=ENOTEMPTY) return # it's now gone: proceed with rmtree else: time.sleep(timeout) # wait for a fraction of second (.001=1 msec) timeout *= 2 # and try again, with longer delay else: return # it's now gone: proceed with rmtree # macOS only, ignore file-not-found for AppleDouble files if RunningOnMacOS and exc_info[0] == FileNotFoundError: itemname = os.path.basename(fullpath) if itemname.startswith('._'): # assume removed by macOS, or other print('...ignored FileNotFoundError for AppleDouble', fullpath) return raise # all other cases, or wait loop end: reraise exception to kill rmtree caller def driveRootPath(path): r""" ------------------------------------------------------------------- Given a path, return its prefix which identifies the root of the drive on which it resides. This is used to ensure that zipfiles are stored on the USB drive root, regardless of user TO settings. This keeps chopping off the path's tail item until a mount point is found, or empty or no more progress. It may suffice to call os.path.splitdrive() on Windows, but that's a no-op on POSIX. Examples: macOS: >>> driveRootPath('/Users/me/MY-STUFF/Code') '/' >>> driveRootPath('/Volumes/SAMSUNG_T5/fold3-vs-deltas') '/Volumes/SAMSUNG_T5' >>> driveRootPath('.') '/' # also '/' for 'folder/file' Windows: >>> driveRootPath(r'C:\Users\me\Desktop\MY-STUFF') 'C:\\' >>> driveRootPath('D:\\test-ads\\z-fold-3.txt') 'D:\\' >>> driveRootPath('.') 'C:\\' # also 'C:\\' for '\Users\me', 'C:folder\file', 'folder\file' Linux: >>> driveRootPath('/home/me/Download/fold3-vs-deltas') '/' >>> driveRootPath('/media/me/SAMSUNG_T5/fold3-vs-deltas') '/media/me/SAMSUNG_T5' >>> driveRootPath('.') '/' # also '/' for 'folder/file' ------------------------------------------------------------------- """ path = os.path.abspath(path) # normalize, use \ on win next = path while next: if os.path.ismount(next): # found the root path return next else: head, tail = os.path.split(next) if head == next: return path # give up: don't loop else: next = head # go to next path prefix return path # give up: no path left