使用 Selenium 进行 Instagram 多处理
Instagram Multiprocessing with Selenium
嘿,相信你们都很好,我正在尝试 运行 多个 Instagram 帐户并行使用 Multiprocessing 和 Selenium,我将如何使用创建的每个进程登录不同的帐户。
我尝试使用带有 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')
嘿,相信你们都很好,我正在尝试 运行 多个 Instagram 帐户并行使用 Multiprocessing 和 Selenium,我将如何使用创建的每个进程登录不同的帐户。
我尝试使用带有 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')