elodie._batch() - Code Metrics - Inspection of "Add a plugin to upload photos to Google Photos (#3..." - jmathai/elodie - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 9e42ed...b2d36d )

by Jaisen

created 2019-07-12 07:40 UTC

elodie._batch() A

↳ Parent: elodie

Complexity

Conditions

Size

Total Lines	9
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	7
nop	1
dl	0
loc	9
rs	10
c	0
b	0
f	0

#!/usr/bin/env python

from __future__ import print_function
import os
import re
import sys
from datetime import datetime

import click
from send2trash import send2trash

# Verify that external dependencies are present first, so the user gets a
# more user-friendly error instead of an ImportError traceback.
from elodie.dependencies import verify_dependencies
if not verify_dependencies():
    sys.exit(1)

from elodie import constants
from elodie import geolocation
from elodie import log
from elodie.compatability import _decode
from elodie.filesystem import FileSystem
from elodie.localstorage import Db
from elodie.media.base import Base, get_all_subclasses
from elodie.media.media import Media
from elodie.media.text import Text
from elodie.media.audio import Audio
from elodie.media.photo import Photo
from elodie.media.video import Video
from elodie.plugins.plugins import Plugins
from elodie.result import Result


FILESYSTEM = FileSystem()


def import_file(_file, destination, album_from_folder, trash, allow_duplicates):
    
    _file = _decode(_file)
    destination = _decode(destination)

    """Set file metadata and move it to destination.
    """
    if not os.path.exists(_file):
        log.warn('Could not find %s' % _file)
        log.all('{"source":"%s", "error_msg":"Could not find %s"}' %
                  (_file, _file))
        return
    # Check if the source, _file, is a child folder within destination
    elif destination.startswith(os.path.abspath(os.path.dirname(_file))+os.sep):
        log.all('{"source": "%s", "destination": "%s", "error_msg": "Source cannot be in destination"}' % (
            _file, destination))
        return


    media = Media.get_class_by_file(_file, get_all_subclasses())
    if not media:
        log.warn('Not a supported file (%s)' % _file)
        log.all('{"source":"%s", "error_msg":"Not a supported file"}' % _file)
        return

    if album_from_folder:
        media.set_album_from_folder()

    dest_path = FILESYSTEM.process_file(_file, destination,
        media, allowDuplicate=allow_duplicates, move=False)
    if dest_path:
        log.all('%s -> %s' % (_file, dest_path))
    if trash:
        send2trash(_file)

    return dest_path or None

@click.command('batch')
@click.option('--debug', default=False, is_flag=True,
              help='Override the value in constants.py with True.')
def _batch(debug):
    """Run batch() for all plugins.
    """
    constants.debug = debug
    plugins = Plugins()
    plugins.run_batch()
       

@click.command('import')
@click.option('--destination', type=click.Path(file_okay=False),
              required=True, help='Copy imported files into this directory.')
@click.option('--source', type=click.Path(file_okay=False),
              help='Import files from this directory, if specified.')
@click.option('--file', type=click.Path(dir_okay=False),
              help='Import this file, if specified.')
@click.option('--album-from-folder', default=False, is_flag=True,
              help="Use images' folders as their album names.")
@click.option('--trash', default=False, is_flag=True,
              help='After copying files, move the old files to the trash.')
@click.option('--allow-duplicates', default=False, is_flag=True,
              help='Import the file even if it\'s already been imported.')
@click.option('--debug', default=False, is_flag=True,
              help='Override the value in constants.py with True.')
@click.argument('paths', nargs=-1, type=click.Path())
def _import(destination, source, file, album_from_folder, trash, allow_duplicates, debug, paths):
    """Import files or directories by reading their EXIF and organizing them accordingly.
    """
    constants.debug = debug
    has_errors = False
    result = Result()

    destination = _decode(destination)
    destination = os.path.abspath(os.path.expanduser(destination))

    files = set()
    paths = set(paths)
    if source:
        source = _decode(source)
        paths.add(source)
    if file:
        paths.add(file)
    for path in paths:
        path = os.path.expanduser(path)
        if os.path.isdir(path):
            files.update(FILESYSTEM.get_all_files(path, None))
        else:
            files.add(path)

    for current_file in files:
        dest_path = import_file(current_file, destination, album_from_folder,
                    trash, allow_duplicates)
        result.append((current_file, dest_path))
        has_errors = has_errors is True or not dest_path

    result.write()

    if has_errors:
        sys.exit(1)


