一、背景/應(yīng)用介紹
- 因一個測試場景的要求,需要做個ip代理功能;并且要可以做到開啟多個窗口下還能指定IP(根據(jù)賬號指定IP不變);
-
在冥思苦想下,就開始動工啦,下圖是這樣的一個構(gòu)思:
需要實(shí)現(xiàn)的功能-思維導(dǎo)圖
二、直接先看最后的成果:
1、起了三個窗口,每個瀏覽器都是獨(dú)立的代理IP,且各瀏覽器的存儲的一些信息也是獨(dú)立的(如cookies)。確保了不會讓數(shù)據(jù)有污染情況,導(dǎo)致瀏覽器storage共用情況;
多窗口不同IP的結(jié)果
多窗口不同IP的結(jié)果2
三、介紹整個構(gòu)思設(shè)計(jì)過程
3.1、流程前置步驟:先取conf.txt配置文件信息(后續(xù)流程需要多處用到):
chrome_v= 85.0.4183.83 #填寫本地谷歌瀏覽器版本(比如:81.0.4044.138)并需要確認(rèn)官方是否有對應(yīng)的驅(qū)動版本.如無,可填近似版本;可進(jìn)入鏈接https://npm.taobao.org/mirrors/chromedriver查看官方驅(qū)動.
chrome_url=https://cdn.npm.taobao.org/dist/chromedriver #填寫谷歌瀏覽器驅(qū)動下載地址或填https://chromedriver.storage.googleapis.com/index.html
log_save=yes #是否要保存日志:可填寫yes或者no
file_name=userid_data.xlsx ##IP的excle文檔命名名稱,注意填全稱包含xlsx
testip1_url=http://httpbin.org/get?show_env=1 #request接口查詢,第一次核驗(yàn)代理IP正常性,如果不一致會被強(qiáng)制退出;一般不可更改
testip2_url=https://www.ip138.com/ #通過瀏覽器打開IP查詢頁面,第二次核驗(yàn)代理IP正常性(供頁面查看);一般不可更改
bocai_url=https://cn.xxxx.com/home/register #目標(biāo)網(wǎng)站鏈接
user_id=uid #用戶賬戶元素定位
password=jpwd #用戶密碼元素定位
captcha_text2=captcha_text2 #驗(yàn)證碼元素定位
time_sleep1=5 #testip1_ur2鏈接,打開后等待時(shí)間(秒)
time_sleep2=5 #bocai_url鏈接,打開后等待時(shí)間(秒)
def config_txt(file_name="config/conf.txt"):#讀取txt配置文件參數(shù)
data_head=data_tail=list()
for line in open(file_name,encoding='utf-8-sig', errors='ignore'):
head, sep,tail = line.partition('=')
data_head.append(head)
if tail.find('#')!=-1:
tail, tail_b,tail_c = tail.partition('#')
data_tail.append(tail.strip('\n '))
txt_data=dict(zip(data_head,data_tail))
return txt_data
3.2、流程第一個步驟:通過requests判斷引用代理ip是否有效、以及IP核驗(yàn)正確性
1)、步驟1-獲取對應(yīng)賬號下的指定IP相關(guān)的信息:因?yàn)槭菍脮r(shí)是需要多個代理IP,故用文檔統(tǒng)一管理(如下圖),根據(jù)賬號固定綁定一個;
備注:對于代理IP哪里來,其實(shí)網(wǎng)上可以搜到一些可以直接免費(fèi)用的IP,但不太穩(wěn)定;我這里是通過淘寶購買了,也賊便宜;

