waweb.py

from datetime import datetime
import base64
import os
import time
import emoji
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.by import By

USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36"

chrome_options = webdriver.ChromeOptions()
chrome_options.add_argument("--no-sandbox")
chrome_options.add_argument("--disable-dev-shm-usage")
chrome_options.add_argument(f"--user-data-dir={os.getcwd()}/chrome_user_data")
chrome_options.add_argument("--headless=new")
chrome_options.add_argument("--remote-debugging-port=9222")  # helps fix DevToolsActivePort error
chrome_options.add_argument("--start-maximized")
chrome_options.add_argument(f"--user-agent={USER_AGENT}")

# disable telemetry (not cool selenium)
os.environ['SE_AVOID_STATS'] = 'true'

chrome_path = "" # your chrome path here
chromedriver_path = "" # your chromedriver path here

if chromedriver_path and chrome_path:
    chrome_options.binary_location = chrome_path
    service = Service(executable_path=chromedriver_path)
    driver = webdriver.Chrome(service=service, options=chrome_options)
else:
    driver = webdriver.Chrome(options=chrome_options)

driver.get("https://web.whatsapp.com/")
print("Chromedriver Version: ", driver.capabilities["chrome"]["chromedriverVersion"])
media_download = {}

def login():
    if os.path.exists("static/images/qrcode.png"):
        os.remove("static/images/qrcode.png")
    WebDriverWait(driver, 30).until(EC.presence_of_element_located((By.TAG_NAME, 'canvas')))

    # this is the qr
    qr_elm = driver.find_element(By.TAG_NAME, 'canvas')
    canvas_base64 = driver.execute_script("return arguments[0].toDataURL('image/png').substring(21);", qr_elm)
    canvas_png = base64.b64decode(canvas_base64)

    # write qr png data to actual png
    directory = "static/images"
    if not os.path.exists(directory):
        os.makedirs(directory)
        print(f"Directory '{directory}' created.")
    with open("static/images/qrcode.png", "wb") as f:
        f.write(canvas_png)

def logged_in():
    return driver.execute_script("""
        return !!(window.localStorage.getItem('last-wid-md') ||
                  window.localStorage.getItem('last-wid') ||
                  window.localStorage.getItem('WALid'));
    """)

def logout():
    driver.execute_script("window.Store.AppState.logout();")

def process_num():
    contacts = driver.execute_script("return window.Store.Chat.map(contacts => contacts.formattedTitle);")
    contact_num = driver.execute_script("return window.Store.Chat.map(contacts => contacts.id._serialized);")

    return zip(contacts,contact_num)

def chat_session(num):
    msgdata = driver.execute_script(f"""return document.msgdata = window.Store.Chat.get('{num}').msgs._models.map(m => ({{
                                            body: m.body,
                                            timestamp: m.t,
                                            from: (m.from.server == "g.us" ? null : window.Store.Contact.get(m.from?._serialized)?.name)
                                                || window.Store.Contact.get(m.author?._serialized)?.name
                                                || m.senderObj.verifiedName || m.senderObj.pushname,
                                            type: m.type,
                                            filename: m.filename || "",
                                            mimetype: m.mimetype,
                                            caption: m.caption || "",
                                            directPath: m.directPath,
                                            encFilehash: m.encFilehash,
                                            filehash: m.filehash,
                                            mediaKey: m.mediaKey,
                                            mediaKeyTimestamp: m.mediaKeyTimestamp,
                                        }}));
    """)
    messages = gather_msg(msgdata)
    # using name in contact (if available) or whatsapp name (verified for business acc, push for non business acc)
    who = driver.execute_script("return document.msgdata.map(m => m.from);")
    time = [datetime.fromtimestamp(timestamp["timestamp"]).time().strftime("%H:%M") for timestamp in msgdata]

    messages.reverse()
    who.reverse()
    time.reverse()

    who_msg_t = list(zip(who, messages, time))
    return who_msg_t

