calvin.erfmann/WhatsApp-Chat-Exporter
1
0
Fork 0
mirror of https://github.com/KnugiHK/WhatsApp-Chat-Exporter.git synced 2026-05-21 16:28:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Rewrite a bit

Browse Source
This commit is contained in:
KnugiHK
2023-06-10 19:45:29 +08:00
parent 42435c38cc
commit 8d036a6d87
2 changed files with 72 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

72
Whatsapp_Chat_Exporter/extract.py
View File

@@ -254,12 +254,13 @@ def messages(db, data):
data[content["key_remote_jid"]] = ChatStore() data[content["key_remote_jid"]] = ChatStore()
if content["key_remote_jid"] is None: if content["key_remote_jid"] is None:
continue # Not sure continue # Not sure
data[content["key_remote_jid"]].add_message(content["_id"], Message( message = Message(
from_me=content["key_from_me"], from_me=content["key_from_me"],
timestamp=content["timestamp"], timestamp=content["timestamp"],
time=content["timestamp"], time=content["timestamp"],
key_id=content["key_id"], key_id=content["key_id"],
)) )
invalid = False
if "-" in content["key_remote_jid"] and content["key_from_me"] == 0: if "-" in content["key_remote_jid"] and content["key_from_me"] == 0:
name = None name = None
if table_message: if table_message:
@@ -282,24 +283,24 @@ def messages(db, data):
else: else:
fallback = None fallback = None
data[content["key_remote_jid"]].messages[content["_id"]].sender = name or fallback message.sender = name or fallback
else: else:
data[content["key_remote_jid"]].messages[content["_id"]].sender = None message.sender = None
if content["quoted"] is not None: if content["quoted"] is not None:
data[content["key_remote_jid"]].messages[content["_id"]].reply = content["quoted"] message.reply = content["quoted"]
data[content["key_remote_jid"]].messages[content["_id"]].quoted_data = content["quoted_data"] message.quoted_data = content["quoted_data"]
else: else:
data[content["key_remote_jid"]].messages[content["_id"]].reply = None message.reply = None
if not table_message and content["media_caption"] is not None: if not table_message and content["media_caption"] is not None:
# Old schema # Old schema
data[content["key_remote_jid"]].messages[content["_id"]].caption = content["media_caption"] message.caption = content["media_caption"]
elif table_message and content["message_type"] == 1 and content["data"] is not None: elif table_message and content["message_type"] == 1 and content["data"] is not None:
# New schema # New schema
data[content["key_remote_jid"]].messages[content["_id"]].caption = content["data"] message.caption = content["data"]
else: else:
data[content["key_remote_jid"]].messages[content["_id"]].caption = None message.caption = None
if content["status"] == 6: # 6 = Metadata, otherwise it's a message if content["status"] == 6: # 6 = Metadata, otherwise it's a message
if (not table_message and "-" in content["key_remote_jid"]) or \ if (not table_message and "-" in content["key_remote_jid"]) or \
@@ -310,10 +311,10 @@ def messages(db, data):
int(content["data"]) int(content["data"])
except ValueError: except ValueError:
msg = f"The group name changed to {content['data']}" msg = f"The group name changed to {content['data']}"
data[content["key_remote_jid"]].messages[content["_id"]].data = msg message.data = msg
data[content["key_remote_jid"]].messages[content["_id"]].meta = True message.meta = True
else: else:
data[content["key_remote_jid"]].delete_message(content["_id"]) invalid = True
else: else:
thumb_image = content["thumb_image"] thumb_image = content["thumb_image"]
if thumb_image is not None: if thumb_image is not None:
@@ -338,25 +339,25 @@ def messages(db, data):
original = content["remote_resource"].split('@')[0] original = content["remote_resource"].split('@')[0]
changed = thumb_image[7:].decode().split('@')[0] changed = thumb_image[7:].decode().split('@')[0]
msg = f"{original} changed to {changed}" msg = f"{original} changed to {changed}"
data[content["key_remote_jid"]].messages[content["_id"]].data = msg message.data = msg
data[content["key_remote_jid"]].messages[content["_id"]].meta = True message.meta = True
else: else:
if content["data"] is None: if content["data"] is None:
data[content["key_remote_jid"]].delete_message(content["_id"]) invalid = True
else: else:
# Private chat # Private chat
if content["data"] is None and content["thumb_image"] is None: if content["data"] is None and content["thumb_image"] is None:
data[content["key_remote_jid"]].delete_message(content["_id"]) invalid = True
else: else:
if content["key_from_me"] == 1: if content["key_from_me"] == 1:
if content["status"] == 5 and content["edit_version"] == 7 or table_message and content["message_type"] == 15: if content["status"] == 5 and content["edit_version"] == 7 or table_message and content["message_type"] == 15:
msg = "Message deleted" msg = "Message deleted"
data[content["key_remote_jid"]].messages[content["_id"]].meta = True message.meta = True
else: else:
if content["media_wa_type"] == "5": if content["media_wa_type"] == "5":
msg = f"Location shared: {content['latitude'], content['longitude']}" msg = f"Location shared: {content['latitude'], content['longitude']}"
data[content["key_remote_jid"]].messages[content["_id"]].meta = True message.meta = True
else: else:
msg = content["data"] msg = content["data"]
if msg is not None: if msg is not None:
@@ -367,11 +368,11 @@ def messages(db, data):
else: else:
if content["status"] == 0 and content["edit_version"] == 7 or table_message and content["message_type"] == 15: if content["status"] == 0 and content["edit_version"] == 7 or table_message and content["message_type"] == 15:
msg = "Message deleted" msg = "Message deleted"
data[content["key_remote_jid"]].messages[content["_id"]].meta = True message.meta = True
else: else:
if content["media_wa_type"] == "5": if content["media_wa_type"] == "5":
msg = f"Location shared: {content['latitude'], content['longitude']}" msg = f"Location shared: {content['latitude'], content['longitude']}"
data[content["key_remote_jid"]].messages[content["_id"]].meta = True message.meta = True
else: else:
msg = content["data"] msg = content["data"]
if msg is not None: if msg is not None:
@@ -379,9 +380,10 @@ def messages(db, data):
msg = msg.replace("\r\n", "<br>") msg = msg.replace("\r\n", "<br>")
if "\n" in msg: if "\n" in msg:
msg = msg.replace("\n", "<br>") msg = msg.replace("\n", "<br>")
message.data = msg
data[content["key_remote_jid"]].messages[content["_id"]].data = msg if not invalid:
data[content["key_remote_jid"]].add_message(content["_id"], message)
i += 1 i += 1
if i % 1000 == 0: if i % 1000 == 0:
print(f"Gathering messages...({i}/{total_row_number})", end="\r") print(f"Gathering messages...({i}/{total_row_number})", end="\r")
@@ -436,17 +438,18 @@ def media(db, data, media_folder):
mime = MimeTypes() mime = MimeTypes()
while content is not None: while content is not None:
file_path = f"{media_folder}/{content['file_path']}" file_path = f"{media_folder}/{content['file_path']}"
data[content["key_remote_jid"]].messages[content["message_row_id"]].media = True message = data[content["key_remote_jid"]].messages[content["message_row_id"]]
message.media = True
if os.path.isfile(file_path): if os.path.isfile(file_path):
data[content["key_remote_jid"]].messages[content["message_row_id"]].data = file_path message.data = file_path
if content["mime_type"] is None: if content["mime_type"] is None:
guess = mime.guess_type(file_path)[0] guess = mime.guess_type(file_path)[0]
if guess is not None: if guess is not None:
data[content["key_remote_jid"]].messages[content["message_row_id"]].mime = guess message.mime = guess
else: else:
data[content["key_remote_jid"]].messages[content["message_row_id"]].mime = "data/data" message.mime = "application/octet-stream"
else: else:
data[content["key_remote_jid"]].messages[content["message_row_id"]].mime = content["mime_type"] message.mime = content["mime_type"]
else: else:
# if "https://mmg" in content[4]: # if "https://mmg" in content[4]:
# try: # try:
@@ -458,9 +461,9 @@ def media(db, data, media_folder):
# data[content[0]]["messages"][content[1]]["media"] = True # data[content[0]]["messages"][content[1]]["media"] = True
# data[content[0]]["messages"][content[1]]["mime"] = "media" # data[content[0]]["messages"][content[1]]["mime"] = "media"
# else: # else:
data[content["key_remote_jid"]].messages[content["message_row_id"]].data = "The media is missing" message.data = "The media is missing"
data[content["key_remote_jid"]].messages[content["message_row_id"]].mime = "media" message.mime = "media"
data[content["key_remote_jid"]].messages[content["message_row_id"]].meta = True message.meta = True
i += 1 i += 1
if i % 100 == 0: if i % 100 == 0:
print(f"Gathering media...({i}/{total_row_number})", end="\r") print(f"Gathering media...({i}/{total_row_number})", end="\r")
@@ -510,11 +513,12 @@ def vcard(db, data):
if not os.path.isfile(file_path): if not os.path.isfile(file_path):
with open(file_path, "w", encoding="utf-8") as f: with open(file_path, "w", encoding="utf-8") as f:
f.write(row["vcard"]) f.write(row["vcard"])
data[row["key_remote_jid"]].messages[row["message_row_id"]].data = media_name + \ message = data[row["key_remote_jid"]].messages[row["message_row_id"]]
message.data = media_name + \
"The vCard file cannot be displayed here, " \ "The vCard file cannot be displayed here, " \
f"however it should be located at {file_path}" f"however it should be located at {file_path}"
data[row["key_remote_jid"]].messages[row["message_row_id"]].mime = "text/x-vcard" message.mime = "text/x-vcard"
data[row["key_remote_jid"]].messages[row["message_row_id"]].meta = True message.meta = True
print(f"Gathering vCards...({index + 1}/{total_row_number})", end="\r") print(f"Gathering vCards...({index + 1}/{total_row_number})", end="\r")

