snapchat-memories-downloader/notebooks/make_df.ipynb

521 lines
14 KiB
Plaintext

{
"cells": [
{
"cell_type": "code",
"execution_count": null,
"id": "ffcab11d",
"metadata": {},
"outputs": [],
"source": [
"import pandas as pd"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "a4cda272",
"metadata": {},
"outputs": [],
"source": [
"df: pd.DataFrame = pd.read_json(\"../temp/file_info.json\", orient='records')"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "09ee7182",
"metadata": {},
"outputs": [],
"source": [
"df.head(10)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "d8fb8fd5",
"metadata": {},
"outputs": [],
"source": [
"def get_join_col(row: pd.Series) -> pd.Series:\n",
" row['file_dt'] = f\"{row['Date'].strftime('%Y-%m-%d_%H-%M-%S')}\"\n",
" return row"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "71007845",
"metadata": {},
"outputs": [],
"source": [
"def extract_lat_long(row: pd.Series) -> pd.Series:\n",
" coordinates = row[\"Location\"].split(\": \")[1]\n",
" lat_str, long_str = coordinates.split(\", \")\n",
" row[\"location_latitude\"] = round(float(lat_str), 5)\n",
" row[\"location_longitude\"] = round(float(long_str), 5)\n",
" return row"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "68ed283f",
"metadata": {},
"outputs": [],
"source": [
"df = df.apply(get_join_col, axis=1)\n",
"df = df.apply(extract_lat_long, axis=1)\n",
"df.head(10)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "713d3202",
"metadata": {},
"outputs": [],
"source": [
"metadata = pd.read_csv(\"../temp/filemetadata.csv\", sep=\"#\")\n",
"metadata.head(10)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "cb28dcd6",
"metadata": {},
"outputs": [],
"source": [
"import re\n",
"\n",
"def dms_to_decimal(dms_string):\n",
" \"\"\"\n",
" Converts a geographical coordinate string in Degrees, Minutes, Seconds (DMS)\n",
" format (e.g., \"44 deg 6' 15.76\" N\") to a decimal degree coordinate.\n",
"\n",
" Args:\n",
" dms_string (str): The DMS coordinate string.\n",
"\n",
" Returns:\n",
" float: The decimal degree equivalent.\n",
" \n",
" Raises:\n",
" ValueError: If the string format is not recognized.\n",
" \"\"\"\n",
" # Regex to capture Degrees, Minutes, Seconds, and Direction\n",
" # Pattern: (\\d+) deg (\\d+)' ([\\d.]+)\" ([NSEW])\n",
" # The pattern is flexible with leading zeros, extra spaces, and decimal seconds.\n",
" match = re.search(r\"(\\d+)\\s*deg\\s*(\\d+)'\\s*([\\d.]+)\\\"\\s*([NSEW])\", dms_string, re.IGNORECASE)\n",
"\n",
" if not match:\n",
" raise ValueError(f\"Coordinate format not recognized: {dms_string}\")\n",
"\n",
" # Extract the parts, ensuring they are converted to appropriate numerical types\n",
" degrees = float(match.group(1))\n",
" minutes = float(match.group(2))\n",
" seconds = float(match.group(3))\n",
" direction = match.group(4).upper()\n",
"\n",
" # Calculate the decimal degree value\n",
" decimal_degrees = degrees + (minutes / 60) + (seconds / 3600)\n",
"\n",
" # Apply the sign based on the direction (S and W are negative)\n",
" if direction in ['S', 'W']:\n",
" decimal_degrees *= -1\n",
" \n",
" decimal_degrees = round(decimal_degrees, 5)\n",
"\n",
" return decimal_degrees"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "97ee9100",
"metadata": {},
"outputs": [],
"source": [
"def convert_lat_long_to_decimal_degrees(row: pd.Series) -> pd.Series:\n",
" # def convert(value: str) -> float:\n",
" # if pd.isna(value):\n",
" # return None\n",
" # if value == '-':\n",
" # return 0.0\n",
" # parts = value.strip().split(' ')\n",
" # degrees = float(parts[0])\n",
" # minutes = float(parts[1]) / 60\n",
" # seconds = float(parts[2]) / 3600\n",
" # direction = parts[3]\n",
" # decimal_degrees = degrees + minutes + seconds\n",
" # if direction in ['S', 'W']:\n",
" # decimal_degrees = -decimal_degrees\n",
" # return decimal_degrees\n",
" def convert(value: str) -> float:\n",
" if pd.isna(value):\n",
" return 0.0\n",
" if value == '-':\n",
" return 0.0\n",
" try:\n",
" return dms_to_decimal(value)\n",
" except ValueError:\n",
" return None\n",
" \n",
" row[\"Latitude\"] = convert(row[\"gpslatitude\"])\n",
" row[\"Longitude\"] = convert(row[\"gpslongitude\"])\n",
" return row"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "8684b468",
"metadata": {},
"outputs": [],
"source": [
"def get_file_name(row: pd.Series) -> pd.Series:\n",
" path_parts = row['file'].split('/')\n",
" file = path_parts[-1]\n",
" row['filename'] = file\n",
" \n",
" row['extension'] = file.split('.')[-1].lower()\n",
" row['file_dt'] = file.split('.')[0]\n",
" return row"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "6a1e641e",
"metadata": {},
"outputs": [],
"source": [
"meta = metadata.apply(convert_lat_long_to_decimal_degrees, axis=1)\n",
"meta = meta.apply(get_file_name, axis=1)\n",
"meta.head(10)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "543eecf8",
"metadata": {},
"outputs": [],
"source": [
"import pytz\n",
"def get_datetime_for_df(row: pd.Series) -> pd.Series:\n",
" from datetime import datetime\n",
" try:\n",
" row['createdate_mod'] = None\n",
" if row['createdate'] != '-':\n",
" createdate = datetime.strptime(row['createdate'], '%Y:%m:%d %H:%M:%S')\n",
" createdate_wtz = createdate.astimezone(pytz.timezone('America/Chicago'))\n",
" \n",
" row['createdate_mod'] = createdate_wtz\n",
" \n",
" \n",
" date = datetime.strptime(row['file_dt'], '%Y-%m-%d_%H-%M-%S')\n",
" dt_w_tz = date.astimezone(pytz.UTC)\n",
" # dt_w_tz = date.dtz_localize('UTC')\n",
" # dt_w_tz = date.tz_localize('UTC')\n",
" row['datetime'] = dt_w_tz\n",
" row['dt_join'] = f\"{dt_w_tz}\"\n",
" # row['datetime_str'] = dt_w_tz.strftime('%Y-%m-%d %H:%M:%S %Z')\n",
" # row['datetime'] = datetime.strptime(row['Date'], '%Y-%m-%d %H-%M-%S')\n",
" except ValueError:\n",
" row['createdate_mod'] = None\n",
" row['datetime'] = None\n",
" row['dt_join'] = ''\n",
" return row"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "a162c993",
"metadata": {},
"outputs": [],
"source": [
"def get_media_type(row: pd.Series) -> pd.Series:\n",
" row['file_type'] = 'Unknown'\n",
" if row['extension'] == 'jpg':\n",
" row['file_type'] = 'Image'\n",
" if row['extension'] == 'mp4':\n",
" row['file_type'] = 'Video'\n",
" return row"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "b57f9726",
"metadata": {},
"outputs": [],
"source": [
"data = meta.apply(get_datetime_for_df, axis=1)\n",
"data = data.apply(get_media_type, axis=1)\n",
"data.head(10)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "cea52cd0",
"metadata": {},
"outputs": [],
"source": [
"joined = pd.merge(df, data, how='left', left_on='file_dt', right_on='file_dt', suffixes=('_df', '_meta'))\n",
"joined.head(10)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "3f350445",
"metadata": {},
"outputs": [],
"source": [
"col_mapper = {\n",
" 'Date': 'correct_date_utc',\n",
" 'Media Type': 'correct_media_type',\n",
" # 'Location': 'correct_location',\n",
" 'location_latitude': 'correct_latitude',\n",
" 'location_longitude': 'correct_longitude',\n",
" 'file': 'path',\n",
" 'filename': 'filename',\n",
" 'createdate': 'actual_date_cst',\n",
" 'Latitude': 'actual_latitude',\n",
" 'Longitude': 'actual_longitude',\n",
" # 'extension': 'actual_extension',\n",
" 'file_type': 'actual_media_type'\n",
"}"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "07ac5dd2",
"metadata": {},
"outputs": [],
"source": [
"df0 = joined[col_mapper.keys()].rename(columns=col_mapper)\n",
"df0.head(10)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "192561b3",
"metadata": {},
"outputs": [],
"source": [
"def check_for_errors(row: pd.Series) -> pd.Series:\n",
" errors = 0\n",
" if row[\"correct_media_type\"] != row['actual_media_type']:\n",
" errors += 1\n",
" if row[\"actual_latitude\"] == 0.00 and row[\"correct_media_type\"] == \"Image\":\n",
" if row['correct_latitude'] != 0.00:\n",
" errors +=2\n",
" if row[\"actual_date_cst\"] == '-':\n",
" errors +=4\n",
" if pd.isna(row['path']):\n",
" errors +=20\n",
" row[\"errors\"] = errors\n",
" return row"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "c89101b9",
"metadata": {},
"outputs": [],
"source": [
"# 1 = Media Type\n",
"# 2 = Latitude\n",
"# 4 = Date\n",
"# 20 = Path\n",
"\n",
"df1 = df0.apply(check_for_errors, axis=1)\n",
"df1.head(10)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "445c1b9b",
"metadata": {},
"outputs": [],
"source": [
"errors = df1[df1['errors'] > 0]\n",
"errors.head(10)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "de9ed1f5",
"metadata": {},
"outputs": [],
"source": [
"errors.shape"
]
},
{
"cell_type": "markdown",
"id": "bd5d80b3",
"metadata": {},
"source": [
"679/3700 (oof)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "0b325144",
"metadata": {},
"outputs": [],
"source": [
"img_errors = errors[errors['correct_media_type'] == 'Image']\n",
"img_errors.shape"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "0649c3d6",
"metadata": {},
"outputs": [],
"source": [
"img_errors.head(10)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "3559f965",
"metadata": {},
"outputs": [],
"source": [
"video_errors = errors[errors['correct_media_type'] == 'Video']\n",
"video_errors.shape"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "49e183fd",
"metadata": {},
"outputs": [],
"source": [
"video_errors.head(10)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "9e5569c5",
"metadata": {},
"outputs": [],
"source": [
"wrong_filetype = df1[df1['correct_media_type'] != df1['actual_media_type']]\n",
"wrong_filetype.head(10)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "0a39979b",
"metadata": {},
"outputs": [],
"source": [
"wrong_filetype = wrong_filetype.fillna('-')"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "9714201c",
"metadata": {},
"outputs": [],
"source": [
"wrong_filetype.head(10)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "2a77f8b5",
"metadata": {},
"outputs": [],
"source": [
"# to_change = wrong_filetype[wrong_filetype['path'] != '-']\n",
"# to_change.head(10)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "f06455b4",
"metadata": {},
"outputs": [],
"source": [
"# import os\n",
"# os.rename(src=\"../downloads/2017-11-03_16-46-57.mp4\", dst=\"../downloads/2017-11-03_16-46-57.jpg\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "f7891aa1",
"metadata": {},
"outputs": [],
"source": [
"import os\n",
"\n",
"def change_filetype(row: pd.Series) -> pd.Series:\n",
" try:\n",
" if row['path'] != '-':\n",
" if row['correct_media_type'] == 'Image':\n",
" os.rename(src=f\"../downloads/{row['filename'][:-4]}.mp4\", dst=f\"../downloads/{row['filename'][:-4]}.jpg\")\n",
" elif row['correct_media_type'] == 'Video':\n",
" os.rename(src=f\"../downloads/{row['filename'][:-4]}.jpg\", dst=f\"../downloads/{row['filename'][:-4]}.mp4\")\n",
" except FileNotFoundError:\n",
" print(f\"File not found: {row['filename']}\")\n",
" return row"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "df5eedbd",
"metadata": {},
"outputs": [],
"source": [
"wrong_filetype.apply(change_filetype, axis=1)"
]
}
],
"metadata": {
"kernelspec": {
"display_name": ".venv",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.13.11"
}
},
"nbformat": 4,
"nbformat_minor": 5
}