fix warnings

.gitignore

@@ -114,6 +114,10 @@ ENV/
 env.bak/
 venv.bak/
 
+# IntelliJ files
+/.idea
+/*.iml
+
 # Spyder project settings
 .spyderproject
 .spyproject
@@ -142,8 +146,9 @@ cython_debug/
 .DS_Store
 .vscode
 output/
-wa.db
-msgstore.db
+**/wa.db
+**/msgstore.db
 TODO.md
 test_locally.py
-tests/unit/tmp/*
+tests/unit/tmp/*
+/whatsapp_backup

main.py

@@ -15,6 +15,9 @@
     chats_to_txt_raw,
 )
 
+CALL_LOGS_DIR = "/call_logs"
+CHAT_DIR = "/chats"
+
 
 def create_db_connection(file_path: str) -> Tuple[sqlite3.Connection, sqlite3.Cursor]:
     """Create a database connection and return it.
@@ -80,7 +83,7 @@ def close_db_connections(databases: List[sqlite3.Connection]) -> None:
         "--backup_specific_or_all_chat_call",
         "-e",
         nargs="*",
-        default="all",
+        default=["all"],
         help="Phone numbers (format: XXXXXXXXXXXX) of the chats and/or call logs that you want to extract from the database",
     )
     args = ap.parse_args()
@@ -95,20 +98,20 @@ def close_db_connections(databases: List[sqlite3.Connection]) -> None:
     )
 
     if args.backup_output_style == "raw_txt":
-        if args.backup_specific_or_all_chat_call == "all":
+        if args.backup_specific_or_all_chat_call == ["all"]:
             if "chats" in args.backup_strategy:
-                output_chat_directory = args.parsed_backup_output_dir + "/chats"
+                output_chat_directory = args.parsed_backup_output_dir + CHAT_DIR
                 if not os.path.exists(output_chat_directory):
                     os.makedirs(output_chat_directory)
                 chats = chat_builder.build_all_chats(msgdb_cursor, wadb_cursor)
                 for chat in tqdm(chats):
                     chats_to_txt_raw(
                         chat=chat,
-                        dir=output_chat_directory,
+                        folder=output_chat_directory,
                     )
             if "call_logs" in args.backup_strategy:
                 output_call_logs_directory = (
-                    args.parsed_backup_output_dir + "/call_logs"
+                        args.parsed_backup_output_dir + CALL_LOGS_DIR
                 )
                 if not os.path.exists(output_call_logs_directory):
                     os.makedirs(output_call_logs_directory)
@@ -118,24 +121,24 @@ def close_db_connections(databases: List[sqlite3.Connection]) -> None:
                 for call_log in tqdm(call_logs):
                     if call_log.calls:
                         call_logs_to_txt_raw(
-                            call_log=call_log, dir=output_call_logs_directory
+                            call_log=call_log, folder=output_call_logs_directory
                         )
         else:
             for ph_no in tqdm(args.backup_specific_or_all_chat_call):
                 if "chats" in args.backup_strategy:
-                    output_chat_directory = args.parsed_backup_output_dir + "/chats"
+                    output_chat_directory = args.parsed_backup_output_dir + CHAT_DIR
                     if not os.path.exists(output_chat_directory):
                         os.makedirs(output_chat_directory)
                     chat = chat_builder.build_chat_for_given_id_or_phone_number(
                         msgdb_cursor, wadb_cursor, phone_number=ph_no
                     )
                     chats_to_txt_raw(
                         chat=chat,
-                        dir=output_chat_directory,
+                        folder=output_chat_directory,
                     )
                 if "call_logs" in args.backup_strategy:
                     output_call_logs_directory = (
-                        args.parsed_backup_output_dir + "/call_logs"
+                            args.parsed_backup_output_dir + CALL_LOGS_DIR
                     )
                     if not os.path.exists(output_call_logs_directory):
                         os.makedirs(output_call_logs_directory)
@@ -146,24 +149,24 @@ def close_db_connections(databases: List[sqlite3.Connection]) -> None:
                     )
                     if call_log.calls:
                         call_logs_to_txt_raw(
-                            call_log=call_log, dir=output_call_logs_directory
+                            call_log=call_log, folder=output_call_logs_directory
                         )
 
     elif args.backup_output_style == "formatted_txt":
-        if args.backup_specific_or_all_chat_call == "all":
+        if args.backup_specific_or_all_chat_call == ["all"]:
             if "chats" in args.backup_strategy:
-                output_chat_directory = args.parsed_backup_output_dir + "/chats"
+                output_chat_directory = args.parsed_backup_output_dir + CHAT_DIR
                 if not os.path.exists(output_chat_directory):
                     os.makedirs(output_chat_directory)
                 chats = chat_builder.build_all_chats(msgdb_cursor, wadb_cursor)
                 for chat in tqdm(chats):
                     chats_to_txt_formatted(
                         chat=chat,
-                        dir=output_chat_directory,
+                        folder=output_chat_directory,
                     )
             if "call_logs" in args.backup_strategy:
                 output_call_logs_directory = (
-                    args.parsed_backup_output_dir + "/call_logs"
+                        args.parsed_backup_output_dir + CALL_LOGS_DIR
                 )
                 if not os.path.exists(output_call_logs_directory):
                     os.makedirs(output_call_logs_directory)
@@ -173,24 +176,24 @@ def close_db_connections(databases: List[sqlite3.Connection]) -> None:
                 for call_log in tqdm(call_logs):
                     if call_log.calls:
                         call_logs_to_txt_formatted(
-                            call_log=call_log, dir=output_call_logs_directory
+                            call_log=call_log, folder=output_call_logs_directory
                         )
         else:
             for ph_no in tqdm(args.backup_specific_or_all_chat_call):
                 if "chats" in args.backup_strategy:
-                    output_chat_directory = args.parsed_backup_output_dir + "/chats"
+                    output_chat_directory = args.parsed_backup_output_dir + CHAT_DIR
                     if not os.path.exists(output_chat_directory):
                         os.makedirs(output_chat_directory)
                     chat = chat_builder.build_chat_for_given_id_or_phone_number(
                         msgdb_cursor, wadb_cursor, phone_number=ph_no
                     )
                     chats_to_txt_formatted(
                         chat=chat,
-                        dir=output_chat_directory,
+                        folder=output_chat_directory,
                     )
                 if "call_logs" in args.backup_strategy:
                     output_call_logs_directory = (
-                        args.parsed_backup_output_dir + "/call_logs"
+                            args.parsed_backup_output_dir + CALL_LOGS_DIR
                     )
                     if not os.path.exists(output_call_logs_directory):
                         os.makedirs(output_call_logs_directory)
@@ -201,24 +204,24 @@ def close_db_connections(databases: List[sqlite3.Connection]) -> None:
                     )
                     if call_log.calls:
                         call_logs_to_txt_formatted(
-                            call_log=call_log, dir=output_call_logs_directory
+                            call_log=call_log, folder=output_call_logs_directory
                         )
 
     elif args.backup_output_style == "json":
-        if args.backup_specific_or_all_chat_call == "all":
+        if args.backup_specific_or_all_chat_call == ["all"]:
             if "chats" in args.backup_strategy:
-                output_chat_directory = args.parsed_backup_output_dir + "/chats"
+                output_chat_directory = args.parsed_backup_output_dir + CHAT_DIR
                 if not os.path.exists(output_chat_directory):
                     os.makedirs(output_chat_directory)
                 chats = chat_builder.build_all_chats(msgdb_cursor, wadb_cursor)
                 for chat in tqdm(chats):
                     chats_to_json(
                         chat=chat,
-                        dir=output_chat_directory,
+                        folder=output_chat_directory,
                     )
             if "call_logs" in args.backup_strategy:
                 output_call_logs_directory = (
-                    args.parsed_backup_output_dir + "/call_logs"
+                        args.parsed_backup_output_dir + CALL_LOGS_DIR
                 )
                 if not os.path.exists(output_call_logs_directory):
                     os.makedirs(output_call_logs_directory)
@@ -228,24 +231,24 @@ def close_db_connections(databases: List[sqlite3.Connection]) -> None:
                 for call_log in tqdm(call_logs):
                     if call_log.calls:
                         call_logs_to_json(
-                            call_log=call_log, dir=output_call_logs_directory
+                            call_log=call_log, folder=output_call_logs_directory
                         )
         else:
             for ph_no in tqdm(args.backup_specific_or_all_chat_call):
                 if "chats" in args.backup_strategy:
-                    output_chat_directory = args.parsed_backup_output_dir + "/chats"
+                    output_chat_directory = args.parsed_backup_output_dir + CHAT_DIR
                     if not os.path.exists(output_chat_directory):
                         os.makedirs(output_chat_directory)
                     chat = chat_builder.build_chat_for_given_id_or_phone_number(
                         msgdb_cursor, wadb_cursor, phone_number=ph_no
                     )
                     chats_to_json(
                         chat=chat,
-                        dir=output_chat_directory,
+                        folder=output_chat_directory,
                     )
                 if "call_logs" in args.backup_strategy:
                     output_call_logs_directory = (
-                        args.parsed_backup_output_dir + "/call_logs"
+                            args.parsed_backup_output_dir + CALL_LOGS_DIR
                     )
                     if not os.path.exists(output_call_logs_directory):
                         os.makedirs(output_call_logs_directory)
@@ -256,11 +259,11 @@ def close_db_connections(databases: List[sqlite3.Connection]) -> None:
                     )
                     if call_log.calls:
                         call_logs_to_json(
-                            call_log=call_log, dir=output_call_logs_directory
+                            call_log=call_log, folder=output_call_logs_directory
                         )
 
     else:
         close_db_connections([msgdb, wadb])
-        raise Exception("Invalid 'chat formatting' requested")
+        raise AssertionError("Invalid 'chat formatting' requested")
 
     close_db_connections([msgdb, wadb])

requirements.txt

@@ -0,0 +1,2 @@
+tqdm
+attrs

src/call_log_extractor/builder.py

@@ -24,7 +24,7 @@ def build_call_for_given_id(
     if call_details:
         return Call(**call_details)
     else:
-        None
+        return None
 
 
 def build_call_log_for_given_id_or_phone_number(
@@ -53,18 +53,16 @@ def build_call_log_for_given_id_or_phone_number(
             msgdb_cursor=msgdb_cursor, phone_number=phone_number
         )
     else:
-        raise Exception("'jid_row_id' and 'phone_number' both cannot be None")
+        raise AssertionError("'jid_row_id' and 'phone_number' cannot both be None")
 
     dm_or_group = contact_resolver(
         wadb_cursor=wadb_cursor, raw_string_jid=raw_string_jid
     )
     call_log["caller_id"] = Contact(raw_string_jid=raw_string_jid, **dm_or_group)
 
     query = f"""SELECT call_log._id FROM 'call_log' WHERE call_log.jid_row_id={call_log.get("jid_row_id")}"""
-    exec = msgdb_cursor.execute(query)
-    res_query = list(chain.from_iterable(exec.fetchall()))
-    if res_query is None:
-        return None
+    execution = msgdb_cursor.execute(query)
+    res_query = list(chain.from_iterable(execution.fetchall()))
     call_log["calls"] = [
         build_call_for_given_id(msgdb_cursor, call_row_id)
         for call_row_id in sorted(res_query)
@@ -86,10 +84,8 @@ def build_all_call_logs(
         A generator of CallLog objects
     """
     query = "SELECT jid._id FROM 'jid'"
-    exec = msgdb_cursor.execute(query)
-    res_query = list(chain.from_iterable(exec.fetchall()))
-    if res_query is None:
-        return None
+    execution = msgdb_cursor.execute(query)
+    res_query = list(chain.from_iterable(execution.fetchall()))
 
     return (
         build_call_log_for_given_id_or_phone_number(

src/call_log_extractor/resolver.py

@@ -2,7 +2,7 @@
 from typing import Any, Dict, Tuple, Union
 
 
-def call_resolver(msgdb_cursor: sqlite3.Cursor, call_row_id: int) -> Dict[str, Any]:
+def call_resolver(msgdb_cursor: sqlite3.Cursor, call_row_id: int) -> Dict[str, Any] | None:
     """Fetch call data for a given call_row_id from the msgdb.
 
     Args:
@@ -16,11 +16,11 @@ def call_resolver(msgdb_cursor: sqlite3.Cursor, call_row_id: int) -> Dict[str, A
     SELECT call_log._id as call_row_id, call_log.from_me, call_log.timestamp, call_log.video_call, call_log.duration, call_log.call_result
     FROM 'call_log'
     WHERE call_log._id={call_row_id}"""
-    exec = msgdb_cursor.execute(msgdb_query)
-    res_query = exec.fetchone()
+    execution = msgdb_cursor.execute(msgdb_query)
+    res_query = execution.fetchone()
     if res_query is None:
         return None
-    res = dict(zip([col[0] for col in exec.description], res_query))
+    res = dict(zip([col[0] for col in execution.description], res_query))
     return res
 
 
@@ -51,15 +51,15 @@ def call_jid_resolver(
         FROM 'jid'
         WHERE jid.raw_string LIKE '%{phone_number}@%'"""
     else:
-        raise Exception("'jid_row_id' and 'phone_number' both cannot be None")
+        raise AssertionError("'jid_row_id' and 'phone_number' both cannot be None")
 
-    exec = msgdb_cursor.execute(msgdb_query)
-    res_query = exec.fetchone()
+    execution = msgdb_cursor.execute(msgdb_query)
+    res_query = execution.fetchone()
     if res_query is None:
         res_query = [
             None,
             None,
         ]  # Need some better logic to resolve when we don't have a contact in msgdb.db
-    res = dict(zip([col[0] for col in exec.description], res_query))
+    res = dict(zip([col[0] for col in execution.description], res_query))
     raw_string_jid = res.pop("raw_string_jid")
     return res, raw_string_jid

src/chat_extractor/builder.py

@@ -84,7 +84,7 @@ def build_chat_for_given_id_or_phone_number(
             msgdb_cursor=msgdb_cursor, phone_number=phone_number
         )
     else:
-        raise Exception("'chat_row_id' and 'phone_number' both cannot be None")
+        raise AssertionError("'chat_row_id' and 'phone_number' both cannot be None")
 
     dm_or_group = contact_resolver(
         wadb_cursor=wadb_cursor, raw_string_jid=raw_string_jid
@@ -97,10 +97,8 @@ def build_chat_for_given_id_or_phone_number(
         )
 
     query = f"""SELECT message._id FROM 'message' WHERE message.chat_row_id={chat.get("chat_id")}"""
-    exec = msgdb_cursor.execute(query)
-    res_query = list(chain.from_iterable(exec.fetchall()))
-    if res_query is None:
-        return None
+    execution = msgdb_cursor.execute(query)
+    res_query = list(chain.from_iterable(execution.fetchall()))
     chat["messages"] = [
         build_message_for_given_id(msgdb_cursor, wadb_cursor, message_id)
         for message_id in res_query
@@ -125,10 +123,8 @@ def build_all_chats(
         A generator of Chat objects.
     """
     query = "SELECT chat._id FROM 'chat'"
-    exec = msgdb_cursor.execute(query)
-    res_query = list(chain.from_iterable(exec.fetchall()))
-    if res_query is None:
-        return None
+    execution = msgdb_cursor.execute(query)
+    res_query = list(chain.from_iterable(execution.fetchall()))
 
     return (
         build_chat_for_given_id_or_phone_number(