9 def walk_media_files(dir_path):
10 for root, dirs, files in os.walk(dir_path):
12 file_path = os.path.join(root, f)
13 if _is_media_file(file_path):
16 def extract_timestamp(file_path, use_meta=False):
19 if _is_media_file(file_path, types=['image']):
20 time = _extract_image_timestamp(file_path)
21 elif _is_media_file(file_path, types=['video']):
22 time = _extract_video_timestamp(file_path)
26 return os.path.getmtime(file_path)
28 def find_file(dir_path, file_name, file_size, exclude_dir):
29 for root, dirs, files in os.walk(dir_path):
30 if root.startswith(exclude_dir):
34 full_path = os.path.join(root, f)
35 if os.path.getsize(full_path) == file_size:
39 def import_file(src_file_path, dst_file_path):
40 shutil.copyfile(src_file_path, dst_file_path)
41 src_stat = os.stat(src_file_path)
42 dst_stat = os.stat(dst_file_path)
43 os.utime( dst_file_path, ns=(dst_stat.st_atime_ns, src_stat.st_mtime_ns) )
45 def cleanup_dir(dir_path):
46 for root, dirs, files in os.walk(dir_path, topdown=False):
48 full_path = os.path.join(root, name)
49 if not _is_media_file(full_path):
52 except Exception as e:
53 logging.warn('Error cleaning file %s: %s', full_path, str(e))
55 full_path = os.path.join(root, name)
58 except Exception as e:
59 logging.warn('Error cleaning dir %s: %s', full_path, str(e))
61 def _is_media_file(file_path, types=['image', 'video']):
62 if not os.path.isfile(file_path):
64 mime_type = mimetypes.guess_type(file_path)[0]
67 if not mime_type.split('/')[0] in types:
71 def _extract_image_timestamp(file_path):
72 with PIL.Image.open(file_path) as image:
73 exif = image._getexif()
74 if exif and 36867 in exif:
75 return datetime.datetime\
76 .strptime(exif[36867], '%Y:%m:%d %H:%M:%S')\
80 def _extract_video_timestamp(file_path):
81 p = subprocess.run(['ffmpeg', '-i', file_path],
82 capture_output=True, encoding='UTF-8')
83 for line in p.stderr.splitlines():
84 m = re.search('^.*creation_time.*: ([^ ]+)$', line)
86 return datetime.datetime\
87 .strptime(m.group(1), '%Y-%m-%dT%H:%M:%S.%fZ')\