def down(num):
   error = ""
   tries = 0
   backoff = 0.50
   driver.execute_script(f"document.lengthc = await window.Store.Chat.find('{num}')")
   length_old = driver.execute_script("return document.lengthc.msgs.length")
   length_new = driver.execute_script("return document.lengthc.msgs.length")
   h_code = load_history(num)

   if h_code == 1:
      error = "No more messages to sync"
   elif h_code == 4:
      error = "Cannot sync history, only available on your phone."
   else:
       while length_old == length_new and tries != 3:
           h_code = load_history(num)
           load_msg(num)
           driver.execute_script("window.Store.Cmd.closeActiveChat()")
           length_new = driver.execute_script("return document.lengthc.msgs.length")
           tries+=1
           time.sleep(backoff)
           backoff+=1.50
       if length_old == length_new:
           error = "History sync failed, please try again."
   return error

# everything needed for all current features
def preload():
    print("hey it works")
    driver.execute_script("window.Store = Object.assign({}, window.require('WAWebCollections'));")
    # history
    driver.execute_script("window.Store.Cmd = window.require('WAWebCmd').Cmd;")
    driver.execute_script("window.Store.WidFactory = window.require('WAWebWidFactory');")
    driver.execute_script("window.Store.HistorySync = window.require('WAWebSendNonMessageDataRequest');")
    # load msg
    driver.execute_script("window.Store.ConversationMsgs = window.require('WAWebChatLoadMessages');")
    # send
    driver.execute_script("window.Store.User = window.require('WAWebUserPrefsMeUser');")
    driver.execute_script("window.Store.MsgKey = window.require('WAWebMsgKey');")
    driver.execute_script("window.Store.SendMessage = window.require('WAWebSendMsgChatAction');")
    driver.execute_script("window.Store.MediaObject = window.require('WAWebMediaStorage');")
    driver.execute_script("window.Store.OpaqueData = window.require('WAWebMediaOpaqueData');")
    driver.execute_script("window.Store.MediaTypes = window.require('WAWebMmsMediaTypes');")
    driver.execute_script("window.Store.MediaPrep = window.require('WAWebPrepRawMedia');")
    driver.execute_script("window.Store.MediaUpload = window.require('WAWebMediaMmsV4Upload');")
    # get media
    driver.execute_script("window.Store.DownloadManager = window.require('WAWebDownloadManager').downloadManager;")
    # logout
    driver.execute_script("window.Store.AppState = window.require('WAWebSocketModel').Socket;")

def load_history(num):
    driver.execute_script(f"document.chatWid = window.Store.WidFactory.createWid('{num}');")
    driver.execute_script("document.chat = window.Store.Chat.get(document.chatWid) ?? (await window.Store.Chat.find(document.chatWid));")
    driver.execute_script("await window.Store.Cmd.openChatBottom({'chat':document.chat});")
    h_code = driver.execute_script("""if(document.chat.endOfHistoryTransferType == 0){
                                            await window.Store.HistorySync.sendPeerDataOperationRequest(3, {
                                            chatId: document.chat.id
                                        });
                                      }
                                      return document.chat.endOfHistoryTransferType;
    """)

    return h_code

def load_msg(num):
    driver.execute_script(f"document.chat = window.Store.Chat.get('{num}');")
    driver.execute_script("await window.Store.ConversationMsgs.loadEarlierMsgs(document.chat);")

def load_chat(num):
    latest_msg = driver.execute_script(f"""return window.Store.Chat.get('{num}').msgs._models.slice(-1)
                                            .map(m => ({{
                                                body: m.body,
                                                timestamp: m.t,
                                                from: m.from,
                                                type: m.type,
                                                filename: m.filename || "",
                                                mimetype: m.mimetype,
                                                caption: m.caption || "",
                                                directPath: m.directPath,
                                                encFilehash: m.encFilehash,
                                                filehash: m.filehash,
                                                mediaKey: m.mediaKey,
                                                mediaKeyTimestamp: m.mediaKeyTimestamp,
                                           }}))[0];
    """)
    
    return latest_msg

