forked/osxphotos
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Added --dry-run option to CLI export, closes #91

Browse Source
This commit is contained in:
Rhet Turnbull
2020-05-25 10:37:30 -07:00
parent 46fdc94398
commit 9eae66030e
13 changed files with 888 additions and 428 deletions
Download Patch File Download Diff File Expand all files Collapse all files

173
osxphotos/__main__.py
View File

@@ -1,5 +1,7 @@
""" command line interface for osxphotos """
import csv
import datetime
import functools
import json
import logging
import os
@@ -20,15 +22,16 @@ from pathvalidate import (
import osxphotos
from ._constants import _EXIF_TOOL_URL, _PHOTOS_4_VERSION, _UNKNOWN_PLACE
from .datetime_formatter import DateTimeFormatter
from ._version import __version__
from .exiftool import get_exiftool_path
from .fileutil import FileUtil, FileUtilNoOp
from .photoinfo import ExportResults
from .photoinfo.template import (
TEMPLATE_SUBSTITUTIONS,
TEMPLATE_SUBSTITUTIONS_MULTI_VALUED,
)
from .utils import _copy_file, create_path_by_date
from ._export_db import ExportDB
from ._export_db import ExportDB, ExportDBInMemory
# global variable to control verbose output
# set via --verbose/-V
@@ -111,9 +114,11 @@ class ExportCommand(click.Command):
+ "should be treated as a backup, not a working copy where you intend to make changes. "
)
formatter.write("\n")
formatter.write_text("Note: The number of files reported for export and the number actually exported "
+"may differ due to live photos, associated RAW images, and edited photos which are reported "
+"in the total photos exported.")
formatter.write_text(
"Note: The number of files reported for export and the number actually exported "
+ "may differ due to live photos, associated RAW images, and edited photos which are reported "
+ "in the total photos exported."
)
formatter.write("\n")
formatter.write_text(
"Implementation note: To determine which files need to be updated, "
@@ -916,6 +921,11 @@ def query(
is_flag=True,
help="Only export new or updated files. See notes below on export and --update.",
)
@click.option(
"--dry-run",
is_flag=True,
help="Dry run (test) the export but don't actually export any files; most useful with --verbose",
)
@click.option(
"--export-as-hardlink",
is_flag=True,
@@ -1068,6 +1078,7 @@ def export(
to_date,
verbose_,
update,
dry_run,
export_as_hardlink,
overwrite,
export_by_date,
@@ -1190,8 +1201,15 @@ def export(
_list_libraries()
return
# open export database
export_db = ExportDB(os.path.join(dest, OSXPHOTOS_EXPORT_DB))
# open export database and assign copy/link/unlink functions
if dry_run:
export_db = ExportDBInMemory(os.path.join(dest, OSXPHOTOS_EXPORT_DB))
# echo = functools.partial(click.echo, err=True)
# fileutil = FileUtilNoOp(verbose=echo)
fileutil = FileUtilNoOp
else:
export_db = ExportDB(os.path.join(dest, OSXPHOTOS_EXPORT_DB))
fileutil = FileUtil
photos = _query(
db=db,
@@ -1269,26 +1287,28 @@ def export(
with click.progressbar(photos) as bar:
for p in bar:
results = export_photo(
p,
dest,
verbose_,
export_by_date,
sidecar,
update,
export_as_hardlink,
overwrite,
export_edited,
original_name,
export_live,
download_missing,
exiftool,
directory,
no_extended_attributes,
export_raw,
album_keyword,
person_keyword,
keyword_template,
export_db,
photo=p,
dest=dest,
verbose_=verbose_,
export_by_date=export_by_date,
sidecar=sidecar,
update=update,
export_as_hardlink=export_as_hardlink,
overwrite=overwrite,
export_edited=export_edited,
original_name=original_name,
export_live=export_live,
download_missing=download_missing,
exiftool=exiftool,
directory=directory,
no_extended_attributes=no_extended_attributes,
export_raw=export_raw,
album_keyword=album_keyword,
person_keyword=person_keyword,
keyword_template=keyword_template,
export_db=export_db,
fileutil=fileutil,
dry_run = dry_run,
)
results_exported.extend(results.exported)
results_new.extend(results.new)
@@ -1298,26 +1318,28 @@ def export(
else:
for p in photos:
results = export_photo(
p,
dest,
verbose_,
export_by_date,
sidecar,
update,
export_as_hardlink,
overwrite,
export_edited,
original_name,
export_live,
download_missing,
exiftool,
directory,
no_extended_attributes,
export_raw,
album_keyword,
person_keyword,
keyword_template,
export_db,
photo=p,
dest=dest,
verbose_=verbose_,
export_by_date=export_by_date,
sidecar=sidecar,
update=update,
export_as_hardlink=export_as_hardlink,
overwrite=overwrite,
export_edited=export_edited,
original_name=original_name,
export_live=export_live,
download_missing=download_missing,
exiftool=exiftool,
directory=directory,
no_extended_attributes=no_extended_attributes,
export_raw=export_raw,
album_keyword=album_keyword,
person_keyword=person_keyword,
keyword_template=keyword_template,
export_db=export_db,
fileutil=fileutil,
dry_run=dry_run,
)
results_exported.extend(results.exported)
results_new.extend(results.new)
@@ -1715,26 +1737,28 @@ def _query(
def export_photo(
photo,
dest,
verbose_,
export_by_date,
sidecar,
update,
export_as_hardlink,
overwrite,
export_edited,
original_name,
export_live,
download_missing,
exiftool,
directory,
no_extended_attributes,
export_raw,
album_keyword,
person_keyword,
keyword_template,
export_db,
photo=None,
dest=None,
verbose_=None,
export_by_date=None,
sidecar=None,
update=None,
export_as_hardlink=None,
overwrite=None,
export_edited=None,
original_name=None,
export_live=None,
download_missing=None,
exiftool=None,
directory=None,
no_extended_attributes=None,
export_raw=None,
album_keyword=None,
person_keyword=None,
keyword_template=None,
export_db=None,
fileutil=FileUtil,
dry_run=None,
):
""" Helper function for export that does the actual export
photo: PhotoInfo object
@@ -1755,6 +1779,9 @@ def export_photo(
album_keyword: boolean; if True, exports album names as keywords in metadata
person_keyword: boolean; if True, exports person names as keywords in metadata
keyword_template: list of strings; if provided use rendered template strings as keywords
export_db: export database instance compatible with ExportDB_ABC
fileutil: file util class compatible with FileUtilABC
dry_run: boolean; if True, doesn't actually export or update any files
returns list of path(s) of exported photo or None if photo was missing
"""
global VERBOSE
@@ -1791,8 +1818,10 @@ def export_photo(
verbose(f"Exporting {photo.filename} as {filename}")
if export_by_date:
date_created = photo.date.timetuple()
dest_path = create_path_by_date(dest, date_created)
date_created = DateTimeFormatter(photo.date)
dest_path = os.path.join(dest, date_created.year, date_created.mm, date_created.dd)
if not dry_run and not os.path.isdir(dest_path):
os.makedirs(dest_path)
dest_paths = [dest_path]
elif directory:
# got a directory template, render it and check results are valid
@@ -1808,7 +1837,7 @@ def export_photo(
dest_path = os.path.join(dest, dirname)
if not is_valid_filepath(dest_path, platform="auto"):
raise ValueError(f"Invalid file path: '{dest_path}'")
if not os.path.isdir(dest_path):
if not dry_run and not os.path.isdir(dest_path):
os.makedirs(dest_path)
dest_paths.append(dest_path)
@@ -1849,6 +1878,8 @@ def export_photo(
keyword_template=keyword_template,
update=update,
export_db=export_db,
fileutil=fileutil,
dry_run = dry_run,
)
results_exported.extend(export_results.exported)
@@ -1902,6 +1933,8 @@ def export_photo(
keyword_template=keyword_template,
update=update,
export_db=export_db,
fileutil=fileutil,
dry_run = dry_run,
)
results_exported.extend(export_results_edited.exported)

71
osxphotos/_export_db.py
View File

@@ -9,6 +9,7 @@ import pathlib
import sqlite3
import sys
from abc import ABC, abstractmethod
from io import StringIO
from sqlite3 import Error
from ._version import __version__
@@ -17,6 +18,7 @@ OSXPHOTOS_EXPORTDB_VERSION = "1.0"
class ExportDB_ABC(ABC):
""" abstract base class for ExportDB """
@abstractmethod
def get_uuid_for_file(self, filename):
pass
@@ -58,7 +60,7 @@ class ExportDB_ABC(ABC):
pass
@abstractmethod
def set_data(self, file, uuid, orig_stat, exif_stat, info_json, exif_json):
def set_data(self, filename, uuid, orig_stat, exif_stat, info_json, exif_json):
pass
@@ -95,7 +97,7 @@ class ExportDBNoOp(ExportDB_ABC):
def set_exifdata_for_file(self, uuid, exifdata):
pass
def set_data(self, file, uuid, orig_stat, exif_stat, info_json, exif_json):
def set_data(self, filename, uuid, orig_stat, exif_stat, info_json, exif_json):
pass
@@ -302,7 +304,7 @@ class ExportDB(ExportDB_ABC):
logging.debug(f"set_exifdata: {filename}, {exifdata}")
def set_data(self, file, uuid, orig_stat, exif_stat, info_json, exif_json):
def set_data(self, filename, uuid, orig_stat, exif_stat, info_json, exif_json):
""" sets all the data for file and uuid at once
calls set_uuid_for_file
set_info_for_uuid
@@ -310,8 +312,6 @@ class ExportDB(ExportDB_ABC):
set_stat_exif_for_file
set_exifdata_for_file
"""
filename = str(pathlib.Path(filename).relative_to(self._path)).lower()
self.set_uuid_for_file(filename, uuid)
self.set_info_for_uuid(uuid, info_json)
self.set_stat_orig_for_file(filename, orig_stat)
@@ -437,3 +437,64 @@ class ExportDB(ExportDB_ABC):
conn.commit()
except Error as e:
logging.warning(e)
class ExportDBInMemory(ExportDB):
""" In memory version of ExportDB
Copies the on-disk database into memory so it may be operated on without
modifying the on-disk verison
"""
def init(self, dbfile):
self._dbfile = dbfile
# _path is parent of the database
# all files referenced by get_/set_uuid_for_file will be converted to
# relative paths to this parent _path
# this allows the entire export tree to be moved to a new disk/location
# whilst preserving the UUID to filename mappping
self._path = pathlib.Path(dbfile).parent
self._conn = self._open_export_db(dbfile)
self._insert_run_info()
def _open_export_db(self, dbfile):
""" open export database and return a db connection
if dbfile does not exist, will create and initialize the database
returns: connection to the database
"""
if not os.path.isfile(dbfile):
logging.debug(f"dbfile {dbfile} doesn't exist, creating in memory version")
conn = self._get_db_connection()
if conn:
self._create_db_tables(conn)
else:
raise Exception("Error getting connection to in-memory database")
else:
logging.debug(f"dbfile {dbfile} exists, opening it and copying to memory")
try:
conn = sqlite3.connect(dbfile)
except Error as e:
logging.warning(e)
raise e
tempfile = StringIO()
for line in conn.iterdump():
tempfile.write("%s\n" % line)
conn.close()
tempfile.seek(0)
# Create a database in memory and import from tempfile
conn = sqlite3.connect(":memory:")
conn.cursor().executescript(tempfile.read())
conn.commit()
return conn
def _get_db_connection(self):
""" return db connection to in memory database """
try:
conn = sqlite3.connect(":memory:")
except Error as e:
logging.warning(e)
conn = None
return conn

54
osxphotos/_filecmp.py
View File

@@ -1,54 +0,0 @@
"""Utilities for comparing files
Modified from CPython/Lib/filecmp.py
Functions:
cmp_file(f1, s2) -> int
file_sig(f1) -> Tuple[int, int, float]
"""
import os
import stat
__all__ = ["cmp", "sig"]
def cmp_file(f1, s2):
"""Compare file f1 to signature s2.
Arguments:
f1 -- File name
s2 -- stats as returned by sig
Return value:
True if the files are the same, False otherwise.
This function uses a cache for past comparisons and the results,
with cache entries invalidated if their stat information
changes. The cache may be cleared by calling clear_cache().
"""
if not s2:
return False
s1 = _sig(os.stat(f1))
if s1[0] != stat.S_IFREG or s2[0] != stat.S_IFREG:
return False
if s1 == s2:
return True
return False
def _sig(st):
return (stat.S_IFMT(st.st_mode), st.st_size, st.st_mtime)
def file_sig(f1):
""" return os.stat signature for file f1 """
return _sig(os.stat(f1))

2
osxphotos/_version.py
View File

@@ -1,3 +1,3 @@
""" version info """
__version__ = "0.29.4"
__version__ = "0.29.5"

175
osxphotos/fileutil.py Normal file
View File

@@ -0,0 +1,175 @@
""" FileUtil class with methods for copy, hardlink, unlink, etc. """
import logging
import os
import pathlib
import stat
import subprocess
import sys
from abc import ABC, abstractmethod
class FileUtilABC(ABC):
""" Abstract base class for FileUtil """
@classmethod
@abstractmethod
def hardlink(cls, src, dest):
pass
@classmethod
@abstractmethod
def copy(cls, src, dest, norsrc=False):
pass
@classmethod
@abstractmethod
def unlink(cls, dest):
pass
@classmethod
@abstractmethod
def cmp_sig(cls, file1, file2):
pass
@classmethod
@abstractmethod
def file_sig(cls, file1):
pass
class FileUtilMacOS(FileUtilABC):
""" Various file utilities """
@classmethod
def hardlink(cls, src, dest):
""" Hardlinks a file from src path to dest path
src: source path as string
dest: destination path as string
Raises exception if linking fails or either path is None """
if src is None or dest is None:
raise ValueError("src and dest must not be None", src, dest)
if not os.path.isfile(src):
raise FileNotFoundError("src file does not appear to exist", src)
# if error on copy, subprocess will raise CalledProcessError
try:
os.link(src, dest)
except Exception as e:
logging.critical(f"os.link returned error: {e}")
raise e
@classmethod
def copy(cls, src, dest, norsrc=False):
""" Copies a file from src path to dest path
src: source path as string
dest: destination path as string
norsrc: (bool) if True, uses --norsrc flag with ditto so it will not copy
resource fork or extended attributes. May be useful on volumes that
don't work with extended attributes (likely only certain SMB mounts)
default is False
Uses ditto to perform copy; will silently overwrite dest if it exists
Raises exception if copy fails or either path is None """
if src is None or dest is None:
raise ValueError("src and dest must not be None", src, dest)
if not os.path.isfile(src):
raise FileNotFoundError("src file does not appear to exist", src)
if norsrc:
command = ["/usr/bin/ditto", "--norsrc", src, dest]
else:
command = ["/usr/bin/ditto", src, dest]
# if error on copy, subprocess will raise CalledProcessError
try:
result = subprocess.run(command, check=True, stderr=subprocess.PIPE)
except subprocess.CalledProcessError as e:
logging.critical(
f"ditto returned error: {e.returncode} {e.stderr.decode(sys.getfilesystemencoding()).rstrip()}"
)
raise e
return result.returncode
@classmethod
def unlink(cls, filepath):
""" unlink filepath; if it's pathlib.Path, use Path.unlink, otherwise use os.unlink """
if isinstance(filepath, pathlib.Path):
filepath.unlink()
else:
os.unlink(filepath)
@classmethod
def cmp_sig(cls, f1, s2):
"""Compare file f1 to signature s2.
Arguments:
f1 -- File name
s2 -- stats as returned by sig
Return value:
True if the files are the same, False otherwise.
"""
if not s2:
return False
s1 = cls._sig(os.stat(f1))
if s1[0] != stat.S_IFREG or s2[0] != stat.S_IFREG:
return False
if s1 == s2:
return True
return False
@classmethod
def file_sig(cls, f1):
""" return os.stat signature for file f1 """
return cls._sig(os.stat(f1))
@staticmethod
def _sig(st):
return (stat.S_IFMT(st.st_mode), st.st_size, st.st_mtime)
class FileUtil(FileUtilMacOS):
""" Various file utilities """
pass
class FileUtilNoOp(FileUtil):
""" No-Op implementation of FileUtil for testing / dry-run mode
all methods with exception of cmp_sig and file_cmp are no-op
cmp_sig functions as FileUtil.cmp_sig does
file_cmp returns mock data
"""
@staticmethod
def noop(*args):
pass
verbose = noop
def __new__(cls, verbose=None):
if verbose:
if callable(verbose):
cls.verbose = verbose
else:
raise ValueError(f"verbose {verbose} not callable")
return super(FileUtilNoOp, cls).__new__(cls)
@classmethod
def hardlink(cls, src, dest):
cls.verbose(f"hardlink: {src} {dest}")
@classmethod
def copy(cls, src, dest, norsrc=False):
cls.verbose(f"copy: {src} {dest}")
@classmethod
def unlink(cls, dest):
cls.verbose(f"unlink: {dest}")
@classmethod
def file_sig(cls, file1):
cls.verbose(f"file_sig: {file1}")
return (42, 42, 42)

265
osxphotos/photoinfo/_photoinfo_export.py
View File

@@ -1,4 +1,13 @@
""" export methods for PhotoInfo """
""" Export methods for PhotoInfo
The following methods are defined and must be imported into PhotoInfo as instance methods:
export
export2
_export_photo
_write_exif_data
_exiftool_json_sidecar
_xmp_sidecar
_write_sidecar
"""
# TODO: should this be its own PhotoExporter class?
@@ -9,10 +18,12 @@ import logging
import os
import pathlib
import re
import tempfile
from collections import namedtuple # pylint: disable=syntax-error
from mako.template import Template
from .._applescript import AppleScript
from .._constants import (
_MAX_IPTC_KEYWORD_LEN,
_OSXPHOTOS_NONE_SENTINEL,
@@ -20,21 +31,131 @@ from .._constants import (
_UNKNOWN_PERSON,
_XMP_TEMPLATE_NAME,
)
from ..exiftool import ExifTool
from .._export_db import ExportDBNoOp
from .._filecmp import cmp_file, file_sig
from ..utils import (
_copy_file,
_export_photo_uuid_applescript,
_hardlink_file,
dd_to_dms_str,
)
from ..exiftool import ExifTool
from ..fileutil import FileUtil
from ..utils import dd_to_dms_str
ExportResults = namedtuple(
"ExportResults", ["exported", "new", "updated", "skipped", "exif_updated"]
)
# _export_photo_uuid_applescript is not a class method, don't import this into PhotoInfo
def _export_photo_uuid_applescript(
uuid,
dest,
filestem=None,
original=True,
edited=False,
live_photo=False,
timeout=120,
burst=False,
dry_run=False,
):
""" Export photo to dest path using applescript to control Photos
If photo is a live photo, exports both the photo and associated .mov file
uuid: UUID of photo to export
dest: destination path to export to
filestem: (string) if provided, exported filename will be named stem.ext
where ext is extension of the file exported by photos (e.g. .jpeg, .mov, etc)
If not provided, file will be named with whatever name Photos uses
If filestem.ext exists, it wil be overwritten
original: (boolean) if True, export original image; default = True
edited: (boolean) if True, export edited photo; default = False
If photo not edited and edited=True, will still export the original image
caller must verify image has been edited
*Note*: must be called with either edited or original but not both,
will raise error if called with both edited and original = True
live_photo: (boolean) if True, export associated .mov live photo; default = False
timeout: timeout value in seconds; export will fail if applescript run time exceeds timeout
burst: (boolean) set to True if file is a burst image to avoid Photos export error
dry_run: (boolean) set to True to run in "dry run" mode which will download file but not actually copy to destination
Returns: list of paths to exported file(s) or None if export failed
Note: For Live Photos, if edited=True, will export a jpeg but not the movie, even if photo
has not been edited. This is due to how Photos Applescript interface works.
"""
# setup the applescript to do the export
export_scpt = AppleScript(
"""
on export_by_uuid(theUUID, thePath, original, edited, theTimeOut)
tell application "Photos"
set thePath to thePath
set theItem to media item id theUUID
set theFilename to filename of theItem
set itemList to {theItem}
if original then
with timeout of theTimeOut seconds
export itemList to POSIX file thePath with using originals
end timeout
end if
if edited then
with timeout of theTimeOut seconds
export itemList to POSIX file thePath
end timeout
end if
return theFilename
end tell
end export_by_uuid
"""
)
dest = pathlib.Path(dest)
if not dest.is_dir:
raise ValueError(f"dest {dest} must be a directory")
if not original ^ edited:
raise ValueError(f"edited or original must be True but not both")
tmpdir = tempfile.TemporaryDirectory(prefix="osxphotos_")
# export original
filename = None
try:
filename = export_scpt.call(
"export_by_uuid", uuid, tmpdir.name, original, edited, timeout
)
except Exception as e:
logging.warning(f"Error exporting uuid {uuid}: {e}")
return None
if filename is not None:
# need to find actual filename as sometimes Photos renames JPG to jpeg on export
# may be more than one file exported (e.g. if Live Photo, Photos exports both .jpeg and .mov)
# TemporaryDirectory will cleanup on return
filename_stem = pathlib.Path(filename).stem
files = glob.glob(os.path.join(tmpdir.name, "*"))
exported_paths = []
for fname in files:
path = pathlib.Path(fname)
if len(files) > 1 and not live_photo and path.suffix.lower() == ".mov":
# it's the .mov part of live photo but not requested, so don't export
logging.debug(f"Skipping live photo file {path}")
continue
if len(files) > 1 and burst and path.stem != filename_stem:
# skip any burst photo that's not the one we asked for
logging.debug(f"Skipping burst photo file {path}")
continue
if filestem:
# rename the file based on filestem, keeping original extension
dest_new = dest / f"{filestem}{path.suffix}"
else:
# use the name Photos provided
dest_new = dest / path.name
logging.debug(f"exporting {path} to dest_new: {dest_new}")
if not dry_run:
FileUtil.copy(str(path), str(dest_new))
exported_paths.append(str(dest_new))
return exported_paths
else:
return None
def export(
self,
dest,
@@ -135,8 +256,10 @@ def export2(
keyword_template=None,
update=False,
export_db=None,
fileutil=FileUtil,
dry_run=False,
):
""" export photo
""" export photo, like export but with update and dry_run options
dest: must be valid destination path (or exception raised)
filename: (optional): name of exported picture; if not provided, will use current filename
**NOTE**: if provided, user must ensure file extension (suffix) is correct.
@@ -171,12 +294,17 @@ def export2(
not export the photo if the current version already exists in the destination
export_db: (ExportDB_ABC); instance of a class that conforms to ExportDB_ABC with methods
for getting/setting data related to exported files to compare update state
fileutil: (FileUtilABC); class that conforms to FileUtilABC with various file utilities
dry_run: (boolean, default=False); set to True to run in "dry run" mode
Returns: ExportResults namedtuple with fields: exported, new, updated, skipped
where each field is a list of file paths
Note: to use dry run mode, you must set dry_run=True and also pass in memory version of export_db,
and no-op fileutil (e.g. ExportDBInMemory and FileUtilNoOp)
"""
# if update, caller may pass function refs to get/set uuid for file being exported
# and for setting/getting the PhotoInfo json info for an exported file
# when called from export(), won't get an export_db, so use no-op version
if export_db is None:
export_db = ExportDBNoOp()
@@ -212,7 +340,7 @@ def export2(
# verify destination is a valid path
if dest is None:
raise ValueError("Destination must not be None")
elif not os.path.isdir(dest):
elif not dry_run and not os.path.isdir(dest):
raise FileNotFoundError("Invalid path passed to export")
if filename and len(filename) == 1:
@@ -279,10 +407,6 @@ def export2(
count += 1
dest = dest.parent / f"{dest_new}{dest.suffix}"
# TODO: need way to check if DB is missing, try to find the right photo anyway by seeing if they're the same and then updating
# move the checks into "if not use_photos_export" block below
# if use_photos_export is True then we'll export wether destination exists or not
# if overwrite==False and #increment==False, export should fail if file exists
if dest.exists() and not update and not overwrite and not increment:
raise FileExistsError(
@@ -330,7 +454,7 @@ def export2(
dest_uuid = self.uuid
export_db.set_uuid_for_file(dest, self.uuid)
export_db.set_info_for_uuid(self.uuid, self.json())
export_db.set_stat_orig_for_file(dest, file_sig(dest))
export_db.set_stat_orig_for_file(dest, fileutil.file_sig(dest))
export_db.set_stat_exif_for_file(dest, (None, None, None))
export_db.set_exifdata_for_file(dest, None)
if dest_uuid != self.uuid:
@@ -360,7 +484,7 @@ def export2(
found_match = True
export_db.set_uuid_for_file(file_, self.uuid)
export_db.set_info_for_uuid(self.uuid, self.json())
export_db.set_stat_orig_for_file(dest, file_sig(dest))
export_db.set_stat_orig_for_file(dest, fileutil.file_sig(dest))
export_db.set_stat_exif_for_file(dest, (None, None, None))
export_db.set_exifdata_for_file(dest, None)
break
@@ -392,6 +516,7 @@ def export2(
no_xattr,
export_as_hardlink,
exiftool,
fileutil,
)
exported_files = results.exported
update_new_files = results.new
@@ -416,6 +541,7 @@ def export2(
no_xattr,
export_as_hardlink,
exiftool,
fileutil,
)
exported_files.extend(results.exported)
update_new_files.extend(results.new)
@@ -440,6 +566,7 @@ def export2(
no_xattr,
export_as_hardlink,
exiftool,
fileutil,
)
exported_files.extend(results.exported)
update_new_files.extend(results.new)
@@ -471,6 +598,7 @@ def export2(
live_photo=live_photo,
timeout=timeout,
burst=self.burst,
dry_run=dry_run,
)
else:
# export original version and not edited
@@ -484,6 +612,7 @@ def export2(
live_photo=live_photo,
timeout=timeout,
burst=self.burst,
dry_run=dry_run,
)
if exported is not None:
@@ -504,11 +633,12 @@ def export2(
use_persons_as_keywords=use_persons_as_keywords,
keyword_template=keyword_template,
)
try:
self._write_sidecar(sidecar_filename, sidecar_str)
except Exception as e:
logging.warning(f"Error writing json sidecar to {sidecar_filename}")
raise e
if not dry_run:
try:
self._write_sidecar(sidecar_filename, sidecar_str)
except Exception as e:
logging.warning(f"Error writing json sidecar to {sidecar_filename}")
raise e
if sidecar_xmp:
logging.debug("writing xmp_sidecar")
@@ -518,11 +648,12 @@ def export2(
use_persons_as_keywords=use_persons_as_keywords,
keyword_template=keyword_template,
)
try:
self._write_sidecar(sidecar_filename, sidecar_str)
except Exception as e:
logging.warning(f"Error writing xmp sidecar to {sidecar_filename}")
raise e
if not dry_run:
try:
self._write_sidecar(sidecar_filename, sidecar_str)
except Exception as e:
logging.warning(f"Error writing xmp sidecar to {sidecar_filename}")
raise e
# if exiftool, write the metadata
if update:
@@ -552,12 +683,13 @@ def export2(
# didn't have old data, assume we need to write it
# or files were different
logging.debug(f"No exifdata for {exported_file}, writing it")
self._write_exif_data(
exported_file,
use_albums_as_keywords=use_albums_as_keywords,
use_persons_as_keywords=use_persons_as_keywords,
keyword_template=keyword_template,
)
if not dry_run:
self._write_exif_data(
exported_file,
use_albums_as_keywords=use_albums_as_keywords,
use_persons_as_keywords=use_persons_as_keywords,
keyword_template=keyword_template,
)
export_db.set_exifdata_for_file(
exported_file,
self._exiftool_json_sidecar(
@@ -566,17 +698,20 @@ def export2(
keyword_template=keyword_template,
),
)
export_db.set_stat_exif_for_file(exported_file, file_sig(exported_file))
export_db.set_stat_exif_for_file(
exported_file, fileutil.file_sig(exported_file)
)
exif_files_updated.append(exported_file)
elif exiftool and exif_files:
for exported_file in exif_files:
logging.debug(f"Writing exif data to {exported_file}")
self._write_exif_data(
exported_file,
use_albums_as_keywords=use_albums_as_keywords,
use_persons_as_keywords=use_persons_as_keywords,
keyword_template=keyword_template,
)
if not dry_run:
self._write_exif_data(
exported_file,
use_albums_as_keywords=use_albums_as_keywords,
use_persons_as_keywords=use_persons_as_keywords,
keyword_template=keyword_template,
)
export_db.set_exifdata_for_file(
exported_file,
self._exiftool_json_sidecar(
@@ -585,7 +720,9 @@ def export2(
keyword_template=keyword_template,
),
)
export_db.set_stat_exif_for_file(exported_file, file_sig(exported_file))
export_db.set_stat_exif_for_file(
exported_file, fileutil.file_sig(exported_file)
)
exif_files_updated.append(exported_file)
return ExportResults(
@@ -607,6 +744,7 @@ def _export_photo(
no_xattr,
export_as_hardlink,
exiftool,
fileutil=FileUtil,
):
""" Helper function for export()
Does the actual copy or hardlink taking the appropriate
@@ -621,6 +759,7 @@ def _export_photo(
no_xattr: don't copy extended attributes
export_as_hardlink: bool
exiftool: bool
fileutil: FileUtil class that conforms to fileutil.FileUtilABC
Returns: ExportResults
"""
@@ -637,12 +776,13 @@ def _export_photo(
# not update, do the the hardlink
if overwrite and dest.exists():
# need to remove the destination first
dest.unlink()
# dest.unlink()
fileutil.unlink(dest)
logging.debug(f"Not update: export_as_hardlink linking file {src} {dest}")
_hardlink_file(src, dest)
fileutil.hardlink(src, dest)
export_db.set_uuid_for_file(dest_str, self.uuid)
export_db.set_info_for_uuid(self.uuid, self.json())
export_db.set_stat_orig_for_file(dest_str, file_sig(dest_str))
export_db.set_stat_orig_for_file(dest_str, fileutil.file_sig(dest_str))
export_db.set_stat_exif_for_file(dest_str, (None, None, None))
export_db.set_exifdata_for_file(dest_str, None)
exported_files.append(dest_str)
@@ -657,11 +797,12 @@ def _export_photo(
logging.debug(
f"Update: removing existing file prior to export_as_hardlink {src} {dest}"
)
dest.unlink()
_hardlink_file(src, dest)
# dest.unlink()
fileutil.unlink(dest)
fileutil.hardlink(src, dest)
export_db.set_uuid_for_file(dest_str, self.uuid)
export_db.set_info_for_uuid(self.uuid, self.json())
export_db.set_stat_orig_for_file(dest_str, file_sig(dest_str))
export_db.set_stat_orig_for_file(dest_str, fileutil.file_sig(dest_str))
export_db.set_stat_exif_for_file(dest_str, (None, None, None))
export_db.set_exifdata_for_file(dest_str, None)
update_updated_files.append(dest_str)
@@ -671,10 +812,10 @@ def _export_photo(
logging.debug(
f"Update: exporting new file with export_as_hardlink {src} {dest}"
)
_hardlink_file(src, dest)
fileutil.hardlink(src, dest)
export_db.set_uuid_for_file(dest_str, self.uuid)
export_db.set_info_for_uuid(self.uuid, self.json())
export_db.set_stat_orig_for_file(dest_str, file_sig(dest_str))
export_db.set_stat_orig_for_file(dest_str, fileutil.file_sig(dest_str))
export_db.set_stat_exif_for_file(dest_str, (None, None, None))
export_db.set_exifdata_for_file(dest_str, None)
exported_files.append(dest_str)
@@ -684,12 +825,13 @@ def _export_photo(
# not update, do the the copy
if overwrite and dest.exists():
# need to remove the destination first
dest.unlink()
# dest.unlink()
fileutil.unlink(dest)
logging.debug(f"Not update: copying file {src} {dest}")
_copy_file(src, dest_str, norsrc=no_xattr)
fileutil.copy(src, dest_str, norsrc=no_xattr)
export_db.set_uuid_for_file(dest_str, self.uuid)
export_db.set_info_for_uuid(self.uuid, self.json())
export_db.set_stat_orig_for_file(dest_str, file_sig(dest_str))
export_db.set_stat_orig_for_file(dest_str, fileutil.file_sig(dest_str))
export_db.set_stat_exif_for_file(dest_str, (None, None, None))
export_db.set_exifdata_for_file(dest_str, None)
exported_files.append(dest_str)
@@ -704,12 +846,12 @@ def _export_photo(
logging.debug(f"Update: skipping identifical original files {src} {dest}")
# call set_stat because code can reach this spot if no export DB but exporting a RAW or live photo
# potentially re-writes the data in the database but ensures database is complete
export_db.set_stat_orig_for_file(dest_str, file_sig(dest_str))
export_db.set_stat_orig_for_file(dest_str, fileutil.file_sig(dest_str))
update_skipped_files.append(dest_str)
elif (
dest_exists
and exiftool
and cmp_file(dest_str, export_db.get_stat_exif_for_file(dest_str))
and fileutil.cmp_sig(dest_str, export_db.get_stat_exif_for_file(dest_str))
and not dest.samefile(src)
):
# destination exists but is identical
@@ -720,11 +862,12 @@ def _export_photo(
logging.debug(f"Update: removing existing file prior to copy {src} {dest}")
stat_src = os.stat(src)
stat_dest = os.stat(dest)
dest.unlink()
_copy_file(src, dest_str, norsrc=no_xattr)
# dest.unlink()
fileutil.unlink(dest)
fileutil.copy(src, dest_str, norsrc=no_xattr)
export_db.set_uuid_for_file(dest_str, self.uuid)
export_db.set_info_for_uuid(self.uuid, self.json())
export_db.set_stat_orig_for_file(dest_str, file_sig(dest_str))
export_db.set_stat_orig_for_file(dest_str, fileutil.file_sig(dest_str))
export_db.set_stat_exif_for_file(dest_str, (None, None, None))
export_db.set_exifdata_for_file(dest_str, None)
exported_files.append(dest_str)
@@ -732,10 +875,10 @@ def _export_photo(
else:
# destination doesn't exist, copy the file
logging.debug(f"Update: copying new file {src} {dest}")
_copy_file(src, dest_str, norsrc=no_xattr)
fileutil.copy(src, dest_str, norsrc=no_xattr)
export_db.set_uuid_for_file(dest_str, self.uuid)
export_db.set_info_for_uuid(self.uuid, self.json())
export_db.set_stat_orig_for_file(dest_str, file_sig(dest_str))
export_db.set_stat_orig_for_file(dest_str, fileutil.file_sig(dest_str))
export_db.set_stat_exif_for_file(dest_str, (None, None, None))
export_db.set_exifdata_for_file(dest_str, None)
exported_files.append(dest_str)

187
osxphotos/utils.py
View File

@@ -18,7 +18,7 @@ import CoreServices
import objc
from Foundation import *
from osxphotos._applescript import AppleScript
from .fileutil import FileUtil
_DEBUG = False
@@ -118,60 +118,6 @@ def _dd_to_dms(dd):
return int(deg_), int(min_), sec_
def _hardlink_file(src, dest):
""" Hardlinks a file from src path to dest path
src: source path as string
dest: destination path as string
Raises exception if linking fails or either path is None """
if src is None or dest is None:
raise ValueError("src and dest must not be None", src, dest)
if not os.path.isfile(src):
raise FileNotFoundError("src file does not appear to exist", src)
# if error on copy, subprocess will raise CalledProcessError
try:
os.link(src, dest)
except Exception as e:
logging.critical(f"os.link returned error: {e}")
raise e
def _copy_file(src, dest, norsrc=False):
""" Copies a file from src path to dest path
src: source path as string
dest: destination path as string
norsrc: (bool) if True, uses --norsrc flag with ditto so it will not copy
resource fork or extended attributes. May be useful on volumes that
don't work with extended attributes (likely only certain SMB mounts)
default is False
Uses ditto to perform copy; will silently overwrite dest if it exists
Raises exception if copy fails or either path is None """
if src is None or dest is None:
raise ValueError("src and dest must not be None", src, dest)
if not os.path.isfile(src):
raise FileNotFoundError("src file does not appear to exist", src)
if norsrc:
command = ["/usr/bin/ditto", "--norsrc", src, dest]
else:
command = ["/usr/bin/ditto", src, dest]
# if error on copy, subprocess will raise CalledProcessError
try:
result = subprocess.run(command, check=True, stderr=subprocess.PIPE)
except subprocess.CalledProcessError as e:
logging.critical(
f"ditto returned error: {e.returncode} {e.stderr.decode(sys.getfilesystemencoding()).rstrip()}"
)
raise e
return result.returncode
def dd_to_dms_str(lat, lon):
""" convert latitude, longitude in degrees to degrees, minutes, seconds as string """
""" lat: latitude in degrees """
@@ -308,24 +254,6 @@ def list_photo_libraries():
return lib_list
def create_path_by_date(dest, dt):
""" Creates a path in dest folder in form dest/YYYY/MM/DD/
dest: valid path as str
dt: datetime.timetuple() object
Checks to see if path exists, if it does, do nothing and return path
If path does not exist, creates it and returns path"""
if not os.path.isdir(dest):
raise FileNotFoundError(f"dest {dest} must be valid path")
yyyy, mm, dd = dt[0:3]
yyyy = str(yyyy).zfill(4)
mm = str(mm).zfill(2)
dd = str(dd).zfill(2)
new_dest = os.path.join(dest, yyyy, mm, dd)
if not os.path.isdir(new_dest):
os.makedirs(new_dest)
return new_dest
def get_preferred_uti_extension(uti):
""" get preferred extension for a UTI type
uti: UTI str, e.g. 'public.jpeg'
@@ -365,117 +293,6 @@ def findfiles(pattern, path_):
# open_scpt.run()
def _export_photo_uuid_applescript(
uuid,
dest,
filestem=None,
original=True,
edited=False,
live_photo=False,
timeout=120,
burst=False,
):
""" Export photo to dest path using applescript to control Photos
If photo is a live photo, exports both the photo and associated .mov file
uuid: UUID of photo to export
dest: destination path to export to
filestem: (string) if provided, exported filename will be named stem.ext
where ext is extension of the file exported by photos (e.g. .jpeg, .mov, etc)
If not provided, file will be named with whatever name Photos uses
If filestem.ext exists, it wil be overwritten
original: (boolean) if True, export original image; default = True
edited: (boolean) if True, export edited photo; default = False
If photo not edited and edited=True, will still export the original image
caller must verify image has been edited
*Note*: must be called with either edited or original but not both,
will raise error if called with both edited and original = True
live_photo: (boolean) if True, export associated .mov live photo; default = False
timeout: timeout value in seconds; export will fail if applescript run time exceeds timeout
burst: (boolean) set to True if file is a burst image to avoid Photos export error
Returns: list of paths to exported file(s) or None if export failed
Note: For Live Photos, if edited=True, will export a jpeg but not the movie, even if photo
has not been edited. This is due to how Photos Applescript interface works.
"""
# setup the applescript to do the export
export_scpt = AppleScript(
"""
on export_by_uuid(theUUID, thePath, original, edited, theTimeOut)
tell application "Photos"
set thePath to thePath
set theItem to media item id theUUID
set theFilename to filename of theItem
set itemList to {theItem}
if original then
with timeout of theTimeOut seconds
export itemList to POSIX file thePath with using originals
end timeout
end if
if edited then
with timeout of theTimeOut seconds
export itemList to POSIX file thePath
end timeout
end if
return theFilename
end tell
end export_by_uuid
"""
)
dest = pathlib.Path(dest)
if not dest.is_dir:
raise ValueError(f"dest {dest} must be a directory")
if not original ^ edited:
raise ValueError(f"edited or original must be True but not both")
tmpdir = tempfile.TemporaryDirectory(prefix="osxphotos_")
# export original
filename = None
try:
filename = export_scpt.call(
"export_by_uuid", uuid, tmpdir.name, original, edited, timeout
)
except Exception as e:
logging.warning("Error exporting uuid %s: %s" % (uuid, str(e)))
return None
if filename is not None:
# need to find actual filename as sometimes Photos renames JPG to jpeg on export
# may be more than one file exported (e.g. if Live Photo, Photos exports both .jpeg and .mov)
# TemporaryDirectory will cleanup on return
filename_stem = pathlib.Path(filename).stem
files = glob.glob(os.path.join(tmpdir.name, "*"))
exported_paths = []
for fname in files:
path = pathlib.Path(fname)
if len(files) > 1 and not live_photo and path.suffix.lower() == ".mov":
# it's the .mov part of live photo but not requested, so don't export
logging.debug(f"Skipping live photo file {path}")
continue
if len(files) > 1 and burst and path.stem != filename_stem:
# skip any burst photo that's not the one we asked for
logging.debug(f"Skipping burst photo file {path}")
continue
if filestem:
# rename the file based on filestem, keeping original extension
dest_new = dest / f"{filestem}{path.suffix}"
else:
# use the name Photos provided
dest_new = dest / path.name
logging.debug(f"exporting {path} to dest_new: {dest_new}")
_copy_file(str(path), str(dest_new))
exported_paths.append(str(dest_new))
return exported_paths
else:
return None
def _open_sql_file(dbname):
""" opens sqlite file dbname in read-only mode
returns tuple of (connection, cursor) """
@@ -527,7 +344,7 @@ def _db_is_locked(dbname):
# except KeyError:
# uuid_str = None
# return uuid_str
# def set_uuid_for_file(filepath, uuid):
# """ sets the UUID associated with an exported file
# filepath: path to exported photo