@click.command('generate-db')
@click.option('--source', type=click.Path(file_okay=False),
              required=True, help='Source of your photo library.')
@click.option('--debug', default=False, is_flag=True,
              help='Override the value in constants.py with True.')
def _generate_db(source, debug):
    """Regenerate the hash.json database which contains all of the sha256 signatures of media files. The hash.json file is located at ~/.elodie/.
    """
    constants.debug = debug
    result = Result()
    source = os.path.abspath(os.path.expanduser(source))

    if not os.path.isdir(source):
        log.error('Source is not a valid directory %s' % source)
        sys.exit(1)
        
    db = Db()
    db.backup_hash_db()
    db.reset_hash_db()

    for current_file in FILESYSTEM.get_all_files(source):
        result.append((current_file, True))
        db.add_hash(db.checksum(current_file), current_file)
        log.progress()
    
    db.update_hash_db()
    log.progress('', True)
    result.write()

@click.command('verify')
@click.option('--debug', default=False, is_flag=True,
              help='Override the value in constants.py with True.')
def _verify(debug):
    constants.debug = debug
    result = Result()
    db = Db()
    for checksum, file_path in db.all():
        if not os.path.isfile(file_path):
            result.append((file_path, False))
            log.progress('x')
            continue

        actual_checksum = db.checksum(file_path)
        if checksum == actual_checksum:
            result.append((file_path, True))
            log.progress()
        else:
            result.append((file_path, False))
            log.progress('x')

    log.progress('', True)
    result.write()


def update_location(media, file_path, location_name):
    """Update location exif metadata of media.
    """
    location_coords = geolocation.coordinates_by_name(location_name)

    if location_coords and 'latitude' in location_coords and \
            'longitude' in location_coords:
        location_status = media.set_location(location_coords[
            'latitude'], location_coords['longitude'])
        if not location_status:
            log.error('Failed to update location')
            log.all(('{"source":"%s",' % file_path,
                       '"error_msg":"Failed to update location"}'))
            sys.exit(1)
    return True


def update_time(media, file_path, time_string):
    """Update time exif metadata of media.
    """
    time_format = '%Y-%m-%d %H:%M:%S'
    if re.match(r'^\d{4}-\d{2}-\d{2}$', time_string):
        time_string = '%s 00:00:00' % time_string
    elif re.match(r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}\d{2}$', time_string):
        msg = ('Invalid time format. Use YYYY-mm-dd hh:ii:ss or YYYY-mm-dd')
        log.error(msg)
        log.all('{"source":"%s", "error_msg":"%s"}' % (file_path, msg))
        sys.exit(1)

    time = datetime.strptime(time_string, time_format)
    media.set_date_taken(time)
    return True


@click.command('update')
@click.option('--album', help='Update the image album.')
@click.option('--location', help=('Update the image location. Location '
                                  'should be the name of a place, like "Las '
                                  'Vegas, NV".'))
@click.option('--time', help=('Update the image time. Time should be in '
                              'YYYY-mm-dd hh:ii:ss or YYYY-mm-dd format.'))
@click.option('--title', help='Update the image title.')
@click.option('--debug', default=False, is_flag=True,
              help='Override the value in constants.py with True.')
@click.argument('paths', nargs=-1,
                required=True)
