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

Passed

Pull Request — master (#319)

by Jaisen

created 2019-07-12 05:35 UTC

elodie._batch() A

↳ Parent: elodie

Complexity

Conditions

Size

Total Lines	7
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

jmathai / elodie

Pull Request — master (#319)

elodie._batch() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like