Instagram Multiprocessing with Selenium

Question

大家都相信你很好，我正在尝试使用Multiprocessing并行运行多个Instagram帐户硒和如果创建了每个进程，我该如何登录不同的帐户呢?

我尝试使用json文件与while循环，但我没有取得太大进展。

我将感激任何我能得到的帮助。

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()

Answer 1

此脚本使用threading(而不是multiprocessing)打开多个独立的浏览器窗口(实例)。函数test_instance中包含的代码在每个窗口中同时运行。

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')