使用 Selenium 进行 Instagram 多处理

Instagram Multiprocessing with Selenium

嘿,相信你们都很好,我正在尝试 运行 多个 Instagram 帐户并行使用 MultiprocessingSelenium,我将如何使用创建的每个进程登录不同的帐户。

我尝试使用带有 while 循环的 json 文件,但是我没有取得太大进展。

如果能得到任何帮助,我将不胜感激。

import time
import json

from selenium import webdriver
from multiprocessing import Pool

f = open('accounts.json',)
datas = json.load(f)

def get_data(url):

    Options = webdriver.ChromeOptions()
    mobile_emulation = {
        "userAgent": "Mozilla/5.0 (Linux; Android 4.2.1; en-us; Nexus 5 Build/JOP40D) AppleWebKit/535.19 (KHTML, like Gecko) Chrome/101.0.4951.64 Mobile Safari/535.19" }
    Options.add_experimental_option("mobileEmulation", mobile_emulation)
    Options.add_argument("--log-level=3")

    bot = webdriver.Chrome(options=Options, executable_path="chromedriver.exe")
    bot.set_window_size(500, 768)
    bot.get(url=url)
    
    time.sleep(10)

    # Login section==========================
    print('Logging in...')
    bot.find_element_by_xpath(
        '//*[@id="react-root"]/section/main/article/div/div/div/div[3]/button[1]').click()
    time.sleep(5)
    username_field = bot.find_element_by_xpath(
        '//*[@id="loginForm"]/div[1]/div[3]/div/label/input')
    username_field.send_keys(data["username"])
    time.sleep(5)
    password_field = bot.find_element_by_xpath(
        '//*[@id="loginForm"]/div[1]/div[4]/div/label/input')
    password_field.send_keys(data["password"])
    time.sleep(5)
    bot.find_element_by_xpath(
        '//*[@id="loginForm"]/div[1]/div[6]/button').click()
    time.sleep(6)
    bot.close()
    bot.quit()


if __name__ == '__main__':
    process_count = int(input("Enter the number of processes: "))
    url = "https://www.instagram.com/"
    urls_list = [url] * process_count
    print(urls_list)
    p = Pool(processes=process_count)
    p.map(get_data, urls_list)
    while True:
        for data in datas:
            get_data()

此脚本使用threading(而不是multiprocessing)打开浏览器的多个独立windows(实例)。函数test_instance中包含的代码在每个window.

中同时是运行
import time
from selenium import webdriver
import threading
import json

def test_instance(data):
    Options = webdriver.ChromeOptions()
    mobile_emulation = {"userAgent": "Mozilla/5.0 (Linux; Android 4.2.1; en-us; Nexus 5 Build/JOP40D) AppleWebKit/535.19 (KHTML, like Gecko) Chrome/101.0.4951.64 Mobile Safari/535.19"}
    Options.add_experimental_option("mobileEmulation", mobile_emulation)
    Options.add_argument("--log-level=3")

    bot = webdriver.Chrome(options=Options, executable_path="chromedriver.exe")
    bot.set_window_size(500, 768)
    bot.get("https://www.instagram.com/")
    
    time.sleep(10)

    # Login section==========================
    print('Logging in...')
    bot.find_element_by_xpath('//*[@id="react-root"]/section/main/article/div/div/div/div[3]/button[1]').click()
    time.sleep(5)
    username_field = bot.find_element_by_xpath('//*[@id="loginForm"]/div[1]/div[3]/div/label/input')
    username_field.send_keys(data['username'])
    time.sleep(5)
    password_field = bot.find_element_by_xpath('//*[@id="loginForm"]/div[1]/div[4]/div/label/input')
    password_field.send_keys(data['password'])
    time.sleep(5)
    bot.find_element_by_xpath('//*[@id="loginForm"]/div[1]/div[6]/button').click()
    time.sleep(6)
    
    bot.quit()

f = open('accounts.json',)
data = json.load(f)
f.close()
process_count = 2 # number of tests to run (each test open a separate browser)
thread_list = []

# Start test
for i in range(process_count):
    t = threading.Thread(name=f'Test {i}', target=test_instance, args=[data[i]])
    t.start()
    time.sleep(1)
    print(t.name + ' started')
    thread_list.append(t)

# Wait for all threads to complete
for thread in thread_list:
    thread.join()

print('Test completed')