def load_send():
    driver.execute_script("""document.mediaInfoToFile = ({ data, mimetype, filename }) => {
                                const binaryData = window.atob(data);
                                const buffer = new ArrayBuffer(binaryData.length);
                                const view = new Uint8Array(buffer);
                                for (let i = 0; i < binaryData.length; i++) {
                                    view[i] = binaryData.charCodeAt(i);
                                }
                                const blob = new Blob([buffer], { type: mimetype });
                                return new File([blob], filename, {
                                    type: mimetype,
                                    lastModified: Date.now()
                                });
                             };
    """)

def media_send(ids, mediainfo, caption, as_attach):
    driver.execute_script(f"document.chat = window.Store.Chat.get('{ids}');")
    driver.execute_script("document.meUser = window.Store.User.getMaybeMePnUser();")
    driver.execute_script("document.newId = await window.Store.MsgKey.newId();")
    driver.execute_script("""document.newMsgId = new window.Store.MsgKey({
                                from: document.meUser,
                                to: document.chat.id,
                                id: document.newId,
                                participant: document.chat.id.isGroup() ? document.meUser : undefined,
                                selfDir: 'out',
                             });
    """)

    driver.execute_script("document.file = document.mediaInfoToFile(arguments[0])", mediainfo)

    # init file
    driver.execute_script(f"""document.mData = await window.Store.OpaqueData.createFromData(document.file, document.file.type);
                              document.mediaPrep = window.Store.MediaPrep.prepRawMedia(document.mData, {{ asDocument: {as_attach} }});
                              document.mediaData = await document.mediaPrep.waitForPrep();
                              document.mediaObject = window.Store.MediaObject.getOrCreateMediaObject(document.mediaData.filehash);
                              document.mediaType = window.Store.MediaTypes.msgToMediaType({{
                                  type: document.mediaData.type,
                                  isGif: document.mediaData.isGif
                              }});
    """)

    # init upload
    driver.execute_script("""if (!(document.mediaData.mediaBlob instanceof window.Store.OpaqueData)) {
                                 document.mediaData.mediaBlob = await window.Store.OpaqueData.createFromData(document.mediaData.mediaBlob, document.mediaData.mediaBlob.type);
                             }
                             document.mediaData.renderableUrl = document.mediaData.mediaBlob.url();
                             document.mediaObject.consolidate(document.mediaData.toJSON());
                             document.mediaData.mediaBlob.autorelease();
                             document.uploadedMedia = await window.Store.MediaUpload.uploadMedia({
                                mimetype: document.mediaData.mimetype,
                                mediaObject: document.mediaObject,
                                mediaType: document.mediaType
                             });
                             document.mediaEntry = document.uploadedMedia.mediaEntry;
                             if (!document.mediaEntry) {
                                throw new Error('upload failed: media entry was not created');
                             }
    """)

    # init send
    driver.execute_script("""document.mediaData.set({
                                clientUrl: document.mediaEntry.mmsUrl,
                                deprecatedMms3Url: document.mediaEntry.deprecatedMms3Url,
                                directPath: document.mediaEntry.directPath,
                                mediaKey: document.mediaEntry.mediaKey,
                                mediaKeyTimestamp: document.mediaEntry.mediaKeyTimestamp,
                                filehash: document.mediaObject.filehash,
                                encFilehash: document.mediaEntry.encFilehash,
                                uploadhash: document.mediaEntry.uploadHash,
                                size: document.mediaObject.size,
                                streamingSidecar: document.mediaEntry.sidecar,
                                firstFrameSidecar: document.mediaEntry.firstFrameSidecar
                             });

    """)

    # prepare message obj
    driver.execute_script(f"""document.message = {{
                                id: document.newMsgId,
                                ack: 0,
                                body: document.mediaData.preview,
                                from: document.meUser,
                                to: document.chat.id,
                                local: true,
                                self: 'out',
                                t: parseInt(new Date().getTime() / 1000),
                                isNewMsg: true,
                                ...document.mediaData,
                                caption: "{caption}"
                             }};
    """)

    driver.execute_script("window.Store.SendMessage.addAndSendMsgToChat(document.chat, document.message)")

