calvin.erfmann/WhatsApp-Chat-Exporter
1
0
Fork 0
mirror of https://github.com/KnugiHK/WhatsApp-Chat-Exporter.git synced 2026-04-27 00:11:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Modular scripts

Browse Source
This commit is contained in:
KnugiHK
2021-01-18 15:24:17 +08:00
parent 32050714af
commit cfb180220c
2 changed files with 465 additions and 381 deletions
Download Patch File Download Diff File Expand all files Collapse all files

459
extract.py
View File

@@ -21,12 +21,9 @@ def determine_day(last, current):
else: else:
return current return current
data = {} def contacts(db, data):
# Get contacts
# Get contacts c = db.cursor()
if os.path.isfile("wa.db"):
wa = sqlite3.connect("wa.db")
c = wa.cursor()
c.execute("""SELECT count() FROM wa_contacts""") c.execute("""SELECT count() FROM wa_contacts""")
total_row_number = c.fetchone()[0] total_row_number = c.fetchone()[0]
print(f"Gathering contacts...({total_row_number})") print(f"Gathering contacts...({total_row_number})")
@@ -36,225 +33,271 @@ if os.path.isfile("wa.db"):
while row is not None: while row is not None:
data[row[0]] = {"name": row[1], "messages":{}} data[row[0]] = {"name": row[1], "messages":{}}
row = c.fetchone() row = c.fetchone()
wa.close()
# Get message history def messages(db, data):
msg = sqlite3.connect("msgstore.db") # Get message history
c = msg.cursor() c = db.cursor()
c.execute("""SELECT count() FROM messages""") c.execute("""SELECT count() FROM messages""")
total_row_number = c.fetchone()[0] total_row_number = c.fetchone()[0]
print(f"Gathering messages...(0/{total_row_number})", end="\r") print(f"Gathering messages...(0/{total_row_number})", end="\r")
phone_number_re = re.compile(r"[0-9]+@s.whatsapp.net") phone_number_re = re.compile(r"[0-9]+@s.whatsapp.net")
c.execute("""SELECT messages.key_remote_jid, messages._id, messages.key_from_me, messages.timestamp, messages.data, messages.status, messages.edit_version, messages.thumb_image, messages.remote_resource, messages.media_wa_type, messages.latitude, messages.longitude, messages_quotes.key_id as quoted, messages.key_id, messages_quotes.data, messages.media_caption FROM messages LEFT JOIN messages_quotes ON messages.quoted_row_id = messages_quotes._id; """) c.execute("""SELECT messages.key_remote_jid, messages._id, messages.key_from_me, messages.timestamp, messages.data, messages.status, messages.edit_version, messages.thumb_image, messages.remote_resource, messages.media_wa_type, messages.latitude, messages.longitude, messages_quotes.key_id as quoted, messages.key_id, messages_quotes.data, messages.media_caption FROM messages LEFT JOIN messages_quotes ON messages.quoted_row_id = messages_quotes._id; """)
i = 0 i = 0
content = c.fetchone() content = c.fetchone()
while content is not None: while content is not None:
if content[0] not in data: if content[0] not in data:
data[content[0]] = {"name": None, "messages": {}} data[content[0]] = {"name": None, "messages": {}}
data[content[0]]["messages"][content[1]] = { data[content[0]]["messages"][content[1]] = {
"from_me": bool(content[2]), "from_me": bool(content[2]),
"timestamp": content[3]/1000, "timestamp": content[3]/1000,
"time": datetime.fromtimestamp(content[3]/1000).strftime("%H:%M"), "time": datetime.fromtimestamp(content[3]/1000).strftime("%H:%M"),
"media": False, "media": False,
"key_id": content[13] "key_id": content[13]
} }
if "-" in content[0] and content[2] == 0: if "-" in content[0] and content[2] == 0:
name = None name = None
if content[8] in data: if content[8] in data:
name = data[content[8]]["name"] name = data[content[8]]["name"]
if "@" in content[8]: if "@" in content[8]:
fallback = content[8].split('@')[0] fallback = content[8].split('@')[0]
else:
fallback = None
else: else:
fallback = None fallback = None
data[content[0]]["messages"][content[1]]["sender"] = name or fallback
else: else:
fallback = None data[content[0]]["messages"][content[1]]["sender"] = None
data[content[0]]["messages"][content[1]]["sender"] = name or fallback if content[12] is not None:
else: data[content[0]]["messages"][content[1]]["reply"] = content[12]
data[content[0]]["messages"][content[1]]["sender"] = None data[content[0]]["messages"][content[1]]["quoted_data"] = content[14]
else:
if content[12] is not None: data[content[0]]["messages"][content[1]]["reply"] = None
data[content[0]]["messages"][content[1]]["reply"] = content[12]
data[content[0]]["messages"][content[1]]["quoted_data"] = content[14] if content[15] is not None:
else: data[content[0]]["messages"][content[1]]["caption"] = content[15]
data[content[0]]["messages"][content[1]]["reply"] = None else:
data[content[0]]["messages"][content[1]]["caption"] = None
if content[15] is not None:
data[content[0]]["messages"][content[1]]["caption"] = content[15] if content[5] == 6:
else: if "-" in content[0]:
data[content[0]]["messages"][content[1]]["caption"] = None # Is Group
if content[4] is not None:
if content[5] == 6: try:
if "-" in content[0]: int(content[4])
# Is Group except:
if content[4] is not None: data[content[0]]["messages"][content[1]]["data"] = "{The group name changed to "f"{content[4]}"" }"
try: else:
int(content[4])
except:
data[content[0]]["messages"][content[1]]["data"] = "{The group name changed to "f"{content[4]}"" }"
else:
del data[content[0]]["messages"][content[1]]
else:
thumb_image = content[7]
if thumb_image is not None:
if b"\x00\x00\x01\x74\x00\x1A" in thumb_image:
# Add user
added = phone_number_re.search(thumb_image.decode("unicode_escape"))[0]
if added in data:
name_right = data[added]["name"]
else:
name_right = added.split('@')[0]
if content[8] is not None:
if content[8] in data:
name_left = data[content[8]]["name"]
else:
name_left = content[8].split('@')[0]
data[content[0]]["messages"][content[1]]["data"] = "{"f"{name_left}"f" added {name_right or 'You'}""}"
else:
data[content[0]]["messages"][content[1]]["data"] = "{"f"Added {name_right or 'You'}""}"
if b"\xac\xed\x00\x05\x74\x00" in thumb_image:
# Changed number
original = content[8].split('@')[0]
changed = thumb_image[7:].decode().split('@')[0]
data[content[0]]["messages"][content[1]]["data"] = "{"f"{original} changed to {changed}""}"
else:
if content[4] is None:
del data[content[0]]["messages"][content[1]] del data[content[0]]["messages"][content[1]]
else:
# Private chat
if content[4] is None and content[7] is None:
del data[content[0]]["messages"][content[1]]
else:
if content[2] == 1:
if content[5] == 5 and content[6] == 7:
data[content[0]]["messages"][content[1]]["data"] = "{Message deleted}"
else:
if content[9] == "5":
data[content[0]]["messages"][content[1]]["data"] = "{ Location shared: "f"{content[10], content[11]}"" }"
else: else:
data[content[0]]["messages"][content[1]]["data"] = content[4] thumb_image = content[7]
else: if thumb_image is not None:
if content[5] == 0 and content[6] == 7: if b"\x00\x00\x01\x74\x00\x1A" in thumb_image:
data[content[0]]["messages"][content[1]]["data"] = "{Message deleted}" # Add user
added = phone_number_re.search(thumb_image.decode("unicode_escape"))[0]
if added in data:
name_right = data[added]["name"]
else:
name_right = added.split('@')[0]
if content[8] is not None:
if content[8] in data:
name_left = data[content[8]]["name"]
else:
name_left = content[8].split('@')[0]
data[content[0]]["messages"][content[1]]["data"] = "{"f"{name_left}"f" added {name_right or 'You'}""}"
else:
data[content[0]]["messages"][content[1]]["data"] = "{"f"Added {name_right or 'You'}""}"
if b"\xac\xed\x00\x05\x74\x00" in thumb_image:
# Changed number
original = content[8].split('@')[0]
changed = thumb_image[7:].decode().split('@')[0]
data[content[0]]["messages"][content[1]]["data"] = "{"f"{original} changed to {changed}""}"
else:
if content[4] is None:
del data[content[0]]["messages"][content[1]]
else: else:
if content[9] == "5": # Private chat
data[content[0]]["messages"][content[1]]["data"] = "{ Location shared: "f"{content[10], content[11]}"" }" if content[4] is None and content[7] is None:
else: del data[content[0]]["messages"][content[1]]
data[content[0]]["messages"][content[1]]["data"] = content[4]
i += 1
if i % 1000 == 0:
print(f"Gathering messages...({i}/{total_row_number})", end="\r")
content = c.fetchone()
print(f"Gathering messages...({total_row_number}/{total_row_number})", end="\r")
# Get media
c.execute("""SELECT count() FROM message_media""")
total_row_number = c.fetchone()[0]
print(f"\nGathering media...(0/{total_row_number})", end="\r")
i = 0
c.execute("""SELECT messages.key_remote_jid, message_row_id, file_path, message_url, mime_type, media_key FROM message_media INNER JOIN messages ON message_media.message_row_id = messages._id ORDER BY messages.key_remote_jid ASC""")
content = c.fetchone()
mime = MimeTypes()
while content is not None:
file_path = f"WhatsApp/{content[2]}"
data[content[0]]["messages"][content[1]]["media"] = True
if os.path.isfile(file_path):
data[content[0]]["messages"][content[1]]["data"] = file_path
if content[4] is None:
guess = mime.guess_type(file_path)[0]
if guess is not None:
data[content[0]]["messages"][content[1]]["mime"] = guess
else:
data[content[0]]["messages"][content[1]]["mime"] = "data/data"
else: else:
data[content[0]]["messages"][content[1]]["mime"] = content[4] if content[2] == 1:
else: if content[5] == 5 and content[6] == 7:
# if "https://mmg" in content[4]: data[content[0]]["messages"][content[1]]["data"] = "{Message deleted}"
# try: else:
# r = requests.get(content[3]) if content[9] == "5":
# if r.status_code != 200: data[content[0]]["messages"][content[1]]["data"] = "{ Location shared: "f"{content[10], content[11]}"" }"
# raise RuntimeError() else:
# except: data[content[0]]["messages"][content[1]]["data"] = content[4]
# data[content[0]]["messages"][content[1]]["data"] = "{The media is missing}" else:
# data[content[0]]["messages"][content[1]]["media"] = True if content[5] == 0 and content[6] == 7:
# data[content[0]]["messages"][content[1]]["mime"] = "media" data[content[0]]["messages"][content[1]]["data"] = "{Message deleted}"
# else: else:
data[content[0]]["messages"][content[1]]["data"] = "{The media is missing}" if content[9] == "5":
data[content[0]]["messages"][content[1]]["mime"] = "media" data[content[0]]["messages"][content[1]]["data"] = "{ Location shared: "f"{content[10], content[11]}"" }"
i += 1 else:
if i % 100 == 0: data[content[0]]["messages"][content[1]]["data"] = content[4]
print(f"Gathering media...({i}/{total_row_number})", end="\r")
i += 1
if i % 1000 == 0:
print(f"Gathering messages...({i}/{total_row_number})", end="\r")
content = c.fetchone()
print(f"Gathering messages...({total_row_number}/{total_row_number})", end="\r")
def media(db, data, media_folder):
# Get media
c = db.cursor()
c.execute("""SELECT count() FROM message_media""")
total_row_number = c.fetchone()[0]
print(f"\nGathering media...(0/{total_row_number})", end="\r")
i = 0
c.execute("""SELECT messages.key_remote_jid, message_row_id, file_path, message_url, mime_type, media_key FROM message_media INNER JOIN messages ON message_media.message_row_id = messages._id ORDER BY messages.key_remote_jid ASC""")
content = c.fetchone() content = c.fetchone()
print(f"Gathering media...({total_row_number}/{total_row_number})", end="\r") mime = MimeTypes()
while content is not None:
file_path = f"{media_folder}/{content[2]}"
data[content[0]]["messages"][content[1]]["media"] = True
if os.path.isfile(file_path):
data[content[0]]["messages"][content[1]]["data"] = file_path
if content[4] is None:
guess = mime.guess_type(file_path)[0]
if guess is not None:
data[content[0]]["messages"][content[1]]["mime"] = guess
else:
data[content[0]]["messages"][content[1]]["mime"] = "data/data"
else:
data[content[0]]["messages"][content[1]]["mime"] = content[4]
else:
# if "https://mmg" in content[4]:
# try:
# r = requests.get(content[3])
# if r.status_code != 200:
# raise RuntimeError()
# except:
# data[content[0]]["messages"][content[1]]["data"] = "{The media is missing}"
# data[content[0]]["messages"][content[1]]["media"] = True
# data[content[0]]["messages"][content[1]]["mime"] = "media"
# else:
data[content[0]]["messages"][content[1]]["data"] = "{The media is missing}"
data[content[0]]["messages"][content[1]]["mime"] = "media"
i += 1
if i % 100 == 0:
print(f"Gathering media...({i}/{total_row_number})", end="\r")
content = c.fetchone()
print(f"Gathering media...({total_row_number}/{total_row_number})", end="\r")
c.execute("""SELECT message_row_id, messages.key_remote_jid, vcard, messages.media_name FROM messages_vcards INNER JOIN messages ON messages_vcards.message_row_id = messages._id ORDER BY messages.key_remote_jid ASC""") def vcard(db, data):
rows = c.fetchall() c = db.cursor()
total_row_number = len(rows) c.execute("""SELECT message_row_id, messages.key_remote_jid, vcard, messages.media_name FROM messages_vcards INNER JOIN messages ON messages_vcards.message_row_id = messages._id ORDER BY messages.key_remote_jid ASC""")
print(f"\nGathering vCards...(0/{total_row_number})", end="\r") rows = c.fetchall()
base = "WhatsApp/vCards" total_row_number = len(rows)
for index, row in enumerate(rows): print(f"\nGathering vCards...(0/{total_row_number})", end="\r")
if not os.path.isdir(base): base = "WhatsApp/vCards"
os.mkdir(base) for index, row in enumerate(rows):
file_name = "".join(x for x in row[3] if x.isalnum()) if not os.path.isdir(base):
file_path = f"{base}/{file_name}.vcf" os.mkdir(base)
if not os.path.isfile(file_path): file_name = "".join(x for x in row[3] if x.isalnum())
with open(file_path, "w", encoding="utf-8") as f: file_path = f"{base}/{file_name}.vcf"
f.write(row[2]) if not os.path.isfile(file_path):
data[row[1]]["messages"][row[0]]["data"] = row[3] + "{ The vCard file cannot be displayed here, however it should be located at " + file_path + "}" with open(file_path, "w", encoding="utf-8") as f:
data[row[1]]["messages"][row[0]]["mime"] = "x-vcard" f.write(row[2])
print(f"Gathering vCards...({index + 1}/{total_row_number})", end="\r") data[row[1]]["messages"][row[0]]["data"] = row[3] + "{ The vCard file cannot be displayed here, however it should be located at " + file_path + "}"
data[row[1]]["messages"][row[0]]["mime"] = "x-vcard"
templateLoader = jinja2.FileSystemLoader(searchpath="./") print(f"Gathering vCards...({index + 1}/{total_row_number})", end="\r")
templateEnv = jinja2.Environment(loader=templateLoader)
templateEnv.globals.update(determine_day=determine_day)
TEMPLATE_FILE = "whatsapp.html"
template = templateEnv.get_template(TEMPLATE_FILE)
total_row_number = len(data) def create_html(data, output_folder):
print(f"\nCreating HTML...(0/{total_row_number})", end="\r") templateLoader = jinja2.FileSystemLoader(searchpath="./")
templateEnv = jinja2.Environment(loader=templateLoader)
templateEnv.globals.update(determine_day=determine_day)
TEMPLATE_FILE = "whatsapp.html"
template = templateEnv.get_template(TEMPLATE_FILE)
if len(sys.argv) < 3: total_row_number = len(data)
output_folder = "temp" print(f"\nCreating HTML...(0/{total_row_number})", end="\r")
else:
output_folder = sys.argv[2]
if not os.path.isdir(output_folder): if not os.path.isdir(output_folder):
os.mkdir(output_folder) os.mkdir(output_folder)
for current, i in enumerate(data): for current, i in enumerate(data):
if len(data[i]["messages"]) == 0: if len(data[i]["messages"]) == 0:
continue continue
phone_number = i.split('@')[0] phone_number = i.split('@')[0]
if "-"in i: if "-"in i:
file_name = "" file_name = ""
else: else:
file_name = phone_number file_name = phone_number
if data[i]["name"] is not None: if data[i]["name"] is not None:
if file_name != "": if file_name != "":
file_name += "-" file_name += "-"
file_name += data[i]["name"].replace("/", "-") file_name += data[i]["name"].replace("/", "-")
name = data[i]["name"] name = data[i]["name"]
else: else:
name = phone_number name = phone_number
safe_file_name = '' safe_file_name = ''
safe_file_name = "".join(x for x in file_name if x.isalnum() or x in "- ") safe_file_name = "".join(x for x in file_name if x.isalnum() or x in "- ")
with open(f"{output_folder}/{safe_file_name}.html", "w", encoding="utf-8") as f: with open(f"{output_folder}/{safe_file_name}.html", "w", encoding="utf-8") as f:
f.write(template.render(name=name, msgs=data[i]["messages"].values(), my_avatar=None, their_avatar=f"WhatsApp/Avatars/{i}.j")) f.write(template.render(name=name, msgs=data[i]["messages"].values(), my_avatar=None, their_avatar=f"WhatsApp/Avatars/{i}.j"))
if current % 10 == 0: if current % 10 == 0:
print(f"Creating HTML...({current}/{total_row_number})", end="\r") print(f"Creating HTML...({current}/{total_row_number})", end="\r")
print(f"Creating HTML...({total_row_number}/{total_row_number})", end="\r") print(f"Creating HTML...({total_row_number}/{total_row_number})", end="\r")
if not os.path.isdir(f"{output_folder}/WhatsApp"):
shutil.move("WhatsApp", f"{output_folder}/")
with open("result.json", "w") as f: if __name__ == "__main__":
data = json.dumps(data) from optparse import OptionParser
print(f"\nWriting JSON file...({int(len(data)/1024/1024)}MB)") parser = OptionParser()
f.write(data) parser.add_option(
"-w",
"--wa",
dest="wa",
default="wa.db",
help="Path to contact database")
parser.add_option(
"-m",
"--media",
dest="media",
default="WhatsApp",
help="Path to WhatsApp media folder"
)
# parser.add_option(
# "-t",
# "--template",
# dest="html",
# default="wa.db",
# help="Path to HTML template")
(options, args) = parser.parse_args()
msg_db = "msgstore.db"
output_folder = "temp"
contact_db = options.wa
media_folder = options.media
print("Everything is done!") if len(args) == 1:
msg_db = args[0]
elif len(args) == 2:
msg_db = args[0]
output_folder = args[1]
data = {}
if os.path.isfile(contact_db):
with sqlite3.connect(contact_db) as db:
contacts(db, data)
if os.path.isfile(msg_db):
with sqlite3.connect(msg_db) as db:
messages(db, data)
media(db, data, media_folder)
vcard(db, data)
create_html(data, output_folder)
if not os.path.isdir(f"{output_folder}/WhatsApp"):
shutil.move(media_folder, f"{output_folder}/")
with open("result.json", "w") as f:
data = json.dumps(data)
print(f"\nWriting JSON file...({int(len(data)/1024/1024)}MB)")
f.write(data)
print("Everything is done!")