65
Whatsapp_Chat_Exporter/extract_iphone.py
View File

@@ -51,12 +51,13 @@ def messages(db, data):
if _id not in data: if _id not in data:
data[_id] = ChatStore() data[_id] = ChatStore()
ts = APPLE_TIME + content["ZMESSAGEDATE"] ts = APPLE_TIME + content["ZMESSAGEDATE"]
data[_id].add_message(Z_PK, Message( message = Message(
from_me=content["ZISFROMME"], from_me=content["ZISFROMME"],
timestamp=ts, timestamp=ts,
time=ts, # TODO: Could be bug time=ts, # TODO: Could be bug
key_id=content["ZSTANZAID"][:17], key_id=content["ZSTANZAID"][:17],
)) )
invalid = False
if "-" in _id and content["ZISFROMME"] == 0: if "-" in _id and content["ZISFROMME"] == 0:
name = None name = None
if content["ZMEMBERJID"] is not None: if content["ZMEMBERJID"] is not None:
@@ -68,9 +69,9 @@ def messages(db, data):
fallback = None fallback = None
else: else:
fallback = None fallback = None
data[_id].messages[Z_PK].sender = name or fallback message.sender = name or fallback
else: else:
data[_id].messages[Z_PK].sender = None message.sender = None
if content["ZMESSAGETYPE"] == 6: if content["ZMESSAGETYPE"] == 6:
# Metadata # Metadata
if "-" in _id: if "-" in _id:
@@ -81,25 +82,25 @@ def messages(db, data):
int(content["ZTEXT"]) int(content["ZTEXT"])
except ValueError: except ValueError:
msg = f"The group name changed to {content['ZTEXT']}" msg = f"The group name changed to {content['ZTEXT']}"
data[_id].messages[Z_PK].data = msg message.data = msg
data[_id].messages[Z_PK].meta = True message.meta = True
else: else:
del data[_id].messages[Z_PK] invalid = True
else: else:
data[_id].messages[Z_PK].data = None message.data = None
else: else:
data[_id].messages[Z_PK].data = None message.data = None
else: else:
# real message # real message
if content["ZMETADATA"] is not None and content["ZMETADATA"].startswith(b"\x2a\x14"): if content["ZMETADATA"] is not None and content["ZMETADATA"].startswith(b"\x2a\x14"):
quoted = content["ZMETADATA"][2:19] quoted = content["ZMETADATA"][2:19]
data[_id].messages[Z_PK].reply = quoted.decode() message.reply = quoted.decode()
data[_id].messages[Z_PK].quoted_data = None # TODO message.quoted_data = None # TODO
if content["ZISFROMME"] == 1: if content["ZISFROMME"] == 1:
if content["ZMESSAGETYPE"] == 14: if content["ZMESSAGETYPE"] == 14:
msg = "Message deleted" msg = "Message deleted"
data[_id].messages[Z_PK].meta = True message.meta = True
else: else:
msg = content["ZTEXT"] msg = content["ZTEXT"]
if msg is not None: if msg is not None:
@@ -110,7 +111,7 @@ def messages(db, data):
else: else:
if content["ZMESSAGETYPE"] == 14: if content["ZMESSAGETYPE"] == 14:
msg = "Message deleted" msg = "Message deleted"
data[_id].messages[Z_PK].meta = True message.meta = True
else: else:
msg = content["ZTEXT"] msg = content["ZTEXT"]
if msg is not None: if msg is not None:
@@ -118,7 +119,9 @@ def messages(db, data):
msg = msg.replace("\r\n", "<br>") msg = msg.replace("\r\n", "<br>")
if "\n" in msg: if "\n" in msg:
msg = msg.replace("\n", "<br>") msg = msg.replace("\n", "<br>")
data[_id].messages[Z_PK].data = msg message.data = msg
if not invalid:
data[_id].add_message(Z_PK, message)
i += 1 i += 1
if i % 1000 == 0: if i % 1000 == 0:
print(f"Gathering messages...({i}/{total_row_number})", end="\r") print(f"Gathering messages...({i}/{total_row_number})", end="\r")
@@ -152,18 +155,19 @@ def media(db, data, media_folder):
file_path = f"{media_folder}/{content['ZMEDIALOCALPATH']}" file_path = f"{media_folder}/{content['ZMEDIALOCALPATH']}"
_id = content["_id"] _id = content["_id"]
ZMESSAGE = content["ZMESSAGE"] ZMESSAGE = content["ZMESSAGE"]
data[_id].messages[ZMESSAGE].media = True message = data[_id].messages[ZMESSAGE]
message.media = True
if os.path.isfile(file_path): if os.path.isfile(file_path):
data[_id].messages[ZMESSAGE].data = file_path message.data = file_path
if content["ZVCARDSTRING"] is None: if content["ZVCARDSTRING"] is None:
guess = mime.guess_type(file_path)[0] guess = mime.guess_type(file_path)[0]
if guess is not None: if guess is not None:
data[_id].messages[ZMESSAGE].mime = guess message.mime = guess
else: else:
data[_id].messages[ZMESSAGE].mime = "data/data" message.mime = "application/octet-stream"
else: else:
data[_id].messages[ZMESSAGE].mime = content["ZVCARDSTRING"] message.mime = content["ZVCARDSTRING"]
else: else:
# if "https://mmg" in content["ZVCARDSTRING"]: # if "https://mmg" in content["ZVCARDSTRING"]:
# try: # try:
@@ -171,14 +175,14 @@ def media(db, data, media_folder):
# if r.status_code != 200: # if r.status_code != 200:
# raise RuntimeError() # raise RuntimeError()
# except: # except:
# data[_id].messages[ZMESSAGE].data"] = "{The media is missing}" # message.data"] = "{The media is missing}"
# data[_id].messages[ZMESSAGE].mime"] = "media" # message.mime"] = "media"
# else: # else:
data[_id].messages[ZMESSAGE].data = "The media is missing" message.data = "The media is missing"
data[_id].messages[ZMESSAGE].mime = "media" message.mime = "media"
data[_id].messages[ZMESSAGE].meta = True message.meta = True
if content["ZTITLE"] is not None: if content["ZTITLE"] is not None:
data[_id].messages[ZMESSAGE].caption = content["ZTITLE"] message.caption = content["ZTITLE"]
i += 1 i += 1
if i % 100 == 0: if i % 100 == 0:
print(f"Gathering media...({i}/{total_row_number})", end="\r") print(f"Gathering media...({i}/{total_row_number})", end="\r")
@@ -213,14 +217,13 @@ def vcard(db, data):
if not os.path.isfile(file_path): if not os.path.isfile(file_path):
with open(file_path, "w", encoding="utf-8") as f: with open(file_path, "w", encoding="utf-8") as f:
f.write(content["ZVCARDSTRING"]) f.write(content["ZVCARDSTRING"])
_id = content["_id"] message = data[content["_id"]].messages[content["ZMESSAGE"]]
ZMESSAGE = content["ZMESSAGE"] message.data = content["ZVCARDNAME"] + \
data[_id].messages[ZMESSAGE].data = content["ZVCARDNAME"] + \
"The vCard file cannot be displayed here, " \ "The vCard file cannot be displayed here, " \
f"however it should be located at {file_path}" f"however it should be located at {file_path}"
data[_id].messages[ZMESSAGE].mime = "text/x-vcard" message.mime = "text/x-vcard"
data[_id].messages[ZMESSAGE].media = True message.media = True
data[_id].messages[ZMESSAGE].meta = True message.meta = True
print(f"Gathering vCards...({index + 1}/{total_row_number})", end="\r") print(f"Gathering vCards...({index + 1}/{total_row_number})", end="\r")