Added check for export db in directory branch, closes #164
This commit is contained in:
@@ -1330,13 +1330,32 @@ def export(
|
|||||||
return
|
return
|
||||||
|
|
||||||
# open export database and assign copy/link/unlink functions
|
# open export database and assign copy/link/unlink functions
|
||||||
|
export_db_path = os.path.join(dest, OSXPHOTOS_EXPORT_DB)
|
||||||
|
|
||||||
|
# check that export isn't in the parent or child of a previously exported library
|
||||||
|
other_db_files = find_files_in_branch(dest, OSXPHOTOS_EXPORT_DB)
|
||||||
|
if other_db_files:
|
||||||
|
click.echo(
|
||||||
|
"WARNING: found other export database files in this destination directory branch. "
|
||||||
|
+ "This likely means you are attempting to export files into a directory "
|
||||||
|
+ "that is either the parent or a child directory of a previous export. "
|
||||||
|
+ "Proceeding may cause your exported files to be overwritten.",
|
||||||
|
err=True,
|
||||||
|
)
|
||||||
|
click.echo(
|
||||||
|
f"You are exporting to {dest}, found {OSXPHOTOS_EXPORT_DB} files in:"
|
||||||
|
)
|
||||||
|
for other_db in other_db_files:
|
||||||
|
click.echo(f"{other_db}")
|
||||||
|
click.confirm("Do you want to continue?", abort=True)
|
||||||
|
|
||||||
if dry_run:
|
if dry_run:
|
||||||
export_db = ExportDBInMemory(os.path.join(dest, OSXPHOTOS_EXPORT_DB))
|
export_db = ExportDBInMemory(export_db_path)
|
||||||
# echo = functools.partial(click.echo, err=True)
|
# echo = functools.partial(click.echo, err=True)
|
||||||
# fileutil = FileUtilNoOp(verbose=echo)
|
# fileutil = FileUtilNoOp(verbose=echo)
|
||||||
fileutil = FileUtilNoOp
|
fileutil = FileUtilNoOp
|
||||||
else:
|
else:
|
||||||
export_db = ExportDB(os.path.join(dest, OSXPHOTOS_EXPORT_DB))
|
export_db = ExportDB(export_db_path)
|
||||||
fileutil = FileUtil
|
fileutil = FileUtil
|
||||||
|
|
||||||
photos = _query(
|
photos = _query(
|
||||||
@@ -1504,7 +1523,7 @@ def export(
|
|||||||
else:
|
else:
|
||||||
photo_str = "photos" if len(results_exported) != 1 else "photo"
|
photo_str = "photos" if len(results_exported) != 1 else "photo"
|
||||||
click.echo(f"Exported: {len(results_exported)} {photo_str}")
|
click.echo(f"Exported: {len(results_exported)} {photo_str}")
|
||||||
click.echo(f"Elapsed time: {stop_time-start_time} seconds")
|
click.echo(f"Elapsed time: {(stop_time-start_time):.3f} seconds")
|
||||||
else:
|
else:
|
||||||
click.echo("Did not find any photos to export")
|
click.echo("Did not find any photos to export")
|
||||||
|
|
||||||
@@ -2186,7 +2205,7 @@ def get_dirnames_from_template(photo, directory, export_by_date, dest, dry_run):
|
|||||||
dest_path = os.path.join(dest, dirname)
|
dest_path = os.path.join(dest, dirname)
|
||||||
if not is_valid_filepath(dest_path, platform="auto"):
|
if not is_valid_filepath(dest_path, platform="auto"):
|
||||||
raise ValueError(f"Invalid file path: '{dest_path}'")
|
raise ValueError(f"Invalid file path: '{dest_path}'")
|
||||||
if not (dry_run or os.path.isdir(dest_path)):
|
if not dry_run and not os.path.isdir(dest_path):
|
||||||
os.makedirs(dest_path)
|
os.makedirs(dest_path)
|
||||||
dest_paths.append(dest_path)
|
dest_paths.append(dest_path)
|
||||||
else:
|
else:
|
||||||
@@ -2194,5 +2213,42 @@ def get_dirnames_from_template(photo, directory, export_by_date, dest, dry_run):
|
|||||||
return dest_paths
|
return dest_paths
|
||||||
|
|
||||||
|
|
||||||
|
def find_files_in_branch(pathname, filename):
|
||||||
|
""" Search a directory branch to find file(s) named filename
|
||||||
|
The branch searched includes all folders below pathname and
|
||||||
|
the parent tree of pathname but not pathname itself.
|
||||||
|
|
||||||
|
e.g. find filename in children folders and parent folders
|
||||||
|
|
||||||
|
Args:
|
||||||
|
pathname: str, full path of directory to search
|
||||||
|
filename: str, filename to search for
|
||||||
|
|
||||||
|
Returns: list of full paths to any matching files
|
||||||
|
"""
|
||||||
|
|
||||||
|
pathname = pathlib.Path(pathname).resolve()
|
||||||
|
files = []
|
||||||
|
|
||||||
|
# walk down the tree
|
||||||
|
for root, directories, filenames in os.walk(pathname):
|
||||||
|
# for directory in directories:
|
||||||
|
# print(os.path.join(root, directory))
|
||||||
|
for fname in filenames:
|
||||||
|
if fname == filename and pathlib.Path(root) != pathname:
|
||||||
|
files.append(os.path.join(root, fname))
|
||||||
|
|
||||||
|
# walk up the tree
|
||||||
|
path = pathlib.Path(pathname)
|
||||||
|
for root in path.parents:
|
||||||
|
filenames = os.listdir(root)
|
||||||
|
for fname in filenames:
|
||||||
|
filepath = os.path.join(root, fname)
|
||||||
|
if fname == filename and os.path.isfile(filepath):
|
||||||
|
files.append(os.path.join(root, fname))
|
||||||
|
|
||||||
|
return files
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
cli() # pylint: disable=no-value-for-parameter
|
cli() # pylint: disable=no-value-for-parameter
|
||||||
|
|||||||
@@ -1,3 +1,3 @@
|
|||||||
""" version info """
|
""" version info """
|
||||||
|
|
||||||
__version__ = "0.29.20"
|
__version__ = "0.29.21"
|
||||||
|
|||||||
@@ -1738,6 +1738,59 @@ def test_export_update_basic():
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_export_update_child_folder():
|
||||||
|
""" test export then update into a child folder of previous export """
|
||||||
|
import glob
|
||||||
|
import os
|
||||||
|
import os.path
|
||||||
|
|
||||||
|
import osxphotos
|
||||||
|
from osxphotos.__main__ import export, OSXPHOTOS_EXPORT_DB
|
||||||
|
|
||||||
|
runner = CliRunner()
|
||||||
|
cwd = os.getcwd()
|
||||||
|
# pylint: disable=not-context-manager
|
||||||
|
with runner.isolated_filesystem():
|
||||||
|
# basic export
|
||||||
|
result = runner.invoke(export, [os.path.join(cwd, CLI_PHOTOS_DB), ".", "-V"])
|
||||||
|
assert result.exit_code == 0
|
||||||
|
|
||||||
|
os.mkdir("foo")
|
||||||
|
|
||||||
|
# update into foo
|
||||||
|
result = runner.invoke(
|
||||||
|
export, [os.path.join(cwd, CLI_PHOTOS_DB), "foo", "--update"], input="N\n"
|
||||||
|
)
|
||||||
|
assert result.exit_code != 0
|
||||||
|
assert "WARNING: found other export database files" in result.output
|
||||||
|
|
||||||
|
|
||||||
|
def test_export_update_parent_folder():
|
||||||
|
""" test export then update into a parent folder of previous export """
|
||||||
|
import glob
|
||||||
|
import os
|
||||||
|
import os.path
|
||||||
|
|
||||||
|
import osxphotos
|
||||||
|
from osxphotos.__main__ import export, OSXPHOTOS_EXPORT_DB
|
||||||
|
|
||||||
|
runner = CliRunner()
|
||||||
|
cwd = os.getcwd()
|
||||||
|
# pylint: disable=not-context-manager
|
||||||
|
with runner.isolated_filesystem():
|
||||||
|
# basic export
|
||||||
|
os.mkdir("foo")
|
||||||
|
result = runner.invoke(export, [os.path.join(cwd, CLI_PHOTOS_DB), "foo", "-V"])
|
||||||
|
assert result.exit_code == 0
|
||||||
|
|
||||||
|
# update into "."
|
||||||
|
result = runner.invoke(
|
||||||
|
export, [os.path.join(cwd, CLI_PHOTOS_DB), ".", "--update"], input="N\n"
|
||||||
|
)
|
||||||
|
assert result.exit_code != 0
|
||||||
|
assert "WARNING: found other export database files" in result.output
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.skipif(exiftool is None, reason="exiftool not installed")
|
@pytest.mark.skipif(exiftool is None, reason="exiftool not installed")
|
||||||
def test_export_update_exiftool():
|
def test_export_update_exiftool():
|
||||||
""" test export then update with exiftool """
|
""" test export then update with exiftool """
|
||||||
|
|||||||
Reference in New Issue
Block a user