def _update(album, location, time, title, paths, debug):
    """Update a file's EXIF. Automatically modifies the file's location and file name accordingly.
    """
    constants.debug = debug
    has_errors = False
    result = Result()

    files = set()
    for path in paths:
        path = os.path.expanduser(path)
        if os.path.isdir(path):
            files.update(FILESYSTEM.get_all_files(path, None))
        else:
            files.add(path)

    for current_file in files:
        if not os.path.exists(current_file):
            has_errors = True
            result.append((current_file, False))
            log.warn('Could not find %s' % current_file)
            log.all('{"source":"%s", "error_msg":"Could not find %s"}' %
                      (current_file, current_file))
            continue

        current_file = os.path.expanduser(current_file)

        # The destination folder structure could contain any number of levels
        #  So we calculate that and traverse up the tree.
        # '/path/to/file/photo.jpg' -> '/path/to/file' ->
        #  ['path','to','file'] -> ['path','to'] -> '/path/to'
        current_directory = os.path.dirname(current_file)
        destination_depth = -1 * len(FILESYSTEM.get_folder_path_definition())
        destination = os.sep.join(
                          os.path.normpath(
                              current_directory
                          ).split(os.sep)[:destination_depth]
                      )

        media = Media.get_class_by_file(current_file, get_all_subclasses())
        if not media:
            continue

        updated = False
        if location:
            update_location(media, current_file, location)
            updated = True
        if time:
            update_time(media, current_file, time)
            updated = True
        if album:
            media.set_album(album)
            updated = True

        # Updating a title can be problematic when doing it 2+ times on a file.
        # You would end up with img_001.jpg -> img_001-first-title.jpg ->
        # img_001-first-title-second-title.jpg.
        # To resolve that we have to track the prior title (if there was one.
        # Then we massage the updated_media's metadata['base_name'] to remove
        # the old title.
        # Since FileSystem.get_file_name() relies on base_name it will properly
        #  rename the file by updating the title instead of appending it.
        remove_old_title_from_name = False
        if title:
            # We call get_metadata() to cache it before making any changes
            metadata = media.get_metadata()
            title_update_status = media.set_title(title)
            original_title = metadata['title']
            if title_update_status and original_title:
                # @TODO: We should move this to a shared method since
                # FileSystem.get_file_name() does it too.
                original_title = re.sub(r'\W+', '-', original_title.lower())
                original_base_name = metadata['base_name']
                remove_old_title_from_name = True
            updated = True

        if updated:
            updated_media = Media.get_class_by_file(current_file,
                                                    get_all_subclasses())
            # See comments above on why we have to do this when titles
            # get updated.
            if remove_old_title_from_name and len(original_title) > 0:

                updated_media.get_metadata()
                updated_media.set_metadata_basename(
                    original_base_name.replace('-%s' % original_title, ''))


            dest_path = FILESYSTEM.process_file(current_file, destination,
                updated_media, move=True, allowDuplicate=True)
            log.info(u'%s -> %s' % (current_file, dest_path))
            log.all('{"source":"%s", "destination":"%s"}' % (current_file,
                                                               dest_path))
            # If the folder we moved the file out of or its parent are empty
            # we delete it.
            FILESYSTEM.delete_directory_if_empty(os.path.dirname(current_file))
            FILESYSTEM.delete_directory_if_empty(
                os.path.dirname(os.path.dirname(current_file)))
            result.append((current_file, dest_path))
            # Trip has_errors to False if it's already False or dest_path is.
            has_errors = has_errors is True or not dest_path
        else:
            has_errors = False
            result.append((current_file, False))

    result.write()
    
    if has_errors:
        sys.exit(1)


@click.group()
def main():
    pass


main.add_command(_import)
main.add_command(_update)
main.add_command(_generate_db)
main.add_command(_verify)
main.add_command(_batch)


if __name__ == '__main__':
    main()


