import pandas as pd import re import pytz import os from timezonefinder import TimezoneFinder from datetime import datetime import subprocess from tqdm.asyncio import tqdm df: pd.DataFrame = pd.read_json("./temp/file_info.json", orient="records") def get_join_col(row: pd.Series) -> pd.Series: row["file_dt"] = f"{row['Date'].strftime('%Y-%m-%d_%H-%M-%S')}" return row def extract_lat_long(row: pd.Series) -> pd.Series: coordinates = row["Location"].split(": ")[1] lat_str, long_str = coordinates.split(", ") row["location_latitude"] = round(float(lat_str), 5) row["location_longitude"] = round(float(long_str), 5) return row df = df.apply(get_join_col, axis=1) df = df.apply(extract_lat_long, axis=1) metadata = pd.read_csv("./temp/filemetadata.csv", sep="#") metadata["file"] = f"./" + metadata["file"] def convert_lat_long_to_decimal_degrees(row: pd.Series) -> pd.Series: def dms_to_decimal(dms_string): match = re.search( r"(\d+)\s*deg\s*(\d+)'\s*([\d.]+)\"\s*([NSEW])", dms_string, re.IGNORECASE ) if not match: raise ValueError(f"Coordinate format not recognized: {dms_string}") degrees = float(match.group(1)) minutes = float(match.group(2)) seconds = float(match.group(3)) direction = match.group(4).upper() decimal_degrees = degrees + (minutes / 60) + (seconds / 3600) if direction in ["S", "W"]: decimal_degrees *= -1 decimal_degrees = round(decimal_degrees, 5) return decimal_degrees def convert(value: str) -> float: if pd.isna(value): return 0.0 if value == "-": return 0.0 try: return dms_to_decimal(value) except ValueError: return 0.0 row["Latitude"] = convert(row["gpslatitude"]) row["Longitude"] = convert(row["gpslongitude"]) return row def get_file_name(row: pd.Series) -> pd.Series: path_parts = row["file"].split("/") file = path_parts[-1] row["filename"] = file row["extension"] = file.split(".")[-1].lower() row["file_dt"] = file.split(".")[0] return row meta = metadata.apply(convert_lat_long_to_decimal_degrees, axis=1) meta = meta.apply(get_file_name, axis=1) def get_datetime_for_df(row: pd.Series) -> pd.Series: try: row["createdate_mod"] = None if row["createdate"] != "-": createdate = datetime.strptime(row["createdate"], "%Y:%m:%d %H:%M:%S") createdate_wtz = createdate.astimezone(pytz.timezone("America/Chicago")) row["createdate_mod"] = createdate_wtz date = datetime.strptime(row["file_dt"], "%Y-%m-%d_%H-%M-%S") dt_w_tz = date.astimezone(pytz.UTC) row["datetime"] = dt_w_tz row["dt_join"] = f"{dt_w_tz}" except ValueError: row["createdate_mod"] = None row["datetime"] = None row["dt_join"] = "" return row def get_media_type(row: pd.Series) -> pd.Series: row["file_type"] = "Unknown" if row["extension"] == "jpg": row["file_type"] = "Image" if row["extension"] == "mp4": row["file_type"] = "Video" return row data = meta.apply(get_datetime_for_df, axis=1) data = data.apply(get_media_type, axis=1) joined = pd.merge( df, data, how="left", left_on="file_dt", right_on="file_dt", suffixes=("_df", "_meta"), ) col_mapper = { "Date": "correct_date_utc", "Media Type": "correct_media_type", "location_latitude": "correct_latitude", "location_longitude": "correct_longitude", "file": "path", "filename": "filename", "createdate": "actual_date_cst", "Latitude": "actual_latitude", "Longitude": "actual_longitude", "file_type": "actual_media_type", } df0 = joined[col_mapper.keys()].rename(columns=col_mapper) def check_for_errors(row: pd.Series) -> pd.Series: errors = 0 if row["correct_media_type"] != row["actual_media_type"]: errors += 1 if row["actual_latitude"] == 0.00 and row["correct_media_type"] == "Image": if row["correct_latitude"] != 0.00: errors += 1 if abs(row["actual_longitude"] - row["correct_longitude"]) > 2.00: errors += 1 if row["actual_date_cst"] == "-": errors += 1 if pd.isna(row["path"]): errors += 1 row["errors"] = errors return row df1 = df0.apply(check_for_errors, axis=1) errors = df1[df1["errors"] > 0] print(f"Errors: {errors.shape[0]}") img_errors = errors[errors["correct_media_type"] == "Image"] print(f"Image errors: {img_errors.shape[0]}") video_errors = errors[errors["correct_media_type"] == "Video"] print(f"Video errors: {video_errors.shape[0]}") def fix_filetype(row: pd.Series) -> pd.Series: try: if pd.isna(row["path"]) or row["path"] == "-": return row mp4_file_path = f"./downloads/{row['filename'][:-4]}.mp4" jpg_file_path = f"./downloads/{row['filename'][:-4]}.jpg" if row["correct_media_type"] == row["actual_media_type"]: return row if row["path"] == "-": return row if row["correct_media_type"] == "Image": os.rename(src=mp4_file_path, dst=jpg_file_path) if os.path.exists(jpg_file_path): row["filename"] = f"{row['filename'][:-4]}.jpg" row["actual_media_type"] = "Image" elif row["correct_media_type"] == "Video": os.rename(src=jpg_file_path, dst=mp4_file_path) if os.path.exists(mp4_file_path): row["filename"] = f"{row['filename'][:-4]}.mp4" row["actual_media_type"] = "Video" except FileNotFoundError: print(f"File not found: {row['filename']}") return row df1.apply(fix_filetype, axis=1) tf = TimezoneFinder() def get_localized_dt_and_offset( utc_dt: datetime, latitude: float, longitude: float ) -> tuple[str, str]: tz_name = tf.timezone_at(lng=longitude, lat=latitude) utc_timezone = pytz.utc central_timezone = pytz.timezone("America/Chicago") mountain_timezone = pytz.timezone("America/Denver") if not tz_name or latitude == 0 or longitude == 0: print( f"⚠️ Warning: No timezone found for {utc_dt.strftime("%Y-%m-%d %H:%M:%S")} UTC. Defaulting to MDT/MST." ) local_tz = mountain_timezone else: local_tz = pytz.timezone(tz_name) dt_local = utc_dt.astimezone(local_tz) dt_str = dt_local.strftime("%Y:%m:%d %H:%M:%S") # Tags require offset in ±HH:MM format offset_str: str = dt_local.strftime("%z") if len(offset_str) == 5: offset_str = offset_str[:3] + ":" + offset_str[3:] return dt_str, offset_str progress_bar = tqdm( total=df1.shape[0], desc="Updating EXIF", unit="file", disable=False, ) def update_exif_with_exiftool(row: pd.Series) -> None: progress_bar.update(1) image_path = row["path"] dt_utc = row["correct_date_utc"] # timezone-aware, UTC datetime latitude = row["correct_latitude"] longitude = row["correct_longitude"] file_type = row["correct_media_type"] abs_latitude = abs(latitude) abs_longitude = abs(longitude) dt_utc_str = dt_utc.strftime("%Y:%m:%d %H:%M:%S") if image_path is None or pd.isna(image_path) or image_path == "": print(f"❌ File not found, skipping: {dt_utc_str}") return lat_ref = "N" if latitude >= 0 else "S" lon_ref = "E" if longitude >= 0 else "W" local_dt, dynamic_tz = get_localized_dt_and_offset( utc_dt=dt_utc, latitude=row["correct_latitude"], longitude=row["correct_longitude"], ) if not os.path.exists(image_path): print(f"❌ File not found, skipping: {image_path}") return jpg_command = [ "exiftool", # Date/Time tags f"-IFD0:DateTime={local_dt}", f"-ExifIFD:DateTimeOriginal={local_dt}", f"-ExifIFD:DateTimeDigitized={local_dt}", f"-OffsetTimeOriginal={dynamic_tz}", # Location tags f"-GPSLatitude={latitude}", f"-GPSLongitude={longitude}", f"-GPSLatitudeRef='{lat_ref}'", f"-GPSLongitudeRef='{lon_ref}'", "-overwrite_original", image_path, ] jpg_command_no_gps = [ "exiftool", # Date/Time tags f"-IFD0:DateTime={local_dt}", f"-ExifIFD:DateTimeOriginal={local_dt}", f"-ExifIFD:DateTimeDigitized={local_dt}", f"-OffsetTimeOriginal={dynamic_tz}", "-overwrite_original", image_path, ] mp4_command = [ "exiftool", # Date/Time tags # EXIF Tags f"-IFD0:DateTime={local_dt}", f"-ExifIFD:DateTimeOriginal={local_dt}", f"-ExifIFD:DateTimeDigitized={local_dt}", f"-OffsetTimeOriginal={dynamic_tz}", # QuickTime tags (must be set to UTC time) f"-QuickTime:CreationDate={dt_utc_str}", f"-QuickTime:TrackCreateDate={dt_utc_str}", f"-QuickTime:MediaCreateDate={dt_utc_str}", f"-QuickTime:TimeZone={dynamic_tz}", # Location tags # Needed for location on Google Photos f"-QuickTime:GPSCoordinates={abs_latitude} {lat_ref}, {abs_longitude} {lon_ref}", # XMP Tags (Adobe/Google f"-xmp:gpslatitude={latitude}", f"-xmp:gpslongitude={longitude}", # QuickTime Tags (Apple/iOS) f"-QuickTime:LocationLatitude={latitude}", f"-QuickTime:LocationLongitude={longitude}", # # Generic GPS tags (Broader compatibility) f"-GPSLatitude={latitude}", f"-GPSLongitude={longitude}", f"-GPSLatitudeRef={lat_ref}", f"-GPSLongitudeRef={lon_ref}", "-overwrite_original", image_path, ] mp4_command_no_gps = [ "exiftool", # EXIF Tags f"-IFD0:DateTime={local_dt}", f"-ExifIFD:DateTimeOriginal={local_dt}", f"-ExifIFD:DateTimeDigitized={local_dt}", f"-OffsetTimeOriginal={dynamic_tz}", # QuickTime tags (must be set to UTC time) f"-QuickTime:CreationDate={dt_utc_str}", f"-QuickTime:TrackCreateDate={dt_utc_str}", f"-QuickTime:MediaCreateDate={dt_utc_str}", f"-QuickTime:TimeZone={dynamic_tz}", "-overwrite_original", image_path, ] try: if file_type == "Video": if latitude == 0.0 and longitude == 0.0: result = subprocess.run( mp4_command_no_gps, check=True, capture_output=True, text=True ) else: result = subprocess.run( mp4_command, check=True, capture_output=True, text=True ) elif file_type == "Image": if latitude == 0.0 and longitude == 0.0: result = subprocess.run( jpg_command_no_gps, check=True, capture_output=True, text=True ) else: result = subprocess.run( jpg_command, check=True, capture_output=True, text=True ) else: print(f"⚠️ Warning: Unknown media type for {image_path}, skipping.") return if "1 image files updated" in result.stdout: pass # print(f"✅ Updated: {image_path}") else: print(f"⚠️ Warning updating {image_path}: {result.stderr.strip()}") except subprocess.CalledProcessError as e: print(f"🛑 Subprocess Error on {image_path}: {e.stderr.strip()}") except FileNotFoundError: print( "🛑 ERROR: ExifTool command not found. Is ExifTool installed and in your PATH?" ) df1.apply(update_exif_with_exiftool, axis=1) # type: ignore progress_bar.close()