Skip to content

Add error handling for missing/iCloud files and export logging #7

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
glenbhermon wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Add error handling for missing/iCloud files and export logging #7

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
219 changes: 82 additions & 137 deletions main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,202 +12,147 @@
import termios
import subprocess


def create_connection(db_file):
"""
create a database connection to the SQLite database specified by the db_file
:param db_file: database file
:return: Connection object or None
"""
conn = None
try:
conn = sqlite3.connect(db_file)
except Error as e:
print(e)

print("Database Error: {}".format(e))
return conn


def get_all_memos(conn):
"""
Query wanted rows in the table ZCLOUDRECORDING
:param conn: the Connection object
:return: rows
"""
cur = conn.cursor()
# ZCUSTOMLABEL is the title you gave it, ZPATH is the filename
cur.execute("SELECT ZDATE, ZDURATION, ZCUSTOMLABEL, ZPATH FROM ZCLOUDRECORDING ORDER BY ZDATE")

return cur.fetchall()


def main():
# Define default paths
_db_path_default = os.path.join(os.path.expanduser("~"), "Library", "Application Support",
"com.apple.voicememos", "Recordings", "CloudRecordings.db")
_export_path_default = os.path.join(os.path.expanduser("~"), "Voice Memos Export")

# Setting up arguments and --help
parser = argparse.ArgumentParser(description='Export audio files from macOS Voice Memo App ' +
'with right filename and date created.')
parser.add_argument("-d", "--db_path", type=str,
help="define path to database of Voice Memos.app",
default=_db_path_default)
parser.add_argument("-e", "--export_path", type=str,
help="define path to folder for exportation",
default=_export_path_default)
parser.add_argument("-a", "--all", action="store_true",
help="export everything at once instead of step by step")
parser.add_argument("--date_in_name", action="store_true",
help="include date in file name")
parser.add_argument("--date_in_name_format", type=str,
help="define the format of the date in file name (if --date_in_name active)",
default="%Y-%m-%d-%H-%M-%S_")
parser.add_argument("--no_finder", action="store_true",
help="prevent to open finder window to show exported memos")
parser = argparse.ArgumentParser(description='Export Voice Memos from macOS 10.15 with Error Logging.')
parser.add_argument("-d", "--db_path", type=str, help="path to database", default=_db_path_default)
parser.add_argument("-e", "--export_path", type=str, help="path for exportation", default=_export_path_default)
parser.add_argument("-a", "--all", action="store_true", help="export all at once")
parser.add_argument("--date_in_name", action="store_true", help="include date in file name")
parser.add_argument("--date_in_name_format", type=str, help="date format", default="%Y-%m-%d-%H-%M-%S_")
parser.add_argument("--no_finder", action="store_true", help="don't open finder")
args = parser.parse_args()

# Define name and width of columns
_cols = [{"n": "Date",
"w": 19},
{"n": "Duration",
"w": 11},
{"n": "Old Path",
"w": 32},
{"n": "New Path",
"w": 60},
{"n": "Status",
"w": 12}]

# offset between datetime starts to count (1.1.1970) and Apple starts to count (1.1.2001)
_cols = [{"n": "Date", "w": 19}, {"n": "Duration", "w": 11}, {"n": "Old Path", "w": 32},
{"n": "New Path", "w": 60}, {"n": "Status", "w": 15}]

_dt_offset = 978307200.825232

def getWidth(name):
"""
get width of column called by name
:param name: name of column
:return: width
"""
for c in _cols:
if c["n"] == name:
return c["w"]
if c["n"] == name: return c["w"]
return False

def helper_str(seperator):
"""
create a helper string for printing table row
Example: helper_str(" | ").format(...)
:param seperator: string to symbol column boundary
:return: helper string like: "{0:10} | {1:50}"
"""
return seperator.join(["{" + str(i) + ":" + str(c["w"]) + "}" for i, c in enumerate(_cols)])

def body_row(content_list):
"""
create a string for a table body row
:param content_list: list of cells in this row
:return: table body row string
"""
return "│ " + helper_str(" │ ").format(*content_list) + " │"

# Check permission
# Permission check
if not os.access(args.db_path, os.R_OK):
print("No permission to read database file. ({})".format(args.db_path))
print("CRITICAL ERROR: No permission to read the database.")
print("Please grant Terminal 'Full Disk Access' in System Preferences.")
exit()

# create a database connection and load rows
conn = create_connection(args.db_path)
if not conn:
exit()
if not conn: exit()
with conn:
rows = get_all_memos(conn)
if not rows:
exit()
if not rows: exit()

# create export folder if it doesn't exist
try:
os.stat(args.export_path)
except:
os.mkdir(args.export_path)

# Print intro and table header
print()
if not args.all:
print("Press ENTER to export the memo shown in the current row or ESC to go to next memo.")
print("Do not press other keys.")
print()
print("┌─" + helper_str("─┬─").format(*["─" * c["w"] for c in _cols]) + "─┐")
if not os.path.exists(args.export_path):
os.makedirs(args.export_path)

# Initialize Log File
log_path = os.path.join(args.export_path, "failed_exports.txt")
with open(log_path, "w") as log_file:
log_file.write("Voice Memos Export Log - {}\n".format(datetime.now()))
log_file.write("="*50 + "\n")