def excel(file_name,users_id):#讀取excel賬號及路由代理信息
try:
data_excle=pd.read_excel(io='config/'+file_name,sheet_name=0)
df = pd.DataFrame(data_excle)
user_id=int(users_id) if type(users_id)==type("") and users_id.isdecimal()==True else users_id
df_data=df[(df.user_id==user_id)].to_dict('list')
user_sum=df[(df.user_id==user_id)].shape[0]
if user_sum != 1:
print("excel文檔查找賬戶出現(xiàn)異常,該"+str(user_id)+"共查到"+str(user_sum)+"個")
input('點(diǎn)擊回車鍵可退出......')
exit()
keys = [key for key in df_data.keys()]
values=[str(value).strip("'[]") for value in df_data.values() ]
txt_data = dict(zip(keys, values))
print(txt_data)
return txt_data["user_id"],txt_data['password'],txt_data['ip'],txt_data['port'],txt_data['city'],\
txt_data['ip_user'],txt_data['ip_password'],txt_data['browser'],txt_data['test']
except :
print("錯誤提示:未在excel文檔內(nèi)找到 "+str(users_id)+" 用戶,請檢查是否已配置")
備注:此處加了個執(zhí)行記錄(相當(dāng)于log作用),把每次執(zhí)行的都保存下來;引用的是第三方Logger庫,需要可以度娘搜搜就有;
def testLogger(now_time,log_save):#log和生成的路徑邏輯
time_a = time.strftime("%Y-%m-%d", time.localtime())
time_log = 'log/'+time_a+"/"
isExists = os.path.exists(time_log)
if not isExists:
os.makedirs(time_log)
sys.stdout= Logger.Logger(time_log+now_time+'.txt', sys.stdout) if log_save=='yes' else None
return time_log
2)、步驟2-引用request庫調(diào)用httpbin.org核驗(yàn)IP有效性、正確性
def start_test(user_id):
global browser
print("本次登錄賬號:",str(user_id))
now_time,conf_txt=str(int(time.time()*1000)),config_txt()
time_log = testLogger(now_time,conf_txt['log_save'])
user_id,password,ip,port,city,ip_user,ip_password,browser,test=excel(conf_txt["file_name"],user_id)
#拼接樣例:'http://649192***:o2m0n***@101.200.187.22:16819'
reque_ip=ip_user+":"+ip_password+"@"+ip+":"+port
affirm=ip_affirm(reque_ip,conf_txt["testip1_url"])
#校驗(yàn)通過httpbin響應(yīng)的IP是否與文檔內(nèi)的一致
if ip == affirm.split(",")[0]:
print("ip校驗(yàn)成功"+ip)
else:
print("IP校驗(yàn)失敗\n配置ip為:"+ip,"查詢結(jié)果IP為:"+affirm)
input('點(diǎn)擊回車鍵可退出......')
exit()
def ip_affirm(proxies,testip1_url):#requests查詢實(shí)際的IP接口地址(代理)
#定義配置到的IP代理信息
proxies={ "http": "http://"+proxies }
try:
response_get=requests.get(url=testip1_url, proxies=proxies)
if response_get.status_code == 200:
print(response_get.text)
data=response_get.json()
return data["origin"]
except :
print ("#####Error: 查詢IP連接出現(xiàn)異常,請檢查IP代理賬戶密碼及鏈接是否正確有效")
input('點(diǎn)擊回車鍵可退出......')
exit()
3.3、打包瀏覽器代理插件:讓每個窗口獨(dú)立指定IP,另外在ui層面檢查一邊(引用selenium)
#谷歌瀏覽器的文件配置
capa=DesiredCapabilities.CHROME
capa["pageLoadStrategy"] = "none"
#第三方的proxyauth庫支持,填入代理信息即可
proxy_conf= proxyauth.create_proxy(host=ip,port=port,username=ip_user,password=ip_password)
co = webdriver.ChromeOptions()
co.add_argument("--start-maximized")
co.add_extension(proxy_conf)
browser = webdriver.Chrome(options=co,desired_capabilities=capa)
wait = WebDriverWait(browser,20)
#打開第三方網(wǎng)址,ui檢查IP地址
browser.get(conf_txt["testip2_url"])
time.sleep(int(conf_txt["time_sleep1"]))
# 進(jìn)入測試頁面,通過selenium自動填入基本信息
browser.get(conf_txt["bocai_url"]);
wait.until(EC.presence_of_element_located((By.XPATH, "http://*[@id='logincaptcha2']")))
time.sleep(int(conf_txt["time_sleep2"]))
browser.execute_script("window.stop();")
browser.find_element_by_id(conf_txt["user_id"]).send_keys(user_id)
browser.find_element_by_id(conf_txt["password"]).send_keys(password)
#這一段是借助度娘的大神們的
# 打包Google代理插件
def create_proxy(host, port, username, password,scheme='http', plugin_path=None):
if plugin_path is None:
# 插件地址
plugin_path = 'config/vimm_chrome_proxyauth_plugin.zip'
manifest_json = """
{
"version": "1.0.0",
"manifest_version": 2,
"name": "Chrome Proxy",
"permissions": [
"proxy",
"tabs",
"unlimitedStorage",
"storage",
"<all_urls>",
"webRequest",
"webRequestBlocking"
],
"background": {
"scripts": ["background.js"]
},
"minimum_chrome_version":"22.0.0"
}
"""
background_js = string.Template(
"""
var config = {
mode: "fixed_servers",
rules: {
singleProxy: {
scheme: "${scheme}",
host: "${host}",
port: parseInt(${port})
},
bypassList: ["foobar.com"]
}
};
chrome.proxy.settings.set({value: config, scope: "regular"}, function() {});
function callbackFn(details) {
return {
authCredentials: {
username: "${username}",
password: "${password}"
}
};
}
chrome.webRequest.onAuthRequired.addListener(
callbackFn,
{urls: ["<all_urls>"]},
['blocking']
);
"""
).substitute(host=host,port=port,username=username,password=password,scheme=scheme,)
with zipfile.ZipFile(plugin_path, 'w') as zp:
zp.writestr("manifest.json", manifest_json)
zp.writestr("background.js", background_js)
return plugin_path
3.4、為方便win執(zhí)行,觸發(fā)命令轉(zhuǎn)換成bat文件(另外也可以用Bat To Exe Converter轉(zhuǎn)成exe就更方便一點(diǎn)):
#另外此處是獲取了文件名,然后給python傳個賬戶實(shí)參
@echo off
set name=%~n0
start python model/test_bocai.py %name%
exit
*生成exe文件后,并命名成對應(yīng)賬戶名,然后雙擊即可使用啦~~~~