387
extract_iphone.py
View File

@@ -11,6 +11,7 @@ import shutil
from datetime import datetime from datetime import datetime
from mimetypes import MimeTypes from mimetypes import MimeTypes
APPLE_TIME = datetime.timestamp(datetime(2001,1,1))
def determine_day(last, current): def determine_day(last, current):
last = datetime.fromtimestamp(last).date() last = datetime.fromtimestamp(last).date()
@@ -20,200 +21,240 @@ def determine_day(last, current):
else: else:
return current return current
data = {} def messages(db, data):
c = db.cursor()
# Get contacts
c.execute("""SELECT count() FROM ZWACHATSESSION""")
total_row_number = c.fetchone()[0]
print(f"Gathering contacts...({total_row_number})")
# Get contacts c.execute("""SELECT ZCONTACTJID, ZPARTNERNAME FROM ZWACHATSESSION; """)
msg = sqlite3.connect("7c7fba66680ef796b916b067077cc246adacf01d")
c = msg.cursor()
c.execute("""SELECT count() FROM ZWACHATSESSION""")
total_row_number = c.fetchone()[0]
print(f"Gathering contacts...({total_row_number})")
c.execute("""SELECT ZCONTACTJID, ZPARTNERNAME FROM ZWACHATSESSION; """)
row = c.fetchone()
while row is not None:
data[row[0]] = {"name": row[1], "messages":{}}
row = c.fetchone() row = c.fetchone()
while row is not None:
data[row[0]] = {"name": row[1], "messages":{}}
row = c.fetchone()
# Get message history # Get message history
c.execute("""SELECT count() FROM ZWAMESSAGE""") c.execute("""SELECT count() FROM ZWAMESSAGE""")
total_row_number = c.fetchone()[0] total_row_number = c.fetchone()[0]
apple_time = datetime.timestamp(datetime(2001,1,1)) print(f"Gathering messages...(0/{total_row_number})", end="\r")
print(f"Gathering messages...(0/{total_row_number})", end="\r")
c.execute("""SELECT COALESCE(ZFROMJID, ZTOJID), ZWAMESSAGE.Z_PK, ZISFROMME, ZMESSAGEDATE, ZTEXT, ZMESSAGETYPE, ZWAGROUPMEMBER.ZMEMBERJID FROM main.ZWAMESSAGE LEFT JOIN main.ZWAGROUPMEMBER ON main.ZWAMESSAGE.ZGROUPMEMBER = main.ZWAGROUPMEMBER.Z_PK;""") c.execute("""SELECT COALESCE(ZFROMJID, ZTOJID), ZWAMESSAGE.Z_PK, ZISFROMME, ZMESSAGEDATE, ZTEXT, ZMESSAGETYPE, ZWAGROUPMEMBER.ZMEMBERJID FROM main.ZWAMESSAGE LEFT JOIN main.ZWAGROUPMEMBER ON main.ZWAMESSAGE.ZGROUPMEMBER = main.ZWAGROUPMEMBER.Z_PK;""")
i = 0 i = 0
content = c.fetchone() content = c.fetchone()
while content is not None: while content is not None:
if content[0] not in data: if content[0] not in data:
data[content[0]] = {"name": None, "messages": {}} data[content[0]] = {"name": None, "messages": {}}
ts = apple_time + content[3] ts = APPLE_TIME + content[3]
data[content[0]]["messages"][content[1]] = { data[content[0]]["messages"][content[1]] = {
"from_me": bool(content[2]), "from_me": bool(content[2]),
"timestamp": ts, "timestamp": ts,
"time": datetime.fromtimestamp(ts).strftime("%H:%M"), "time": datetime.fromtimestamp(ts).strftime("%H:%M"),
"media": False, "media": False,
"reply": None, "reply": None,
"caption": None "caption": None
} }
if "-" in content[0] and content[2] == 0: if "-" in content[0] and content[2] == 0:
name = None name = None
if content[6] is not None: if content[6] is not None:
if content[6] in data: if content[6] in data:
name = data[content[6]]["name"] name = data[content[6]]["name"]
if "@" in content[6]: if "@" in content[6]:
fallback = content[6].split('@')[0] fallback = content[6].split('@')[0]
else:
fallback = None
else: else:
fallback = None fallback = None
data[content[0]]["messages"][content[1]]["sender"] = name or fallback
else: else:
fallback = None data[content[0]]["messages"][content[1]]["sender"] = None
data[content[0]]["messages"][content[1]]["sender"] = name or fallback if content[5] == 6:
else: # Metadata
data[content[0]]["messages"][content[1]]["sender"] = None if "-" in content[0]:
if content[5] == 6: # Group
# Metadata if content[4] is not None:
if "-" in content[0]: # Chnaged name
# Group try:
if content[4] is not None: int(content[4])
# Chnaged name except:
try: data[content[0]]["messages"][content[1]]["data"] = "{The group name changed to "f"{content[4]}"" }"
int(content[4]) else:
except: del data[content[0]]["messages"][content[1]]
data[content[0]]["messages"][content[1]]["data"] = "{The group name changed to "f"{content[4]}"" }"
else: else:
del data[content[0]]["messages"][content[1]] data[content[0]]["messages"][content[1]]["data"] = None
else: else:
data[content[0]]["messages"][content[1]]["data"] = None data[content[0]]["messages"][content[1]]["data"] = None
else: else:
data[content[0]]["messages"][content[1]]["data"] = None # real message
else: if content[2] == 1:
# real message if content[5] == 14:
if content[2] == 1: data[content[0]]["messages"][content[1]]["data"] = "{Message deleted}"
if content[5] == 14: else:
data[content[0]]["messages"][content[1]]["data"] = "{Message deleted}" data[content[0]]["messages"][content[1]]["data"] = content[4]
else: else:
data[content[0]]["messages"][content[1]]["data"] = content[4] if content[5] == 14:
else: data[content[0]]["messages"][content[1]]["data"] = "{Message deleted}"
if content[5] == 14: else:
data[content[0]]["messages"][content[1]]["data"] = "{Message deleted}" data[content[0]]["messages"][content[1]]["data"] = content[4]
else:
data[content[0]]["messages"][content[1]]["data"] = content[4] i += 1
if i % 1000 == 0:
i += 1 print(f"Gathering messages...({i}/{total_row_number})", end="\r")
if i % 1000 == 0: content = c.fetchone()
print(f"Gathering messages...({i}/{total_row_number})", end="\r") print(f"Gathering messages...({total_row_number}/{total_row_number})", end="\r")
def media(db, data, media_folder):
c = db.cursor()
# Get media
c.execute("""SELECT count() FROM ZWAMEDIAITEM""")
total_row_number = c.fetchone()[0]
print(f"\nGathering media...(0/{total_row_number})", end="\r")
i = 0
c.execute("""SELECT COALESCE(ZWAMESSAGE.ZFROMJID, ZWAMESSAGE.ZTOJID) as _id, ZMESSAGE, ZMEDIALOCALPATH, ZMEDIAURL, ZVCARDSTRING, ZMEDIAKEY, ZTITLE FROM ZWAMEDIAITEM INNER JOIN ZWAMESSAGE ON ZWAMEDIAITEM.ZMESSAGE = ZWAMESSAGE.Z_PK WHERE ZMEDIALOCALPATH IS NOT NULL ORDER BY _id ASC""")
content = c.fetchone() content = c.fetchone()
print(f"Gathering messages...({total_row_number}/{total_row_number})", end="\r") mime = MimeTypes()
# Get media while content is not None:
file_path = f"Message/{content[2]}"
c.execute("""SELECT count() FROM ZWAMEDIAITEM""") data[content[0]]["messages"][content[1]]["media"] = True
total_row_number = c.fetchone()[0]
print(f"\nGathering media...(0/{total_row_number})", end="\r") if os.path.isfile(file_path):
i = 0 data[content[0]]["messages"][content[1]]["data"] = file_path
c.execute("""SELECT COALESCE(ZWAMESSAGE.ZFROMJID, ZWAMESSAGE.ZTOJID) as _id, ZMESSAGE, ZMEDIALOCALPATH, ZMEDIAURL, ZVCARDSTRING, ZMEDIAKEY, ZTITLE FROM ZWAMEDIAITEM INNER JOIN ZWAMESSAGE ON ZWAMEDIAITEM.ZMESSAGE = ZWAMESSAGE.Z_PK WHERE ZMEDIALOCALPATH IS NOT NULL ORDER BY _id ASC""") if content[4] is None:
content = c.fetchone() guess = mime.guess_type(file_path)[0]
mime = MimeTypes() if guess is not None:
while content is not None: data[content[0]]["messages"][content[1]]["mime"] = guess
file_path = f"Message/{content[2]}" else:
data[content[0]]["messages"][content[1]]["media"] = True data[content[0]]["messages"][content[1]]["mime"] = "data/data"
if os.path.isfile(file_path):
data[content[0]]["messages"][content[1]]["data"] = file_path
if content[4] is None:
guess = mime.guess_type(file_path)[0]
if guess is not None:
data[content[0]]["messages"][content[1]]["mime"] = guess
else: else:
data[content[0]]["messages"][content[1]]["mime"] = "data/data" data[content[0]]["messages"][content[1]]["mime"] = content[4]
else: else:
data[content[0]]["messages"][content[1]]["mime"] = content[4] # if "https://mmg" in content[4]:
else: # try:
# if "https://mmg" in content[4]: # r = requests.get(content[3])
# try: # if r.status_code != 200:
# r = requests.get(content[3]) # raise RuntimeError()
# if r.status_code != 200: # except:
# raise RuntimeError() # data[content[0]]["messages"][content[1]]["data"] = "{The media is missing}"
# except: # data[content[0]]["messages"][content[1]]["mime"] = "media"
# data[content[0]]["messages"][content[1]]["data"] = "{The media is missing}" # else:
# data[content[0]]["messages"][content[1]]["mime"] = "media" data[content[0]]["messages"][content[1]]["data"] = "{The media is missing}"
# else: data[content[0]]["messages"][content[1]]["mime"] = "media"
data[content[0]]["messages"][content[1]]["data"] = "{The media is missing}" if content[6] is not None:
data[content[0]]["messages"][content[1]]["mime"] = "media" data[content[0]]["messages"][content[1]]["caption"] = content[6]
if content[6] is not None: i += 1
data[content[0]]["messages"][content[1]]["caption"] = content[6] if i % 100 == 0:
i += 1 print(f"Gathering media...({i}/{total_row_number})", end="\r")
if i % 100 == 0: content = c.fetchone()
print(f"Gathering media...({i}/{total_row_number})", end="\r") print(f"Gathering media...({total_row_number}/{total_row_number})", end="\r")
content = c.fetchone()
print(f"Gathering media...({total_row_number}/{total_row_number})", end="\r")
c.execute("""SELECT DISTINCT ZWAVCARDMENTION.ZMEDIAITEM, ZWAMEDIAITEM.ZMESSAGE, COALESCE(ZWAMESSAGE.ZFROMJID, ZWAMESSAGE.ZTOJID) as _id, ZVCARDNAME, ZVCARDSTRING FROM ZWAVCARDMENTION INNER JOIN ZWAMEDIAITEM ON ZWAVCARDMENTION.ZMEDIAITEM = ZWAMEDIAITEM.Z_PK INNER JOIN ZWAMESSAGE ON ZWAMEDIAITEM.ZMESSAGE = ZWAMESSAGE.Z_PK""") def vcard(db, data):
rows = c.fetchall() c = db.cursor()
total_row_number = len(rows) c.execute("""SELECT DISTINCT ZWAVCARDMENTION.ZMEDIAITEM, ZWAMEDIAITEM.ZMESSAGE, COALESCE(ZWAMESSAGE.ZFROMJID, ZWAMESSAGE.ZTOJID) as _id, ZVCARDNAME, ZVCARDSTRING FROM ZWAVCARDMENTION INNER JOIN ZWAMEDIAITEM ON ZWAVCARDMENTION.ZMEDIAITEM = ZWAMEDIAITEM.Z_PK INNER JOIN ZWAMESSAGE ON ZWAMEDIAITEM.ZMESSAGE = ZWAMESSAGE.Z_PK""")
print(f"\nGathering vCards...(0/{total_row_number})", end="\r") rows = c.fetchall()
base = "Message/vCards" total_row_number = len(rows)
for index, row in enumerate(rows): print(f"\nGathering vCards...(0/{total_row_number})", end="\r")
if not os.path.isdir(base): base = "Message/vCards"
os.mkdir(base) for index, row in enumerate(rows):
file_name = "".join(x for x in row[3] if x.isalnum()) if not os.path.isdir(base):
file_path = f"{base}/{file_name[:200]}.vcf" os.mkdir(base)
if not os.path.isfile(file_path): file_name = "".join(x for x in row[3] if x.isalnum())
with open(file_path, "w", encoding="utf-8") as f: file_path = f"{base}/{file_name[:200]}.vcf"
f.write(row[4]) if not os.path.isfile(file_path):
data[row[2]]["messages"][row[1]]["data"] = row[3] + "{ The vCard file cannot be displayed here, however it should be located at " + file_path + "}" with open(file_path, "w", encoding="utf-8") as f:
data[row[2]]["messages"][row[1]]["mime"] = "x-vcard" f.write(row[4])
data[row[2]]["messages"][row[1]]["media"] = True data[row[2]]["messages"][row[1]]["data"] = row[3] + "{ The vCard file cannot be displayed here, however it should be located at " + file_path + "}"
print(f"Gathering vCards...({index + 1}/{total_row_number})", end="\r") data[row[2]]["messages"][row[1]]["mime"] = "x-vcard"
data[row[2]]["messages"][row[1]]["media"] = True
print(f"Gathering vCards...({index + 1}/{total_row_number})", end="\r")
templateLoader = jinja2.FileSystemLoader(searchpath="./") def create_html(data, output_folder):
templateEnv = jinja2.Environment(loader=templateLoader) templateLoader = jinja2.FileSystemLoader(searchpath="./")
templateEnv.globals.update(determine_day=determine_day) templateEnv = jinja2.Environment(loader=templateLoader)
TEMPLATE_FILE = "whatsapp.html" templateEnv.globals.update(determine_day=determine_day)
template = templateEnv.get_template(TEMPLATE_FILE) TEMPLATE_FILE = "whatsapp.html"
template = templateEnv.get_template(TEMPLATE_FILE)
total_row_number = len(data) total_row_number = len(data)
print(f"\nCreating HTML...(0/{total_row_number})", end="\r") print(f"\nCreating HTML...(0/{total_row_number})", end="\r")
if len(sys.argv) < 3: if not os.path.isdir(output_folder):
output_folder = "temp" os.mkdir(output_folder)
else:
output_folder = sys.argv[2]
if not os.path.isdir(output_folder): for current, i in enumerate(data):
os.mkdir(output_folder) if len(data[i]["messages"]) == 0:
continue
phone_number = i.split('@')[0]
if "-"in i:
file_name = ""
else:
file_name = phone_number
if data[i]["name"] is not None:
if file_name != "":
file_name += "-"
file_name += data[i]["name"].replace("/", "-")
name = data[i]["name"]
else:
name = phone_number
safe_file_name = ''
safe_file_name = "".join(x for x in file_name if x.isalnum() or x in "- ")
with open(f"{output_folder}/{safe_file_name}.html", "w", encoding="utf-8") as f:
f.write(template.render(name=name, msgs=data[i]["messages"].values(), my_avatar=None, their_avatar=f"WhatsApp/Avatars/{i}.j"))
if current % 10 == 0:
print(f"Creating HTML...({current}/{total_row_number})", end="\r")
print(f"Creating HTML...({total_row_number}/{total_row_number})", end="\r")
for current, i in enumerate(data): if __name__ == "__main__":
if len(data[i]["messages"]) == 0: from optparse import OptionParser
continue parser = OptionParser()
phone_number = i.split('@')[0] parser.add_option(
if "-"in i: "-w",
file_name = "" "--wa",
else: dest="wa",
file_name = phone_number default="wa.db",
help="Path to contact database")
if data[i]["name"] is not None: parser.add_option(
if file_name != "": "-m",
file_name += "-" "--media",
file_name += data[i]["name"].replace("/", "-") dest="media",
name = data[i]["name"] default="Message",
else: help="Path to WhatsApp media folder"
name = phone_number )
# parser.add_option(
safe_file_name = '' # "-t",
safe_file_name = "".join(x for x in file_name if x.isalnum() or x in "- ") # "--template",
with open(f"{output_folder}/{safe_file_name}.html", "w", encoding="utf-8") as f: # dest="html",
f.write(template.render(name=name, msgs=data[i]["messages"].values(), my_avatar=None, their_avatar=f"WhatsApp/Avatars/{i}.j")) # default="wa.db",
if current % 10 == 0: # help="Path to HTML template")
print(f"Creating HTML...({current}/{total_row_number})", end="\r") (options, args) = parser.parse_args()
msg_db = "7c7fba66680ef796b916b067077cc246adacf01d"
print(f"Creating HTML...({total_row_number}/{total_row_number})", end="\r") output_folder = "temp"
contact_db = options.wa
media_folder = options.media
if not os.path.isdir(f"{output_folder}/Message"): if len(args) == 1:
shutil.move("Message", f"{output_folder}/") msg_db = args[0]
elif len(args) == 2:
msg_db = args[0]
output_folder = args[1]
data = {}
with open("result.json", "w") as f: if os.path.isfile(msg_db):
data = json.dumps(data) with sqlite3.connect(msg_db) as db:
print(f"\nWriting JSON file...({int(len(data)/1024/1024)}MB)") messages(db, data)
f.write(data) media(db, data, media_folder)
vcard(db, data)
create_html(data, output_folder)
print("Everything is done!") if not os.path.isdir(f"{output_folder}/WhatsApp"):
shutil.move(media_folder, f"{output_folder}/")
with open("result.json", "w") as f:
data = json.dumps(data)
print(f"\nWriting JSON file...({int(len(data)/1024/1024)}MB)")
f.write(data)
print("Everything is done!")