1			#!/usr/bin/env python
2
3			from __future__ import print_function
4			import os
5			import re
6			import sys
7			from datetime import datetime
8
9			import click
10			from send2trash import send2trash
11
12			# Verify that external dependencies are present first, so the user gets a
13			# more user-friendly error instead of an ImportError traceback.
14			from elodie.dependencies import verify_dependencies
15			if not verify_dependencies():
16			sys.exit(1)
17
18			from elodie import constants
19			from elodie import geolocation
20			from elodie import log
21			from elodie.compatability import _decode
22			from elodie.filesystem import FileSystem
23			from elodie.localstorage import Db
24			from elodie.media.base import Base, get_all_subclasses
25			from elodie.media.media import Media
26			from elodie.media.text import Text
27			from elodie.media.audio import Audio
28			from elodie.media.photo import Photo
29			from elodie.media.video import Video
30			from elodie.plugins.plugins import Plugins
31			from elodie.result import Result
32
33
34			FILESYSTEM = FileSystem()
35
36
37			def import_file(_file, destination, album_from_folder, trash, allow_duplicates):
38
39			_file = _decode(_file)
40			destination = _decode(destination)
41
42			"""Set file metadata and move it to destination.
43			"""
44			if not os.path.exists(_file):
45			log.warn('Could not find %s' % _file)
46			log.all('{"source":"%s", "error_msg":"Could not find %s"}' %
47			(_file, _file))
48			return
49			# Check if the source, _file, is a child folder within destination
50			elif destination.startswith(os.path.abspath(os.path.dirname(_file))+os.sep):
51			log.all('{"source": "%s", "destination": "%s", "error_msg": "Source cannot be in destination"}' % (
52			_file, destination))
53			return
54
55
56			media = Media.get_class_by_file(_file, get_all_subclasses())
57			if not media:
58			log.warn('Not a supported file (%s)' % _file)
59			log.all('{"source":"%s", "error_msg":"Not a supported file"}' % _file)
60			return
61
62			if album_from_folder:
63			media.set_album_from_folder()
64
65			dest_path = FILESYSTEM.process_file(_file, destination,
66			media, allowDuplicate=allow_duplicates, move=False)
67			if dest_path:
68			log.all('%s -> %s' % (_file, dest_path))
69			if trash:
70			send2trash(_file)
71
72			return dest_path or None
73
74			@click.command('batch')
75			@click.option('--debug', default=False, is_flag=True,
76			help='Override the value in constants.py with True.')
77			def _batch(debug):
78			"""Run batch() for all plugins.
79			"""
80			constants.debug = debug
81			plugins = Plugins()
82			plugins.run_batch()
83
84
85			@click.command('import')
86			@click.option('--destination', type=click.Path(file_okay=False),
87			required=True, help='Copy imported files into this directory.')
88			@click.option('--source', type=click.Path(file_okay=False),
89			help='Import files from this directory, if specified.')
90			@click.option('--file', type=click.Path(dir_okay=False),
91			help='Import this file, if specified.')
92			@click.option('--album-from-folder', default=False, is_flag=True,
93			help="Use images' folders as their album names.")
94			@click.option('--trash', default=False, is_flag=True,
95			help='After copying files, move the old files to the trash.')
96			@click.option('--allow-duplicates', default=False, is_flag=True,
97			help='Import the file even if it\'s already been imported.')
98			@click.option('--debug', default=False, is_flag=True,
99			help='Override the value in constants.py with True.')
100			@click.argument('paths', nargs=-1, type=click.Path())
101			def _import(destination, source, file, album_from_folder, trash, allow_duplicates, debug, paths):
102			"""Import files or directories by reading their EXIF and organizing them accordingly.
103			"""
104			constants.debug = debug
105			has_errors = False
106			result = Result()
107
108			destination = _decode(destination)
109			destination = os.path.abspath(os.path.expanduser(destination))
110
111			files = set()
112			paths = set(paths)
113			if source:
114			source = _decode(source)
115			paths.add(source)
116			if file:
117			paths.add(file)
118			for path in paths:
119			path = os.path.expanduser(path)
120			if os.path.isdir(path):
121			files.update(FILESYSTEM.get_all_files(path, None))
122			else:
123			files.add(path)
124
125			for current_file in files:
126			dest_path = import_file(current_file, destination, album_from_folder,
127			trash, allow_duplicates)
128			result.append((current_file, dest_path))
129			has_errors = has_errors is True or not dest_path
130
131			result.write()
132
133			if has_errors:
134			sys.exit(1)
135
136
137			@click.command('generate-db')
138			@click.option('--source', type=click.Path(file_okay=False),
139			required=True, help='Source of your photo library.')
140			@click.option('--debug', default=False, is_flag=True,
141			help='Override the value in constants.py with True.')
142			def _generate_db(source, debug):
143			"""Regenerate the hash.json database which contains all of the sha256 signatures of media files. The hash.json file is located at ~/.elodie/.
144			"""
145			constants.debug = debug
146			result = Result()
147			source = os.path.abspath(os.path.expanduser(source))
148
149			if not os.path.isdir(source):
150			log.error('Source is not a valid directory %s' % source)
151			sys.exit(1)
152
153			db = Db()
154			db.backup_hash_db()
155			db.reset_hash_db()
156
157			for current_file in FILESYSTEM.get_all_files(source):
158			result.append((current_file, True))
159			db.add_hash(db.checksum(current_file), current_file)
160			log.progress()
161
162			db.update_hash_db()
163			log.progress('', True)
164			result.write()
165
166			@click.command('verify')
167			@click.option('--debug', default=False, is_flag=True,
168			help='Override the value in constants.py with True.')
169			def _verify(debug):
170			constants.debug = debug
171			result = Result()
172			db = Db()
173			for checksum, file_path in db.all():
174			if not os.path.isfile(file_path):
175			result.append((file_path, False))
176			log.progress('x')
177			continue
178
179			actual_checksum = db.checksum(file_path)
180			if checksum == actual_checksum:
181			result.append((file_path, True))
182			log.progress()
183			else:
184			result.append((file_path, False))
185			log.progress('x')
186
187			log.progress('', True)
188			result.write()
189
190
191			def update_location(media, file_path, location_name):
192			"""Update location exif metadata of media.
193			"""
194			location_coords = geolocation.coordinates_by_name(location_name)
195
196			if location_coords and 'latitude' in location_coords and \
197			'longitude' in location_coords:
198			location_status = media.set_location(location_coords[
199			'latitude'], location_coords['longitude'])
200			if not location_status:
201			log.error('Failed to update location')
202			log.all(('{"source":"%s",' % file_path,
203			'"error_msg":"Failed to update location"}'))
204			sys.exit(1)
205			return True
206
207
208			def update_time(media, file_path, time_string):
209			"""Update time exif metadata of media.
210			"""
211			time_format = '%Y-%m-%d %H:%M:%S'
212			if re.match(r'^\d{4}-\d{2}-\d{2}$', time_string):
213			time_string = '%s 00:00:00' % time_string
214			elif re.match(r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}\d{2}$', time_string):
215			msg = ('Invalid time format. Use YYYY-mm-dd hh:ii:ss or YYYY-mm-dd')
216			log.error(msg)
217			log.all('{"source":"%s", "error_msg":"%s"}' % (file_path, msg))
218			sys.exit(1)
219
220			time = datetime.strptime(time_string, time_format)
221			media.set_date_taken(time)
222			return True
223
224
225			@click.command('update')
226			@click.option('--album', help='Update the image album.')
227			@click.option('--location', help=('Update the image location. Location '
228			'should be the name of a place, like "Las '
229			'Vegas, NV".'))
230			@click.option('--time', help=('Update the image time. Time should be in '
231			'YYYY-mm-dd hh:ii:ss or YYYY-mm-dd format.'))
232			@click.option('--title', help='Update the image title.')
233			@click.option('--debug', default=False, is_flag=True,
234			help='Override the value in constants.py with True.')
235			@click.argument('paths', nargs=-1,
236			required=True)
237			def _update(album, location, time, title, paths, debug):
238			"""Update a file's EXIF. Automatically modifies the file's location and file name accordingly.
239			"""
240			constants.debug = debug
241			has_errors = False
242			result = Result()
243
244			files = set()
245			for path in paths:
246			path = os.path.expanduser(path)
247			if os.path.isdir(path):
248			files.update(FILESYSTEM.get_all_files(path, None))
249			else:
250			files.add(path)
251
252			for current_file in files:
253			if not os.path.exists(current_file):
254			has_errors = True
255			result.append((current_file, False))
256			log.warn('Could not find %s' % current_file)
257			log.all('{"source":"%s", "error_msg":"Could not find %s"}' %
258			(current_file, current_file))
259			continue
260
261			current_file = os.path.expanduser(current_file)
262
263			# The destination folder structure could contain any number of levels
264			# So we calculate that and traverse up the tree.
265			# '/path/to/file/photo.jpg' -> '/path/to/file' ->
266			# ['path','to','file'] -> ['path','to'] -> '/path/to'
267			current_directory = os.path.dirname(current_file)
268			destination_depth = -1 * len(FILESYSTEM.get_folder_path_definition())
269			destination = os.sep.join(
270			os.path.normpath(
271			current_directory
272			).split(os.sep)[:destination_depth]
273			)
274
275			media = Media.get_class_by_file(current_file, get_all_subclasses())
276			if not media:
277			continue
278
279			updated = False
280			if location:
281			update_location(media, current_file, location)
282			updated = True
283			if time:
284			update_time(media, current_file, time)
285			updated = True
286			if album:
287			media.set_album(album)
288			updated = True
289
290			# Updating a title can be problematic when doing it 2+ times on a file.
291			# You would end up with img_001.jpg -> img_001-first-title.jpg ->
292			# img_001-first-title-second-title.jpg.
293			# To resolve that we have to track the prior title (if there was one.
294			# Then we massage the updated_media's metadata['base_name'] to remove
295			# the old title.
296			# Since FileSystem.get_file_name() relies on base_name it will properly
297			# rename the file by updating the title instead of appending it.
298			remove_old_title_from_name = False
299			if title:
300			# We call get_metadata() to cache it before making any changes
301			metadata = media.get_metadata()
302			title_update_status = media.set_title(title)
303			original_title = metadata['title']
304			if title_update_status and original_title:
305			# @TODO: We should move this to a shared method since
306			# FileSystem.get_file_name() does it too.
307			original_title = re.sub(r'\W+', '-', original_title.lower())
308			original_base_name = metadata['base_name']
309			remove_old_title_from_name = True
310			updated = True
311
312			if updated:
313			updated_media = Media.get_class_by_file(current_file,
314			get_all_subclasses())
315			# See comments above on why we have to do this when titles
316			# get updated.
317			if remove_old_title_from_name and len(original_title) > 0:
			0 ignored issues – show introduced 2019-03-11 22:04 UTC by Report Bug Copy Issue Report The variable `original_title` does not seem to be defined for all execution paths. Loading history...
