1
0
mirror of https://github.com/deadc0de6/dotdrop.git synced 2026-02-05 21:23:02 +00:00

refactor ignore

This commit is contained in:
deadc0de6
2023-08-07 23:37:14 +02:00
committed by deadc0de
parent 3d22b7beeb
commit 018cd3decd
4 changed files with 182 additions and 154 deletions

View File

@@ -12,7 +12,8 @@ import shutil
from dotdrop.logger import Logger
from dotdrop.utils import strip_home, get_default_file_perms, \
get_file_perm, get_umask, must_ignore, \
get_unique_tmp_name, removepath
get_unique_tmp_name, removepath, copytree_with_ign, \
copyfile
from dotdrop.linktypes import LinkTypes
from dotdrop.comparator import Comparator
from dotdrop.templategen import Templategen
@@ -148,7 +149,7 @@ class Importer:
self.log.dbg(f'import dotfile: src:{src} dst:{dst}')
if not self._import_file(src, dst, trans_write=trans_write):
if not self._import_to_dotpath(src, dst, trans_write=trans_write):
return -1
return self._import_in_config(path, src, dst, perm, linktype,
@@ -165,7 +166,6 @@ class Importer:
1: 1 dotfile imported
0: ignored
"""
# handle file mode
chmod = None
dflperm = get_default_file_perms(dst, self.umask)
@@ -209,68 +209,46 @@ class Importer:
self.log.dbg('will overwrite existing file')
return True
def _import_file(self, src, dst, trans_write=None):
def _import_to_dotpath(self, in_dotpath, in_fs, trans_write=None):
"""
prepare hierarchy for dotfile in dotpath
and copy file
src is file in dotpath
dst is file on filesystem
prepare hierarchy for dotfile in dotpath and copy file
"""
srcf = os.path.join(self.dotpath, src)
srcfd = os.path.dirname(srcf)
# check if must be ignored
if self._ignore(srcf) or self._ignore(srcfd):
return False
srcf = os.path.join(self.dotpath, in_dotpath)
# check we are not overwritting
if not self._check_existing_dotfile(srcf, dst):
if not self._check_existing_dotfile(srcf, in_fs):
return False
# create directory hierarchy
if self.dry:
cmd = f'mkdir -p {srcfd}'
self.log.dry(f'would run: {cmd}')
else:
try:
os.makedirs(srcfd, exist_ok=True)
except OSError:
self.log.err(f'importing \"{dst}\" failed!')
return False
# import the file
if self.dry:
self.log.dry(f'would copy {dst} to {srcf}')
else:
# apply trans_w
dst = self._apply_trans_w(dst, trans_write)
if not dst:
# transformation failed
return False
# copy the file to the dotpath
try:
if os.path.isdir(dst):
if os.path.exists(srcf):
shutil.rmtree(srcf)
ign = shutil.ignore_patterns(*self.ignore)
shutil.copytree(dst, srcf,
copy_function=self._cp,
ignore=ign)
else:
shutil.copy2(dst, srcf)
except shutil.Error as exc:
src = exc.args[0][0][0]
why = exc.args[0][0][2]
self.log.err(f'importing \"{src}\" failed: {why}')
self.log.dry(f'would copy {in_fs} to {srcf}')
return True
return True
# apply trans_w
in_fs = self._apply_trans_w(in_fs, trans_write)
if not in_fs:
# transformation failed
return False
# copy the file to the dotpath
try:
if not os.path.isdir(in_fs):
# is a file
self.log.dbg(f'{in_fs} is file')
copyfile(in_fs, srcf, debug=self.debug)
else:
# is a dir
if os.path.exists(srcf):
shutil.rmtree(srcf)
self.log.dbg(f'{in_fs} is dir')
copytree_with_ign(in_fs, srcf,
ignore_func=self._ignore,
debug=self.debug)
except shutil.Error as exc:
in_dotpath = exc.args[0][0][0]
why = exc.args[0][0][2]
self.log.err(f'importing \"{in_fs}\" failed: {why}')
def _cp(self, src, dst):
"""the copy function for copytree"""
# test if must be ignored
if self._ignore(src):
return
shutil.copy2(src, dst)
return os.path.exists(srcf)
def _already_exists(self, src, dst):
"""

View File

@@ -12,7 +12,12 @@ import shutil
# local imports
from dotdrop.logger import Logger
from dotdrop.linktypes import LinkTypes
from dotdrop import utils
from dotdrop.utils import copyfile, get_file_perm, \
pivot_path, must_ignore, removepath, \
samefile, write_to_tmpfile, fastdiff, \
content_empty
from dotdrop.utils import chmod as chmodit
from dotdrop.utils import diff as diffit
from dotdrop.exceptions import UndefinedException
from dotdrop.cfg_yaml import CfgYaml
@@ -162,7 +167,7 @@ class Installer:
ignore=ignore)
if self.log.debug and chmod:
cur = utils.get_file_perm(dst)
cur = get_file_perm(dst)
if chmod == CfgYaml.chmod_ignore:
chmodstr = CfgYaml.chmod_ignore
else:
@@ -188,9 +193,9 @@ class Installer:
apply_chmod = apply_chmod and chmod != CfgYaml.chmod_ignore
if apply_chmod:
if not chmod:
chmod = utils.get_file_perm(src)
chmod = get_file_perm(src)
self.log.dbg(f'applying chmod {chmod:o} to {dst}')
dstperms = utils.get_file_perm(dst)
dstperms = get_file_perm(dst)
if dstperms != chmod:
# apply mode
msg = f'chmod {dst} to {chmod:o}'
@@ -200,7 +205,7 @@ class Installer:
else:
if not self.comparing:
self.log.sub(f'chmod {dst} to {chmod:o}')
if utils.chmod(dst, chmod, debug=self.debug):
if chmodit(dst, chmod, debug=self.debug):
ret = True
else:
ret = False
@@ -250,7 +255,7 @@ class Installer:
self.totemp = None
# install the dotfile to a temp directory
tmpdst = utils.pivot_path(dst, tmpdir, logger=self.log)
tmpdst = pivot_path(dst, tmpdir, logger=self.log)
ret, err = self.install(templater, src, tmpdst,
LinkTypes.NOLINK,
is_template=is_template,
@@ -331,8 +336,8 @@ class Installer:
"""
if is_template:
self.log.dbg(f'is a template, installing to {self.workdir}')
tmp = utils.pivot_path(dst, self.workdir,
striphome=True, logger=self.log)
tmp = pivot_path(dst, self.workdir,
striphome=True, logger=self.log)
ret, err = self.install(templater, src, tmp,
LinkTypes.NOLINK,
actionexec=actionexec,
@@ -388,7 +393,7 @@ class Installer:
subsrc = srcs[i]
subdst = dsts[i]
if utils.must_ignore([subsrc, subdst], ignore, debug=self.debug):
if must_ignore([subsrc, subdst], ignore, debug=self.debug):
self.log.dbg(
f'ignoring install of {src} to {dst}',
)
@@ -399,8 +404,8 @@ class Installer:
if is_template:
self.log.dbg('child is a template')
self.log.dbg(f'install to {self.workdir} and symlink')
tmp = utils.pivot_path(subdst, self.workdir,
striphome=True, logger=self.log)
tmp = pivot_path(subdst, self.workdir,
striphome=True, logger=self.log)
ret2, err2 = self.install(templater, subsrc, tmp,
LinkTypes.NOLINK,
actionexec=actionexec,
@@ -456,7 +461,7 @@ class Installer:
# remove symlink
overwrite = True
try:
utils.removepath(dst)
removepath(dst)
except OSError as exc:
err = f'something went wrong with {src}: {exc}'
return False, err
@@ -481,7 +486,7 @@ class Installer:
if self.safe and not overwrite and not self.log.ask(msg):
return False, 'aborted'
try:
utils.removepath(dst)
removepath(dst)
except OSError as exc:
err = f'something went wrong with {src}: {exc}'
return False, err
@@ -497,7 +502,7 @@ class Installer:
os.symlink(lnk_src, dst)
self.log.dbg(
f'symlink {dst} to {lnk_src} '
f'(mode:{utils.get_file_perm(dst):o})'
f'(mode:{get_file_perm(dst):o})'
)
if not self.comparing:
self.log.sub(f'linked {dst} to {lnk_src}')
@@ -522,12 +527,12 @@ class Installer:
self.log.dbg(f'no empty: {noempty}')
# ignore file
if utils.must_ignore([src, dst], ignore, debug=self.debug):
if must_ignore([src, dst], ignore, debug=self.debug):
self.log.dbg(f'ignoring install of {src} to {dst}')
return False, None
# check no loop
if utils.samefile(src, dst):
if samefile(src, dst):
err = f'dotfile points to itself: {dst}'
return False, err
@@ -548,7 +553,7 @@ class Installer:
finally:
templater.restore_vars(saved)
# test is empty
if noempty and utils.content_empty(content):
if noempty and content_empty(content):
self.log.dbg(f'ignoring empty template: {src}')
return False, None
if content is None:
@@ -561,7 +566,7 @@ class Installer:
actionexec=actionexec)
if ret and not err:
rights = f'{utils.get_file_perm(src):o}'
rights = f'{get_file_perm(src):o}'
self.log.dbg(f'installed file {src} to {dst} ({rights})')
if not self.dry and not self.comparing:
self.log.sub(f'install {src} to {dst}')
@@ -627,7 +632,6 @@ class Installer:
try:
with open(dst, 'wb') as file:
file.write(content)
# shutil.copymode(src, dst)
except NotADirectoryError as exc:
err = f'opening dest file: {exc}'
return False, err
@@ -638,8 +642,8 @@ class Installer:
else:
# copy file
try:
# do NOT copy meta here
shutil.copyfile(src, dst)
# shutil.copymode(src, dst)
except OSError as exc:
return False, str(exc)
return True, None
@@ -708,9 +712,9 @@ class Installer:
return False, 'aborted'
# writing to file
self.log.dbg(f'before writing to {dst} ({utils.get_file_perm(src):o})')
self.log.dbg(f'before writing to {dst} ({get_file_perm(src):o})')
ret = self._write_content_to_file(content, src, dst)
self.log.dbg(f'written to {dst} ({utils.get_file_perm(src):o})')
self.log.dbg(f'written to {dst} ({get_file_perm(src):o})')
return ret
########################################################
@@ -732,13 +736,13 @@ class Installer:
# check file content
tmp = None
if content:
tmp = utils.write_to_tmpfile(content)
tmp = write_to_tmpfile(content)
src = tmp
ret = utils.fastdiff(src, dst)
ret = fastdiff(src, dst)
if ret:
self.log.dbg('content differ')
if content:
utils.removepath(tmp)
removepath(tmp)
return ret
def _show_diff_before_write(self, src, dst, content=None):
@@ -749,12 +753,12 @@ class Installer:
"""
tmp = None
if content:
tmp = utils.write_to_tmpfile(content)
tmp = write_to_tmpfile(content)
src = tmp
diff = utils.diff(modified=src, original=dst,
diff_cmd=self.diff_cmd)
diff = diffit(modified=src, original=dst,
diff_cmd=self.diff_cmd)
if tmp:
utils.removepath(tmp, logger=self.log)
removepath(tmp, logger=self.log)
if diff:
self._print_diff(src, dst, diff)
@@ -790,7 +794,7 @@ class Installer:
# copy to preserve mode on chmod=preserve
# since we expect dotfiles this shouldn't have
# such a big impact but who knows.
shutil.copy2(path, dst)
copyfile(path, dst, debug=self.debug)
stat = os.stat(path)
os.chown(dst, stat.st_uid, stat.st_gid)

View File

@@ -8,14 +8,13 @@ handle the update of dotfiles
import os
import shutil
import filecmp
import fnmatch
# local imports
from dotdrop.logger import Logger
from dotdrop.templategen import Templategen
from dotdrop.utils import ignores_to_absolute, removepath, \
get_unique_tmp_name, write_to_tmpfile, must_ignore, \
mirror_file_rights, get_file_perm
mirror_file_rights, get_file_perm, copytree_with_ign
from dotdrop.exceptions import UndefinedException
@@ -288,30 +287,22 @@ class Updater:
self.log.dry(f'would cp -r {exist} {new}')
continue
self.log.dbg(f'cp -r {exist} {new}')
# Newly created directory should be copied as is (for efficiency).
def ign(src, names):
whitelist, blacklist = set(), set()
for ignore in ignores:
for name in names:
path = os.path.join(src, name)
if ignore.startswith('!') and \
fnmatch.fnmatch(path, ignore[1:]):
# add to whitelist
whitelist.add(name)
elif fnmatch.fnmatch(path, ignore):
# add to blacklist
blacklist.add(name)
return blacklist - whitelist
try:
shutil.copytree(exist, new, ignore=ign)
ign_func = self._ignore(ignores, debug=self.debug)
copytree_with_ign(exist, new,
ignore_func=ign_func,
debug=self.debug)
except OSError as exc:
msg = f'error copying dir {exist}'
self.log.err(f'{msg}: {exc}')
continue
self.log.sub(f'\"{new}\" dir added')
def _ignore(self, ignores, debug=False):
def ignore_func(path):
return must_ignore([path], ignores, debug=debug)
return ignore_func
def _merge_dirs_remove_right_only(self, diff, left, right,
ignore_missing_in_dotdrop,
ignores):

View File

@@ -222,70 +222,125 @@ def strip_home(path):
return path
def _match_ignore_pattern(path, pattern, debug=False):
"""
returns true if path matches the pattern
"""
ret = fnmatch.fnmatch(path, pattern)
if debug:
LOG.dbg(f'ignore \"{pattern}\" match: {path}',
force=True)
return ret
def _must_ignore(path, ignores, neg_ignores, debug=False):
"""
return true if path matches any ignore patterns
"""
match_ignore_pattern = []
# test for ignore pattern
for pattern in ignores:
if _match_ignore_pattern(path, pattern):
match_ignore_pattern.append(path)
# remove negative match
for pattern in neg_ignores:
# remove '!'
pattern = pattern[1:]
if not _match_ignore_pattern(path, pattern):
if debug:
msg = f'negative ignore \"{pattern}\" NO match: {path}'
LOG.dbg(msg, force=True)
continue
# remove from the list
try:
match_ignore_pattern.remove(path)
except ValueError:
warn = 'no files that are currently being '
warn += f'ignored match \"{pattern}\". In order '
warn += 'for a negative ignore pattern '
warn += 'to work, it must match a file '
warn += 'that is being ignored by a '
warn += 'previous ignore pattern.'
LOG.warn(warn)
if len(match_ignore_pattern) < 1:
return False
if os.path.isdir(path):
# this ensures whoever calls this function will
# descend into the directory to explore the possiblity
# of a file matching the non-ignore pattern
if debug:
msg = 'ignore would have match but neg ignores'
msg += f' present and is a dir: \"{path}\" -> not ignored!'
LOG.dbg(msg, force=True)
return False
if debug:
LOG.dbg(f'effectively ignoring \"{path}\"', force=True)
return True
def must_ignore(paths, ignores, debug=False):
"""return true if any paths in list matches any ignore patterns"""
"""
return true if any paths in list matches any ignore patterns
"""
if not ignores:
return False
if debug:
LOG.dbg(f'must ignore? \"{paths}\" against {ignores}',
force=True)
ignored_negative, ignored = categorize(
nign, ign = categorize(
lambda ign: ign.startswith('!'), ignores)
for path in paths:
ignore_matches = []
isdir = os.path.isdir(path)
# First ignore dotfiles
for i in ignored:
if fnmatch.fnmatch(path, i):
if debug:
LOG.dbg(f'ignore \"{i}\" match: {path}',
force=True)
ignore_matches.append(path)
# Then remove any matches that actually shouldn't be ignored
for nign in ignored_negative:
# Each of these will start with an '!' so we need to remove that
nign = nign[1:]
if debug:
msg = f'trying to match :\"{path}\" '
msg += f'with non-ignore-pattern:\"{nign}\"'
LOG.dbg(msg, force=True)
if fnmatch.fnmatch(path, nign):
if debug:
msg = f'negative ignore \"{nign}\" match: {path}'
LOG.dbg(msg, force=True)
try:
ignore_matches.remove(path)
except ValueError:
warn = 'no files that are currently being '
warn += f'ignored match \"{nign}\". In order '
warn += 'for a negative ignore pattern '
warn += 'to work, it must match a file '
warn += 'that is being ignored by a '
warn += 'previous ignore pattern.'
LOG.warn(warn)
else:
if debug:
msg = f'negative ignore \"{nign}\" NO match: {path}'
LOG.dbg(msg, force=True)
if ignore_matches:
if debug:
LOG.dbg(f'effectively ignoring \"{paths}\"', force=True)
if isdir and len(ignored_negative) > 0:
# this ensures whoever calls this function will
# descend into the directory to explore the possiblity
# of a file matching the non-ignore pattern
if debug:
msg = 'ignore would have match but neg ignores'
msg += f' present and is a dir: \"{path}\" -> not ignored!'
LOG.dbg(msg, force=True)
return False
if _must_ignore(path, ign, nign, debug=debug):
return True
if debug:
LOG.dbg(f'NOT ignoring \"{paths}\"', force=True)
return False
def _cp(src, dst, ignore_func=None, debug=False):
"""the copy function for copytree"""
if ignore_func and ignore_func(src):
return
dstdir = os.path.dirname(dst)
if debug:
LOG.dbg(f'mkdir \"{dstdir}\"',
force=True)
os.makedirs(dstdir, exist_ok=True)
if debug:
LOG.dbg(f'cp {src} {dst}',
force=True)
shutil.copy2(src, dst)
def copyfile(src, dst, debug=False):
"""
copy file from src to dst
no dir expected!
"""
_cp(src, dst, debug=debug)
def copytree_with_ign(src, dst, ignore_func=None, debug=False):
"""copytree with support for ignore"""
if debug:
LOG.dbg(f'copytree \"{src}\" to \"{dst}\"', force=True)
for entry in os.listdir(src):
srcf = os.path.join(src, entry)
dstf = os.path.join(dst, entry)
if os.path.isdir(srcf):
if debug:
LOG.dbg(f'mkdir \"{dstf}\"',
force=True)
os.makedirs(dstf, exist_ok=True)
copytree_with_ign(srcf, dstf, ignore_func=ignore_func)
else:
if debug:
LOG.dbg(f'copytree, copy file \"{src}\" to \"{dst}\"',
force=True)
_cp(srcf, dstf, ignore_func=ignore_func, debug=debug)
def uniq_list(a_list):
"""unique elements of a list while preserving order"""
new = []