![[Home page]](../PythonPowered.gif "Home page"){kind=small}
| Books | Code | Blog | Python | Author | Train | Find | ©M.Lutz |
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