print("\n┌─" + helper_str("─┬─").format(*["─" * c["w"] for c in _cols]) + "─┐")
print("│ " + helper_str(" │ ").format(*[c["n"] for c in _cols]) + " │")
print("├─" + helper_str("─┼─").format(*["─" * c["w"] for c in _cols]) + "─┤")

# iterate over memos found in database
for row in rows:
failed_count = 0
success_count = 0

# get information from database and modify them for exportation
for row in rows:
date = datetime.fromtimestamp(row[0] + _dt_offset)
date_str = date.strftime("%d.%m.%Y %H:%M:%S")
duration_str = str(timedelta(seconds=row[1]))
duration_str = duration_str[:duration_str.rfind(".") + 3] if "." in duration_str else duration_str + ".00"
duration_str = "0" + duration_str if len(duration_str) == 10 else duration_str
label = row[2].encode('ascii', 'ignore').decode("ascii").replace("/", "_")
path_old = row[3] if row[3] else ""
if path_old:
path_new = label + path_old[path_old.rfind("."):]
duration_str = str(timedelta(seconds=row[1])).split('.')[0]

# Clean the label for filenames
label = row[2].encode('ascii', 'ignore').decode("ascii").replace("/", "_") if row[2] else "Untitled"
path_old_raw = row[3] if row[3] else ""

if path_old_raw:
# Reconstruct absolute path
if not path_old_raw.startswith("/"):
path_old = os.path.join(os.path.dirname(args.db_path), path_old_raw)
else:
path_old = path_old_raw

path_new = label + os.path.splitext(path_old)[1]
path_new = date.strftime(args.date_in_name_format) + path_new if args.date_in_name else path_new
path_new = os.path.join(args.export_path, path_new)
else:
path_old = ""
path_new = ""
if len(path_old) < getWidth("Old Path") - 3:
path_old_short = path_old
else:
path_old_short = "..." + path_old[-getWidth("Old Path") + 3:]
if len(path_new) < getWidth("New Path") - 3:
path_new_short = path_new
else:
path_new_short = "..." + path_new[-getWidth("New Path") + 3:]

# print body row and wait for keys (if needed)
if not path_old:
print(body_row((date_str, duration_str, path_old_short, path_new_short, "No File")))
p_old_short = ("..." + path_old[-(getWidth("Old Path")-3):]) if len(path_old) > getWidth("Old Path") else path_old
p_new_short = ("..." + path_new[-(getWidth("New Path")-3):]) if len(path_new) > getWidth("New Path") else path_new

# INTERACTIVE CHECK
if args.all:
key = 10
else:
if args.all:
key = 10
else:
key = 0
print(body_row((date_str, duration_str, path_old_short, path_new_short, "Export?")), end="\r")
fd = sys.stdin.fileno()
old = termios.tcgetattr(fd)
new = termios.tcgetattr(fd)
new[3] = new[3] & ~termios.ECHO
termios.tcsetattr(fd, termios.TCSADRAIN, new)
tty.setcbreak(sys.stdin)
while key not in (10, 27):
try:
key = ord(sys.stdin.read(1))
# print("Key: {}".format(key))
finally:
termios.tcsetattr(fd, termios.TCSADRAIN, old)

# copy file and modify file times if this memo should be exported
if key == 10:
print(body_row((date_str, duration_str, p_old_short, p_new_short, "Export?")), end="\r")
fd = sys.stdin.fileno()
old_set = termios.tcgetattr(fd)
try:
tty.setcbreak(fd)
key = ord(sys.stdin.read(1))
finally:
termios.tcsetattr(fd, termios.TCSADRAIN, old_set)

if key == 10: # User chose to export
try:
if not path_old or not os.path.exists(path_old):
raise FileNotFoundError("Audio file not found on disk (likely in iCloud)")

copyfile(path_old, path_new)
mod_time = time.mktime(date.timetuple())
os.utime(path_new, (mod_time, mod_time))
print(body_row((date_str, duration_str, path_old_short, path_new_short, "Exported!")))
print(body_row((date_str, duration_str, p_old_short, p_new_short, "Success!")))
success_count += 1

# skip this memo if desired
elif key == 27:
print(body_row((date_str, duration_str, path_old_short, path_new_short, "Not Exported")))
except Exception as e:
# LOG THE FAILURE AND CONTINUE
with open(log_path, "a") as log_file:
log_file.write("FAILED: {} | Memo: {} | Reason: {}\n".format(date_str, label, str(e)))
print(body_row((date_str, duration_str, p_old_short, p_new_short, "FAILED (Logged)")))
failed_count += 1

elif key == 27: # ESC
print(body_row((date_str, duration_str, p_old_short, p_new_short, "Skipped")))

# print bottom table border and closing statement
print("└─" + helper_str("─┴─").format(*["─" * c["w"] for c in _cols]) + "─┘")
print()
print("Done. Memos exported to: {}".format(args.export_path))
print()
print("\n--- SUMMARY ---")
print("Successfully Exported: {}".format(success_count))
print("Failed/Inconsistent: {}".format(failed_count))
print("Log file saved at: {}".format(log_path))
print("\nDone. Check the folder: {}".format(args.export_path))

# open finder if desired
if not args.no_finder:
subprocess.Popen(["open", args.export_path])


if __name__ == '__main__':
main()