#! /usr/bin/env python3 # WhatsApp Chat Archiver import sqlite3 import os import argparse import html from datetime import datetime, timedelta import shutil # WhatsApp's epoch starts on 2001-01-01 00:00:00 (Core Data timestamp) WHATSAPP_EPOCH = datetime(2001, 1, 1) def convert_whatsapp_timestamp(ts): """Converts WhatsApp's Core Data timestamp to a human-readable string.""" if not ts: return "" try: # Timestamps are seconds since the WhatsApp epoch dt = WHATSAPP_EPOCH + timedelta(seconds=ts) return dt.strftime('%Y-%m-%d %H:%M:%S') except (ValueError, TypeError): return "Invalid date" def get_media_tag(media_path, output_dir): """Generates the appropriate HTML tag for a given media file and copies it.""" if not media_path: return "" # Path in the DB is often relative like 'Media/WhatsApp Images/IMG-...' test_path = os.path.join(output_dir, 'Message', media_path) full_media_path = '' if not os.path.exists(test_path): return f'
Media not found: {html.escape(test_path)}
' full_media_path = os.path.join('Message', media_path) # remove ./ in the beginning if present full_media_path = full_media_path.lstrip('./') ext = os.path.splitext(media_path)[1].lower() if ext in ['.jpg', '.jpeg', '.png', '.gif', '.webp']: return f'Image' elif ext in ['.mp4', '.mov', '.webm']: return f'' elif ext in ['.mp3', '.ogg', '.opus', '.m4a']: return f'' else: return f'View Media: {os.path.basename(media_path)}' def generate_html_chat(db_path, media_path, output_dir, chat_id, chat_name, is_group, contact_jid): """Generates an HTML file for a single chat session.""" conn = sqlite3.connect(db_path) cursor = conn.cursor() # Updated query to fetch more potential name fields (like ZFIRSTNAME) to find the best one. query = """ SELECT m.ZISFROMME, m.ZTEXT, m.ZMESSAGEDATE, m.ZFROMJID, g.ZCONTACTNAME AS GroupMemberContactName, cs.ZPARTNERNAME AS ChatPartnerName, p.ZPUSHNAME AS ProfilePushName, mi.ZMEDIALOCALPATH, cs.ZCONTACTJID AS ChatJID FROM ZWAMESSAGE m LEFT JOIN ZWAGROUPMEMBER g ON m.ZGROUPMEMBER = g.Z_PK LEFT JOIN ZWACHATSESSION cs ON m.ZCHATSESSION = cs.Z_PK LEFT JOIN ZWAPROFILEPUSHNAME p ON m.ZFROMJID = p.ZJID LEFT JOIN ZWAMEDIAITEM mi ON m.ZMEDIAITEM = mi.Z_PK WHERE m.ZCHATSESSION = ? ORDER BY m.ZMESSAGEDATE ASC; """ cursor.execute(query, (chat_id,)) messages = cursor.fetchall() conn.close() if not messages: print(f"No messages found for chat: {chat_name}") return # Sanitize contact_jid for a unique and safe filename if contact_jid: safe_filename = "".join(c if c.isalnum() else "_" for c in contact_jid) else: # Fallback to chat_id if contact_jid is not available safe_filename = str(chat_id) chats_dir = os.path.join(output_dir, "chats") os.makedirs(chats_dir, exist_ok=True) html_filename = os.path.join(chats_dir, f"{safe_filename}.html") with open(html_filename, 'w', encoding='utf-8') as f: f.write(f""" Chat with {html.escape(chat_name)}
{html.escape(chat_name)}
{contact_jid}
""") # Write messages for is_from_me, text, timestamp, from_jid, group_member_contact_name, chat_partner_name, profile_push_name, media_local_path, contact_jid in messages: msg_class = "sent" if is_from_me else "received" f.write(f'
') # Determine and display the sender's name for incoming messages if not is_from_me: # Prioritize group member contact name for group chats if is_group: # Try names in order of preference, avoiding encoded-looking strings potential_names = [ group_member_contact_name, profile_push_name, from_jid, chat_partner_name, ] # Filter out None values and strings that look like they're encoded valid_names = [name for name in potential_names if name and not ( name.startswith('CK') and any(c.isupper() for c in name[2:]) and '=' in name )] sender_name = next((name for name in valid_names), "Unknown") else: # For individual chats, prefer partner name or push name sender_name = chat_partner_name or profile_push_name or from_jid or "Unknown" f.write(f'
{html.escape(str(sender_name))}
') if text: # Replace newline characters with
tags for proper display escaped_text = html.escape(text) f.write(f'
{escaped_text.replace(chr(10), "
")}
') if media_local_path: # print("Media path:", media_local_path) f.write(get_media_tag(media_local_path, output_dir)) f.write(f'
{convert_whatsapp_timestamp(timestamp)}
') f.write('
') f.write("""
""") print(f"Successfully generated HTML for: {chat_name}") # Step: iPhone backup manifest.db processing def process_iphone_backup(backup_path, output_dir): """ Processes the iPhone backup manifest.db, extracts WhatsApp shared files, and recreates the file structure in output_dir. Acts as an archiver to accumulate data across multiple imports without overwriting existing data. """ manifest_db_path = os.path.join(backup_path, 'Manifest.db') if not os.path.exists(manifest_db_path): print(f"Manifest.db not found in backup path: {manifest_db_path}") return # Connect to manifest.db and extract WhatsApp shared files backup_conn = sqlite3.connect(manifest_db_path) backup_cursor = backup_conn.cursor() backup_cursor.execute("SELECT fileID, domain, relativePath FROM Files WHERE domain = ?", ('AppDomainGroup-group.net.whatsapp.WhatsApp.shared',)) files = backup_cursor.fetchall() print(f"Found {len(files)} WhatsApp shared files in manifest.db.") backup_conn.close() # Count for statistics new_files = 0 updated_files = 0 skipped_files = 0 special_db_files = 0 # Check for SQLite database files that need special handling db_files_to_merge = [ 'ChatStorage.sqlite', 'CallHistory.sqlite', 'DeviceAgents.sqlite', 'Labels.sqlite', 'Ranking.sqlite', 'Sticker.sqlite' ] # Prepare to recreate file structure for fileID, domain, relativePath in files: src_file = os.path.join(backup_path, fileID[:2], fileID) dest_file = os.path.join(output_dir, relativePath) os.makedirs(os.path.dirname(dest_file), exist_ok=True) if not os.path.exists(src_file): print(f"Source file missing: {src_file}") skipped_files += 1 continue # Handle SQLite database files specially - merge data instead of overwriting file_basename = os.path.basename(dest_file) if file_basename in db_files_to_merge and os.path.exists(dest_file): special_db_files += 1 try: # For SQLite databases, we need to merge the data if file_basename == 'ChatStorage.sqlite': merge_chat_database(src_file, dest_file) else: # For other SQLite databases, make a backup and then replace # Future enhancement: implement proper merging for all database types backup_file = f"{dest_file}.backup_{datetime.now().strftime('%Y%m%d%H%M%S')}" shutil.copy2(dest_file, backup_file) print(f"Created backup of {file_basename} as {os.path.basename(backup_file)}") shutil.copy2(src_file, dest_file) except Exception as e: print(f"Error processing database {dest_file}: {e}") continue # For non-database files if os.path.exists(dest_file): # If file exists, we want to keep the newer one # For media files, we always keep them (accumulate data) is_media_file = any(relativePath.startswith(prefix) for prefix in ['Media/', 'Message/', 'ProfilePictures/', 'Avatar/']) if is_media_file: # For media files, don't overwrite but create a version with timestamp if different if not files_are_identical(src_file, dest_file): filename, ext = os.path.splitext(dest_file) timestamp = datetime.now().strftime('%Y%m%d%H%M%S') new_dest_file = f"{filename}_{timestamp}{ext}" try: shutil.copy2(src_file, new_dest_file) print(f"Saved additional version of media file: {os.path.relpath(new_dest_file, output_dir)}") new_files += 1 except Exception as e: print(f"Error copying alternate version {src_file}: {e}") skipped_files += 1 else: skipped_files += 1 else: # For non-media files, we'll take the newer one try: shutil.copy2(src_file, dest_file) updated_files += 1 except Exception as e: print(f"Error updating {dest_file}: {e}") skipped_files += 1 else: # If file doesn't exist, copy it try: shutil.copy2(src_file, dest_file) new_files += 1 except Exception as e: print(f"Error copying {src_file} to {dest_file}: {e}") skipped_files += 1 print(f"\nBackup import summary:") print(f"- Added {new_files} new files") print(f"- Updated {updated_files} existing files") print(f"- Special handling for {special_db_files} database files") print(f"- Skipped {skipped_files} files") def files_are_identical(file1, file2): """Compare two files to see if they are identical in content.""" if os.path.getsize(file1) != os.path.getsize(file2): return False # For larger files, just compare a sample to avoid reading entire files into memory if os.path.getsize(file1) > 1024*1024: # 1MB threshold with open(file1, 'rb') as f1, open(file2, 'rb') as f2: # Compare the first and last 4KB of the file start1 = f1.read(4096) start2 = f2.read(4096) if start1 != start2: return False f1.seek(-4096, 2) # 2 is os.SEEK_END f2.seek(-4096, 2) end1 = f1.read(4096) end2 = f2.read(4096) return end1 == end2 else: # For smaller files, read entire contents for comparison with open(file1, 'rb') as f1, open(file2, 'rb') as f2: return f1.read() == f2.read() def merge_chat_database(src_file, dest_file): """ Merge WhatsApp chat databases to combine messages from multiple backups. This preserves all existing messages and adds only new ones. """ print(f"Merging chat databases to preserve existing messages...") # Create a temporary copy for processing temp_file = f"{dest_file}.temp" shutil.copy2(dest_file, temp_file) try: # Connect to both databases src_conn = sqlite3.connect(src_file) dest_conn = sqlite3.connect(temp_file) # Make it safer by enabling foreign keys src_conn.execute("PRAGMA foreign_keys = OFF") dest_conn.execute("PRAGMA foreign_keys = OFF") # Get all messages from source src_cursor = src_conn.cursor() src_cursor.execute("SELECT Z_PK FROM ZWAMESSAGE") src_message_ids = {row[0] for row in src_cursor.fetchall()} # Get all messages from destination to avoid duplicates dest_cursor = dest_conn.cursor() dest_cursor.execute("SELECT Z_PK FROM ZWAMESSAGE") dest_message_ids = {row[0] for row in dest_cursor.fetchall()} # Find new message IDs that don't exist in the destination new_message_ids = src_message_ids - dest_message_ids if not new_message_ids: print("No new messages to import") src_conn.close() dest_conn.close() os.remove(temp_file) return print(f"Found {len(new_message_ids)} new messages to import") # Tables that need to be merged (simplified for this example) tables_to_check = [ "ZWAMESSAGE", "ZWAMEDIAITEM", "ZWAGROUPMEMBER", "ZWACHATSESSION", "ZWAPROFILEPUSHNAME" ] # For each table, copy new records for table in tables_to_check: # Check if table exists src_cursor.execute(f"SELECT name FROM sqlite_master WHERE type='table' AND name='{table}'") if not src_cursor.fetchone(): print(f"Table {table} doesn't exist in source database, skipping...") continue # Get column names src_cursor.execute(f"PRAGMA table_info({table})") columns = [row[1] for row in src_cursor.fetchall()] column_str = ", ".join(columns) # For each message ID, copy related records for msg_id in new_message_ids: # This is simplified - in reality you'd need more complex logic to follow foreign key relationships src_cursor.execute(f"SELECT {column_str} FROM {table} WHERE Z_PK = ?", (msg_id,)) rows = src_cursor.fetchall() for row in rows: # Skip existing records with same primary key dest_cursor.execute(f"SELECT 1 FROM {table} WHERE Z_PK = ?", (row[0],)) if dest_cursor.fetchone(): continue # Insert new record placeholders = ", ".join(["?" for _ in row]) dest_cursor.execute(f"INSERT OR IGNORE INTO {table} ({column_str}) VALUES ({placeholders})", row) # Commit changes dest_conn.commit() # Close connections src_conn.close() dest_conn.close() # Replace destination file with merged file os.rename(temp_file, dest_file) print(f"Successfully merged chat databases") except Exception as e: print(f"Error merging databases: {e}") if os.path.exists(temp_file): os.remove(temp_file) def main(): parser = argparse.ArgumentParser(description="WhatsApp Chat Exporter") parser.add_argument("--output", default="_html_export", help="Directory to save the HTML files.") parser.add_argument("--backup-path", default=None, help="Path to iPhone backup directory (for manifest.db processing)") args = parser.parse_args() if args.backup_path: process_iphone_backup(args.backup_path, args.output) # Use backup paths for archive creation db_path = os.path.join(args.output, "ChatStorage.sqlite") media_path = os.path.join(args.output, "Message/") else: parser.add_argument("db_path", help="Path to the ChatStorage.sqlite file.") parser.add_argument("media_path", help="Path to the root 'Media' directory.") args = parser.parse_args() db_path = args.db_path media_path = args.media_path if not os.path.exists(db_path): print(f"Error: Database file not found at '{db_path}'") return if not os.path.exists(media_path): print(f"Error: Media directory not found at '{media_path}'") return os.makedirs(args.output, exist_ok=True) conn = sqlite3.connect(db_path) cursor = conn.cursor() # Get all chats, joining with ZWAPROFILEPUSHNAME and using COALESCE to get the best possible name. cursor.execute(""" SELECT cs.Z_PK, COALESCE(p.ZPUSHNAME, cs.ZPARTNERNAME) AS ChatName, cs.ZCONTACTJID, cs.ZMESSAGECOUNTER, MIN(m.ZMESSAGEDATE) as FirstMessageDate, MAX(m.ZMESSAGEDATE) as LastMessageDate, COALESCE(gi.ZPICTUREPATH, pic.ZPATH) AS AvatarPath FROM ZWACHATSESSION cs LEFT JOIN ZWAPROFILEPUSHNAME p ON cs.ZCONTACTJID = p.ZJID LEFT JOIN ZWAMESSAGE m ON cs.Z_PK = m.ZCHATSESSION LEFT JOIN ZWAGROUPINFO gi ON cs.ZGROUPINFO = gi.Z_PK LEFT JOIN ZWAPROFILEPICTUREITEM pic ON cs.ZCONTACTJID = pic.ZJID WHERE cs.ZCONTACTJID NOT LIKE '%@status' GROUP BY cs.Z_PK, ChatName, cs.ZCONTACTJID, cs.ZMESSAGECOUNTER, AvatarPath ORDER BY LastMessageDate DESC NULLS LAST, ChatName """) chats = cursor.fetchall() conn.close() print(f"Found {len(chats)} chats to export.") index_path = os.path.join(args.output, "whatsapp-chats.html") with open(index_path, 'w', encoding='utf-8') as index_f: index_f.write(f""" WhatsApp Chat Export

WhatsApp Chat Export

Exported on {datetime.now().strftime('%Y-%m-%d %H:%M')}
") # Create a simple redirect index.html redirect_index = os.path.join(args.output, "index.html") with open(redirect_index, 'w', encoding='utf-8') as f: f.write(f""" Redirecting to WhatsApp Chats...

Redirecting to WhatsApp Chats...

""") print(f"\nExport complete!") print(f"View your chats by opening either of these files in your browser:") print(f" • {os.path.abspath(index_path)}") print(f" • {os.path.abspath(redirect_index)}") if __name__ == "__main__": main()