318			updated_media.get_metadata()
319			updated_media.set_metadata_basename(
320			original_base_name.replace('-%s' % original_title, ''))
			0 ignored issues – show introduced 2019-03-11 22:04 UTC by Report Bug Copy Issue Report The variable `original_base_name` does not seem to be defined for all execution paths. Loading history...
321
322			dest_path = FILESYSTEM.process_file(current_file, destination,
323			updated_media, move=True, allowDuplicate=True)
324			log.info(u'%s -> %s' % (current_file, dest_path))
325			log.all('{"source":"%s", "destination":"%s"}' % (current_file,
326			dest_path))
327			# If the folder we moved the file out of or its parent are empty
328			# we delete it.
329			FILESYSTEM.delete_directory_if_empty(os.path.dirname(current_file))
330			FILESYSTEM.delete_directory_if_empty(
331			os.path.dirname(os.path.dirname(current_file)))
332			result.append((current_file, dest_path))
333			# Trip has_errors to False if it's already False or dest_path is.
334			has_errors = has_errors is True or not dest_path
335			else:
336			has_errors = False
337			result.append((current_file, False))
338
339			result.write()
340
341			if has_errors:
342			sys.exit(1)
343
344
345			@click.group()
346			def main():
347			pass
348
349
350			main.add_command(_import)
351			main.add_command(_update)
352			main.add_command(_generate_db)
353			main.add_command(_verify)
354			main.add_command(_batch)
355
356
357			if __name__ == '__main__':
358			main()
359

jmathai / elodie

Push — master ( 9e42ed...b2d36d )

elodie._batch() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like