3.5、自動檢測及修復(fù):python依賴包的以及谷歌驅(qū)動下載
1)、步驟1:對于python依賴包,相對用了最簡單去處理了。通過import判斷是否安裝對應(yīng)庫;如沒有就pip所需庫的下載(我這里是引用了阿里云鏡像源,相對比較穩(wěn)定比較快):
# -*-coding:utf-8 -*-
# python各庫自動檢測及安裝
import os
from test_item import Classdriver
aliyun=' -i http://mirrors.aliyun.com/pypi/simple --trusted-host mirrors.aliyun.com'
print('正常進(jìn)行自動檢測、自動安裝中,請勿退出...')
print("本系統(tǒng)python版本及安裝目錄:")
os.system('python -V')
os.system('where python')
try:
import pandas as pd
except:
print('未找到<pands>庫,正自動安裝中:')
os.system('pip install pandas'+aliyun)
print("pandas庫版本:")
os.system('pip list | findstr pandas')
try:
import selenium
except:
print('未找到<selenium>庫,正自動安裝中:')
os.system('pip install selenium'+aliyun)
print("selenium庫版本:")
os.system('pip list | findstr selenium')
try:
import requests
except:
print('未找到<requests>庫,正自動安裝中:')
os.system('pip install requests'+aliyun)
print("requests庫版本:")
os.system('pip list | findstr requests')
try:
from aip import AipOcr
except:
print('未找到<baidu-aip>庫,正自動安裝中:')
os.system('pip install baidu-aip '+aliyun)
print("baidu-api版本:")
os.system('pip list | findstr baidu-aip')
print("將自動打開谷歌瀏覽器,進(jìn)行檢測驅(qū)動版本")
Classdriver().driver()
input('自動檢測修復(fù)已完畢,請點(diǎn)擊回車鍵可退出......')
2)、步驟2--谷歌驅(qū)動這塊的邏輯(對應(yīng)功能點(diǎn)3、4),實(shí)現(xiàn)自動從官方下載對應(yīng)的驅(qū)動,并配置到python路徑中
也可通過鏈接,手動下載:https://npm.taobao.org/mirrors/chromedriver
import os
import zipfile
import requests
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
class Classdriver(object):
def __init__(self):
self.python_path_a=list(os.popen('where python'))
self.python_path=str(self.python_path_a[0].strip("\n"))
def pyhon_model(self,chrome_v,driver="chromedriver_win32.zip",url='https://cdn.npm.taobao.org/dist/chromedriver'):
# 拼接下載鏈接,鏈接2:https://chromedriver.storage.googleapis.com/index.html'
if url.find("taobao")!= -1:
download_url = url+'/'+chrome_v+'/'+driver
else:
download_url = url + '?path='+chrome_v+'/'+driver
file_a = requests.get(download_url)
if file_a.status_code==200:
file=file_a
else:
print('###驅(qū)動下載出現(xiàn)異常:請檢查<chrome_v>版本號是否有對應(yīng)的官方驅(qū)動版本,如未匹配可填寫最接近的驅(qū)動版本號...')
print('本次填寫的<chrome_v>版本號:',chrome_v)
print('請進(jìn)入官方驅(qū)動版本鏈接檢查是否有此版本號:https://npm.taobao.org/mirrors/chromedriver')
input('點(diǎn)擊回車鍵可退出......')
exit()
unzip=self.python_path[:-10]+driver
with open(unzip, 'wb') as zip_file: # 保存文件到腳本所在目錄
zip_file.write(file.content)
print("unzip",unzip)
zip_file = zipfile.ZipFile(unzip)
zip_list = zip_file.namelist() # 得到壓縮包里所有文件
for i in range(len(self.python_path_a)):
for f in zip_list:
zip_file.extract(f, self.python_path_a[i].strip("\n")[:-10]) # 循環(huán)解壓文件到指定目錄
zip_file.close()
def driver(self):
chrome_options = Options()
chrome_options.add_argument('--no-sandbox')
chrome_options.add_argument('--disable-dev-shm-usage')
chrome_options.add_argument('--headless')
chrome_options.add_experimental_option('excludeSwitches', ['enable-logging'])
#讀取config配置內(nèi)的配置信息
config=self.config_txt()
if os.path.isfile(self.python_path[:-10]+'chromedriver.exe'):
browser = webdriver.Chrome(options=chrome_options)
rr=browser.capabilities['chrome']['chromedriverVersion']
head, sep,tail = rr.partition(' ')
browser.quit()
#核驗(yàn)配置內(nèi)的谷歌版本與當(dāng)前驅(qū)動版本一致性,不一致將自動下載官方驅(qū)動
if head==config['chrome_v']:
print('谷歌驅(qū)動已安裝、版本正確=====版本號:'+head)
else:
self.pyhon_model(chrome_v=config['chrome_v'],url=config['chrome_url'])
else:
self.pyhon_model(chrome_v=config['chrome_v'],url=config['chrome_url'])
def config_txt(self,file_name="conf.txt"):#讀取txt配置文件參數(shù)
data_head=data_tail=list()
for line in open(file_name,encoding='utf-8-sig', errors='ignore'):
head, sep,tail = line.partition('=')
data_head.append(head)
if tail.find('#')!=-1:
tail, tail_b,tail_c = tail.partition('#')
data_tail.append(tail.strip('\n '))
txt_data=dict(zip(data_head,data_tail))
return txt_data
四、收尾語(主要寫累了,就草草收尾了??):
- 其中漏了一個圖片驗(yàn)證碼ocr的識別;這個屬于其他也業(yè)務(wù)場景了,就未另外補(bǔ)充了(或可見我另一篇文章);
- 然后,整體就需求功能就算完了,按自己的初學(xué)者的思路去做了一個小小項(xiàng)目;具體應(yīng)該還是有很多可以優(yōu)化的,繼續(xù)努力把,一步一個腳?。?/li>