def send_message(ids, response):
    driver.execute_script(f"document.chat = window.Store.Chat.get('{ids}');")
    driver.execute_script("document.meUser = window.Store.User.getMaybeMePnUser();")
    driver.execute_script("document.newId = await window.Store.MsgKey.newId();")
    driver.execute_script("""document.newMsgId = new window.Store.MsgKey({
                                from: document.meUser,
                                to: document.chat.id,
                                id: document.newId,
                                participant: document.chat.id.isGroup() ? document.meUser : undefined,
                                selfDir: 'out',
                             });
    """)

    driver.execute_script(f"""document.message = {{
                                id: document.newMsgId,
                                ack: 0,
                                body: "{response}",
                                from: document.meUser,
                                to: document.chat.id,
                                local: true,
                                self: 'out',
                                t: parseInt(new Date().getTime() / 1000),
                                isNewMsg: true,
                                type: 'chat',
                              }};
    """)

    driver.execute_script("window.Store.SendMessage.addAndSendMsgToChat(document.chat, document.message)")

def gather_msg(msgs):
    messages = []
    for msg in msgs:
        if msg == "No message history":
            messages.append(msg)
        elif msg["type"] == "chat":
            messages.append(emoji.demojize(msg["body"]))
        elif msg["type"] == "image" or msg["type"] == "sticker":
            # using mimetype for stickers
            messages.append([(msg["type"], msg["mimetype"], decrypt_media(msg), emoji.demojize(msg["caption"]))])
        elif msg["type"] == "revoked":
            messages.append("Message deleted")
        else:
            msg["caption"] = emoji.demojize(msg["caption"])
            messages.append(msg)
    return messages

def decrypt_media(msg):
    driver.execute_script(f"""try {{ 
                              document.mockQpl = {{
                                addAnnotations: function() {{ return this; }},
                                addPoint: function() {{ return this; }}
                              }};
                              document.decryptedMedia = await window.Store.DownloadManager.downloadAndMaybeDecrypt({{
                                directPath: "{msg["directPath"]}",
                                encFilehash: "{msg["encFilehash"]}",
                                filehash: "{msg["filehash"]}",
                                mediaKey: "{msg["mediaKey"]}",
                                mediaKeyTimestamp: "{msg["mediaKeyTimestamp"]}",
                                type: "{msg["type"]}",
                                signal: (new AbortController).signal,
                                downloadQpl: document.mockQpl
                              }})}} 
                              catch(e) {{ if(e.status && e.status == 404) document.decryptedMedia = undefined }};
    """)

    driver.execute_script("""document.base64str = (arrayBuffer) => new Promise((resolve, reject) => {
                                const blob = new Blob([arrayBuffer], {
                                    type: 'application/octet-stream',
                                });
                                const fileReader = new FileReader();
                                fileReader.onload = () => {
                                    const [, data] = fileReader.result.split(',');
                                    resolve(data);
                                };
                                fileReader.onerror = (e) => reject(e);
                                fileReader.readAsDataURL(blob);
                             });
    """)

    base64str = driver.execute_script("if(document.decryptedMedia != undefined) return await document.base64str(document.decryptedMedia)")

    return base64str

def your_name():
    name = driver.execute_script("return window.Store.Contact.get(window.Store.User.getMaybeMePnUser()._serialized).name")
    return name

def chats():
    try:
        WebDriverWait(driver, 60).until(EC.presence_of_element_located((By.CSS_SELECTOR, '[aria-label="Chats"]')))
    except:
        return {"Not visible": "Chats are not visible yet, try reloading"}
    latest_msg = []
    all_num = driver.execute_script("return window.Store.Chat.map(contacts => contacts.id._serialized)")
    contacts = driver.execute_script("return window.Store.Chat.map(contacts => contacts.formattedTitle);")
    for num in all_num:
        load_c = load_chat(num)
        if load_c is None:
            load_history(num)
            load_c = load_chat(num)
            driver.execute_script("window.Store.Cmd.closeActiveChat()")
            if load_c is None:
                latest_msg.append("No message history")
            else:
                latest_msg.append(load_c)
        else:
            latest_msg.append(load_c)

    all_l_msg = gather_msg(latest_msg)
               
    load_send()
    contact_msg = dict(zip(contacts, all_l_msg))
    return contact_msg