標籤：

python 爬蟲 ip池怎麼做？

12-28

其實提這個問題是想知道知乎問題ID到多少了。。沒想到這麼久還有人回答。。對於爬蟲我覺得應使用requests庫，挺強大的。尤其是爬內網的時候遇到NTLM驗證。。代理池最主要的就是抓取ip，去重，驗證，丟棄。應該用queue，先進先出，率先丟棄時間長的ip概率也比較大。我覺得池最重要的就是活性。。純屬吹牛逼。。

自己曾經搭建過一個通用化的爬蟲平台，這個通用化爬蟲平台主要包含6大服務：
1、規則配置服務；
2、爬蟲採集服務；
3、代理採集服務；
4、動態頁面渲染服務（渲染JS載入的頁面）；
5、數據清洗服務；
6、數據推送和輸出服務；
每個服務工作純粹，各司其職，方便開發和維護。

這次主要說一下代理服務（也就是你說的代理IP池）是如何實現的：
爬蟲代理服務主要分為付費代理和免費代理：
1、付費代理：
現在有很多賣代理的網站，這些網站的代理IP來源主要是：
a.網上採集免費代理
b.自己買雲伺服器自己搭建
對於第一種方式，我們自己也可以無成本做到，下面說如何完成。
第二種方式，則需要花費一定成本，當然，這種類型的代理相對第一種比較穩定，如果是要購買的話，建議購買這種代理。

2、免費代理：
這種方式就是自己寫個程序，到網上採集免費的代理IP下來，供自己使用。
採集的過程無非是：程序訪問代理頁面 --&> 正則/XPATH匹配出IP --&> 保存到本地

理清楚了思路之後，程序當然很好寫出來。但是在這個過程中，會遇到一些問題。
例如：
在哪裡尋找代理網站？
採集到的代理質量如何？能用多久？
如果代理都失效，如何持續不斷採集代理？
代理是否可做成服務化，讓代理變得更通用？

下面一一解決這些問題：
a.在哪裡尋找代理網站？
很簡單，作為程序員，最基本的技能就是使用搜索引擎，輸入關鍵字 免費代理IP，就會看到很多的代理IP網站
打開看一下，每個網站代理IP少則幾十個，多則幾百個。
OK，寫個代理抓取程序，把這些網站的代理抓下來。

b.採集到的代理質量如何？能用多久？
這麼多IP，難道別人真的就免費送給你了么？當然不是，這些代理中，有很大一部分已經失效了。
如何測試這些代理是否可用？
寫一個代理測試程序，掛上這些代理，訪問某一個穩定的網站，看是否可以正常訪問。
可以正常訪問則這個代理就是可用的，不能訪問則認為是失效的。
當然，這裡用多線程或非同步的方式更高效。
通過這個代理測試程序，你就能獲取到最近可用的代理列表了。

c.如果代理都失效，如何持續不斷採集代理？
通過上面的 代理採集程序 和 代理測試程序兩個配合，定時啟動採集程序，然後把這些代理寫到資料庫中，
同時，定時啟動代理測試程序，代理可用就打上可用標記，在這個測試程序中，你還可以記錄下測試訪問的響應時間，存入資料庫。
如果代理網站上有顯示代理的協議、代理類型(透明、普通匿名、高匿名)、代理位置、運營商，那就一併存入庫中，
這樣你後期篩選代理，不就多了更多維度嗎？（按測試時間排序查詢、按類型查詢、按運營商查詢等等）
只要這2個程序定時啟動間隔控制好，就能持續輸出可用代理。

d.代理是否可做成服務化，讓代理變得更通用？
現在有了這麼多代理，你應該知道如何使用吧？爬蟲掛上代理，採集頁面即可。
但是，有一種更好的方式使用代理，就是把代理做成服務化。
使用squid的cache_peer機制，把這些代理按照一定格式(具體格式參考文檔)寫入到配置文件中，
配置好squid的埠，那麼squid就可以幫你調度代理了，而且還可以摒棄失效的代理。
這麼做的好處是，你的爬蟲程序只需要指定一個代理：squid的埠就可以了！

詳細可以參考我的博客：爬蟲代理服務

知乎確實對單個ip的訪問限制挺嚴格的，但是對於http訪問來說，並不一定需要抓取別人做好的代理。國外的GAE，AWS，以及各種免費的虛擬主機，用python，php都有現成的代理服務，寫個自動化腳本不停地去配置、刪除代理服器就可以了。

要是僅僅短期使用的話其實用不著這麼麻煩，在訪問的http request里添加x-forward-for標籤，client隨機生成，宣稱自己是一台透明代理伺服器，代理其他人的訪問就能繞過許多限制了，一般服務商不會限制透明代理。

看了那麼多知乎，也來貢獻一下自己的力量吧。
按照 @路人甲 的思路繼續下去，比起路人甲兄的做法有一下幾個新的feature：

多線程代理池：
1. 新增西刺代理(xicidaili)+快代理(kuaidaili)兩個來源的上千個免費代理
2. 保證代理的時效性和質量，增加國內外代理的有效性認證(validation)
3. 認證過程採用多線程操作
4. 為了方便配置，不使用資料庫，改成文件操作，認證成功的保存在本地的txt文檔中。


# coding=utf-8
import requests
import time
from bs4 import BeautifulSoup
from selenium import webdriver
from threading import Thread
from Queue import Queue

_BE_PROXY_QUEUE = Queue()
path = "mt_proxy.txt"

class Consumer_Thread(Thread):
def run(self):
global _BE_PROXY_QUEUE
while not _BE_PROXY_QUEUE.empty():
p = _BE_PROXY_QUEUE.get()
try:
if test_useful(p):
with open(path, "a") as f:
f.write(p + "
")
except Exception, e:
print "[HERE]", e
pass
finally:
_BE_PROXY_QUEUE.task_done()

def test_useful(proxy):
print "[INFO] Testing proxy ", proxy, " now..."
try:
proxies = {"http": proxy}
requests.get("http://ip.cip.cc", timeout=20, proxies=proxies)
print "[Congra] Successfully get one"
return True
except Exception, e:
print e
return False

def get_proxies_from_KDL(max_page):
print "[Scrapy] Start Scrapying Proxies in KDL"

base_url = " http://www.kuaidaili.com/"
options = ["intr/", "inha/"]

p_pool = []

print "===============
", "Scraping XICI DAILI" "
===============
"
xici_page = 1

while xici_page &<= 3: # xicidaili new_count = 0 print "PAGE", str(xici_page) xici_url = " http://www.xicidaili.com/nt/" + str(xici_page) headers = { "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/46.0.2490.86 Safari/537.36"} try: rx = requests.get(xici_url, timeout=15, headers=headers) bobj_2 = BeautifulSoup(rx.content.decode("utf-8")) sibs = bobj_2.findAll("table", {"id": "ip_list"})[0].tr.next_siblings except Exception, e: try: print "error 1:", e rx = requests.get(xici_url, timeout=15, headers=headers) bobj_2 = BeautifulSoup(rx.content.decode("utf-8")) sibs = bobj_2.findAll("table", {"id": "ip_list"})[0].tr.next_siblings except Exception, e: print "error 2", e break for sib in sibs: try: get_proxy = sib.findAll("td")[2].get_text() + ":" + sib.findAll("td")[3].get_text() p_pool.append(get_proxy) new_count += 1 except AttributeError: pass print "get ", new_count, " proxies in page", xici_page xici_page += 1 # 第幾個分頁面 opt = 0 while opt &<= 1: page = 1 print "=============== ", "Scraping ", options[opt], " =============== " while page &< max_page: url = base_url + options[opt] + str(page) + "/" driver = webdriver.PhantomJS( executable_path=r"/root/csdn_spider/phantomjs-2.1.1-linux-x86_64/bin/phantomjs") #driver = webdriver.PhantomJS(executable_path=r"D:phantomjs-2.1.1-windowsinphantomjs.exe") driver.get(url) print "Sleep 0.7 sec..." time.sleep(0.7) bobj = BeautifulSoup(driver.page_source) driver.close() siblings = bobj.findAll(name="table", attrs={"class": "table table-bordered table-striped"})[ 0].tbody.tr.next_siblings count = 0 for sibling in siblings: try: get_proxy = sibling.findAll(name="td")[0].get_text() + ":" + sibling.findAll(name="td")[ 1].get_text() p_pool.append(get_proxy) count += 1 except AttributeError: pass print "Get", str(count), "proxy" page += 1 opt += 1 print "*****************************" print "Finished! Get ", len(p_pool), " useful proxies in total" return p_pool # with open("proxy_kdl.txt", "w") as f: # for p in p_pool: # p = p + " " # f.write(p) # print "Successfully written in "proxy_kdl.txt"" def get_proxies_from_file(): with open("proxy_kdl.txt", "r") as f: return f.readlines() def test_proxies_efficience(proxy): proxies = {"http": proxy} start = time.time() for i in range(3): r = requests.get("http://www.baidu.com", proxies=proxies) print i, " ", r.text cost = time.time() - start print "With Proxy: cost ", cost / 3, " seconds" start = time.time() for i in range(3): r = requests.get("http://ip.cip.cc") print i, " ", r.text cost = time.time() - start print "Without Proxy: cost ", cost / 3, " seconds" def main(): # 清空已有的文件 with open(path, "w") as f: f.write(time.ctime() + " ") f.write("======================== ") global _BE_PROXY_QUEUE max_thread = 100 threads = [] # 2大頁面，每個大頁面3個分頁 pool = get_proxies_from_KDL(3) print "uher2" for i in range(len(pool)): _BE_PROXY_QUEUE.put(pool[i]) for i in range(max_thread): threads.append(Consumer_Thread()) for i in range(max_thread): threads[i].start() # 陷入等待 線程不夠 是因為線程沒有死循環就退出 _BE_PROXY_QUEUE.join() print "###########################################" print "SUCCESS!" print "###########################################" if __name__ == "__main__": main()

代碼開源在 iFe1er/multithread-proxy · GitHub ，需要使用的同學直接git clone下來用就好了，記得要下載對應版本的PhantomJS，並在第90行處修改一下可執行文件的路徑！

----------

乾死黃旭東，Python能對空！lol

在做的項目，免費爬蟲代理IP池 知乎專欄


1、問題

  • 代理IP從何而來？

　　剛自學爬蟲的時候沒有代理IP就去西刺、快代理之類有免費代理的網站去爬，還是有個別代理能用。當然，如果你有更好的代理介面也可以自己接入。
　　免費代理的採集也很簡單，無非就是：訪問頁面頁面 —&> 正則/xpath提取 —&> 保存

  • 如何保證代理質量？

　　可以肯定免費的代理IP大部分都是不能用的，不然別人為什麼還提供付費的(不過事實是很多代理商的付費IP也不穩定，也有很多是不能用)。所以採集回來的代理IP不能直接使用，可以寫檢測程序不斷的去用這些代理訪問一個穩定的網站，看是否可以正常使用。這個過程可以使用多線程或非同步的方式，因為檢測代理是個很慢的過程。

  • 採集回來的代理如何存儲？

　　這裡不得不推薦一個高性能支持多種數據結構的NoSQL資料庫SSDB，用於代理Redis。支持隊列、hash、set、k-v對，支持T級別數據。是做分散式爬蟲很好中間存儲工具。

  • 如何讓爬蟲更簡單的使用這些代理？

　　答案肯定是做成服務咯，python有這麼多的web框架，隨便拿一個來寫個api供爬蟲調用。這樣有很多好處，比如：當爬蟲發現代理不能使用可以主動通過api去delete代理IP，當爬蟲發現代理池IP不夠用時可以主動去refresh代理池。這樣比檢測程序更加靠譜。


2、代理池設計

　　代理池由四部分組成:

  • ProxyGetter:

　　代理獲取介面，目前有5個免費代理源，每調用一次就會抓取這個5個網站的最新代理放入DB，可自行添加額外的代理獲取介面；

  • DB:

　　用於存放代理IP，現在暫時只支持SSDB。至於為什麼選擇SSDB，大家可以參考這篇文章,個人覺得SSDB是個不錯的Redis替代方案，如果你沒有用過SSDB，安裝起來也很簡單，可以參考這裡；

  • Schedule:

　　計劃任務用戶定時去檢測DB中的代理可用性，刪除不可用的代理。同時也會主動通過ProxyGetter去獲取最新代理放入DB；

  • ProxyApi:

　　代理池的外部介面，由於現在這麼代理池功能比較簡單，花兩個小時看了下Flask，愉快的決定用Flask搞定。功能是給爬蟲提供get/delete/refresh等介面，方便爬蟲直接使用。

設計

3、代碼模塊

　　Python中高層次的數據結構,動態類型和動態綁定,使得它非常適合於快速應用開發,也適合於作為膠水語言連接已有的軟體部件。用Python來搞這個代理IP池也很簡單，代碼分為6個模塊：

  • Api:
    　　api介面相關代碼，目前api是由Flask實現，代碼也非常簡單。客戶端請求傳給Flask，Flask調用ProxyManager中的實現，包括get/delete/refresh/get_all；

  • DB:
    　　資料庫相關代碼，目前資料庫是採用SSDB。代碼用工廠模式實現，方便日後擴展其他類型資料庫；

  • Manager:
    　　get/delete/refresh/get_all等介面的具體實現類，目前代理池只負責管理proxy，日後可能會有更多功能，比如代理和爬蟲的綁定，代理和賬號的綁定等等；

  • ProxyGetter:
    　　代理獲取的相關代碼，目前抓取了快代理、代理66、有代理、西刺代理、guobanjia這個五個網站的免費代理，經測試這個5個網站每天更新的可用代理只有六七十個，當然也支持自己擴展代理介面；

  • Schedule:
    　　定時任務相關代碼，現在只是實現定時去刷新代碼，並驗證可用代理，採用多進程方式；

  • Util:
    　　存放一些公共的模塊方法或函數，包含GetConfig:讀取配置文件config.ini的類，ConfigParse: 集成重寫ConfigParser的類，使其對大小寫敏感， Singleton:實現單例，LazyProperty:實現類屬性惰性計算。等等；

  • 其他文件:
    　　配置文件:Config.ini,資料庫配置和代理獲取介面配置，可以在GetFreeProxy中添加新的代理獲取方法，並在Config.ini中註冊即可使用；

4、安裝

下載代碼:

git clone git@github.com:jhao104/proxy_pool.git

或者直接到https://github.com/jhao104/proxy_pool 下載zip文件

安裝依賴:

pip install -r requirements.txt

啟動:

需要分別啟動定時任務和api
到Config.ini中配置你的SSDB

到Schedule目錄下:
&>&>&>python ProxyRefreshSchedule.py

到Api目錄下:
&>&>&>python ProxyApi.py

5、使用

　　定時任務啟動後，會通過代理獲取方法fetch所有代理放入資料庫並驗證。此後默認每20分鐘會重複執行一次。定時任務啟動大概一兩分鐘後，便可在SSDB中看到刷新出來的可用的代理：

useful_proxy

　　啟動ProxyApi.py後即可在瀏覽器中使用介面獲取代理，一下是瀏覽器中的截圖:

　　index頁面:

index

　　get：

get

　　get_all：

get_all

　　爬蟲中使用，如果要在爬蟲代碼中使用的話， 可以將此api封裝成函數直接使用，例如:

import requests

def get_proxy():
return requests.get("http://127.0.0.1:5000/get/").content

def delete_proxy(proxy):
requests.get("http://127.0.0.1:5000/delete/?proxy={}".format(proxy))

# your spider code

def spider():
# ....
requests.get("https://www.example.com", proxies={"http": "http://{}".format(get_proxy)})
# ....

6、最後

　　時間倉促，功能和代碼都比較簡陋，以後有時間再改進。喜歡的在github上給個star。感謝！

日常使用python爬蟲多的，會自己有一個代理ip的資料庫。

平時一直用自己寫的python腳本來抓取更新代理地址。
代碼就不貼了，有興趣前行到 http://30daydo.com/article/94
代碼裡頭有注釋幫助，易懂。

順便統計了一下 http://www.xicidaili.com 上面顯示出來的代理的可用率，最新最前面的可用率有76%，到時後面20個之後降到了30%。

########## 2016-08-13 更新 還是貼上代碼吧（個人網站代碼會定期更新，知乎的估計第一次怎樣以後就怎樣的了）################

# -*- coding=utf-8 -*-
__author__ = "Rocky"
import urllib2, time, datetime
from lxml import etree
import sqlite3,time

class getProxy():

def __init__(self):
self.user_agent = "Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; Trident/5.0)"
self.header = {"User-Agent": self.user_agent}
self.dbname="proxy.db"
self.now = time.strftime("%Y-%m-%d")

def getContent(self, num):
nn_url = "http://www.xicidaili.com/nn/" + str(num)
#國內高匿
req = urllib2.Request(nn_url, headers=self.header)
resp = urllib2.urlopen(req, timeout=10)
content = resp.read()
et = etree.HTML(content)
result_even = et.xpath("//tr[@class=""]")
result_odd = et.xpath("//tr[@class="odd"]")
#因為網頁源碼中class 分開了奇偶兩個class，所以使用lxml最方便的方式就是分開獲取。
#剛開始我使用一個方式獲取，因而出現很多不對稱的情況，估計是網站會經常修改源碼，怕被其他爬蟲的抓到
#使用上面的方法可以不管網頁怎麼改，都可以抓到ip 和port
for i in result_even:
t1 = i.xpath("./td/text()")[:2]
print "IP:%s Port:%s" % (t1[0], t1[1])
if self.isAlive(t1[0], t1[1]):

self.insert_db(self.now,t1[0],t1[1])
for i in result_odd:
t2 = i.xpath("./td/text()")[:2]
print "IP:%s Port:%s" % (t2[0], t2[1])
if self.isAlive(t2[0], t2[1]):
self.insert_db(self.now,t2[0],t2[1])

def insert_db(self,date,ip,port):
dbname=self.dbname
try:
conn=sqlite3.connect(dbname)
except:
print "Error to open database%" %self.dbname
create_tb="""
CREATE TABLE IF NOT EXISTS PROXY
(DATE TEXT,
IP TEXT,
PORT TEXT
);
"""
conn.execute(create_tb)
insert_db_cmd="""
INSERT INTO PROXY (DATE,IP,PORT) VALUES ("%s","%s","%s");
""" %(date,ip,port)
conn.execute(insert_db_cmd)
conn.commit()
conn.close()

def loop(self,page):
for i in range(1,page):
self.getContent(i)

#查看爬到的代理IP是否還能用
def isAlive(self,ip,port):
proxy={"http":ip+":"+port}
print proxy

#使用這個方式是全局方法。
proxy_support=urllib2.ProxyHandler(proxy)
opener=urllib2.build_opener(proxy_support)
urllib2.install_opener(opener)
#使用代理訪問騰訊官網，進行驗證代理是否有效
test_url="http://www.qq.com"
req=urllib2.Request(test_url,headers=self.header)
try:
#timeout 設置為10，如果你不能忍受你的代理延時超過10，就修改timeout的數字
resp=urllib2.urlopen(req,timeout=10)

if resp.code==200:
print "work"
return True
else:
print "not work"
return False
except :
print "Not work"
return False

if __name__ == "__main__":
now = datetime.datetime.now()
print "Start at %s" % now
obj=getProxy()
obj.loop(5)

懶得抓代理，直接段號重連，每次都是新的ip

如果ip需求量不是很大，請使用代理網站，如果使用量很大，有幾種方法，1、放慢你的爬取速度，即sleep時間長一點，2、使用 tor，3、撥號上網使用電信局的ip池，4、循環使用池中ip。
===============================================
現在我對我自己的回答重新編輯一下，之前我回答的僅僅是考慮到如何應對的問題，我想說的是，所有這些都是誤導，雖然我沒有寫出代碼，但是經過驗證之後，一般平台是不會對ip進行特殊處理的，你要知道很多地方的小區甚至地區，都是一個ip地址，而且大家都知道撥號的ip池，如果封殺ip的話會誤傷多大一篇，如果不是特別明顯的惡意腳本，一般會視而不見。
所以真正在說各類ip的，一般是僅僅了解繞開的路而已，試試上ip那又那麼容易封啊。難不成還要專門維護個大的ip池用於對黑名單的訪問，然後再考慮白名單。
當然我都是意淫。。。。。。
逃：）

實在看不下去了，這個問題里的代碼簡直都是垃圾，我來回答一個。

要解決這個問題，當然只有兩個辦法，一個是去買代理，一個是抓一些代理，方法一不說了，買好了自己配置好，用就可以了，我們來思考一下怎樣抓代理。

抓取的代理有一個問題就是的一般代理過一段時間就失效了，所以需要不斷判斷是不是好使的，然後不斷把失效的代理從代理池裡踢出去同時去抓一些新的代理來用。如果我們用一個 ProxyPool 類的話，至少得有兩個方法： refresh刷新代理池，get_proxy獲取一個代理。代碼如下

class Proxy:
pass

class ProxyPool:

def __init__(self):
self.pool = []

def refresh(self):
"先把原來的代理都過濾一遍，如果活著的代理不多了就獲取一些新的"""
self.pool = filter(lambda p: self._is_alive(p), self.pool)
if len(self.pool) &< 10: new_proxies = find_proxies() self.pool.extend(new_proxies) def _is_alive(self): """判定代理是不是活著的，簡單點用代理 urlopen 一下百度就知道了""" # return whether the proxy is alive def get_proxy(self): """從池子里隨機返回一個代理""" return random.choice(self.pool)

有了代理池的實現之後，我們發現還缺一個 find_proxies 的方法，這兒的問題是免費代理的網站那麼多，但是一個網站可能每天也就更新幾百個代理根本不夠用怎麼辦，所以得需要抓不同的網站。這時候我們定義一個介面吧

class IProxyFinder(ABC):

@abstractmethod
def find(self):
"""具體邏輯實現在這裡，返回一個代理的列表"""

class XXXProxyFinder(IProxyFinder):

def find(self):
# 如上

class ProxyPool:

def __init__(self, proxy_finder):
self.pool = []
self.proxy_finder = proxy_finder

def refresh(self):
# 這裡就可以改成使用 proxy_finder了

進一步的思考，其實抓代理需要啥，就三個參數，ip，埠，協議，只要能把每個提供代理的頁面的抓取規則整理出來，可以做一個 ProxyFinderFactory 。
還有，應該為每個代理設計一個 ttl，這樣每次 refresh 的時候就不要都刷新了，只刷新 ttl 到0的代理就好了。
進一步，還應該實現一個存儲層，把爬下來的代理放存儲到硬碟或者資料庫中，供其他程序使用。

歡迎關注微信公眾號 爬蟲技術學習 spider-learn

抓代理。

那些建議題主買代理ip的人真的買過么？我買過三款，都爛的要死，http協議的還算勉強能用，https的簡直不能忍，比較靠譜的方法是買動態vps，有那種20省40市的，重新撥號就更換ip了

買吧，免費的都非常不穩定
剛找到之前寫的一個抓xici代理的代碼，可以參考下，沒完全寫完，懶得改了，主要是因為可用的ip比例很低，意義不大。

#僅僅寫了mysql的,使用了pymysql。請確保自己的資料庫中無名為proxy的資料庫，否則會出錯。
import pymysql
import requests
from bs4 import BeautifulSoup
import time
import re

def dbstatus(user,pwd):#檢測資料庫proxy狀態，無則創建。
conn=pymysql.connect(host="127.0.0.1",user=user,passwd=pwd,charset="utf8")
cur=conn.cursor()
conn.select_db("information_schema")
dbstat=cur.execute("SELECT * from SCHEMATA where SCHEMA_NAME="proxy"")
if dbstat==0:
print("資料庫不存在，正在創建資料庫...")
cur.execute("CREATE DATABASE proxy DEFAULT CHARACTER SET utf8 COLLATE utf8_general_ci")#添加proxy資料庫，默認utf8
conn.select_db("proxy")
print("創建temp表...")
cur.execute("""
CREATE TABLE `temp` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`ip` varchar(255) COLLATE utf8_bin NOT NULL,
`port` varchar(255) COLLATE utf8_bin NOT NULL,
`place` varchar(255) COLLATE utf8_bin,
`iptype` varchar(255) COLLATE utf8_bin NOT NULL,
`time` varchar(255) COLLATE utf8_bin NOT NULL,
PRIMARY KEY (`id`)
)ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin AUTO_INCREMENT=1 ;
""")
print("創建temp表成功")
print("創建iptable表...")
cur.execute("""
CREATE TABLE `iptable` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`ip` varchar(255) COLLATE utf8_bin NOT NULL,
`port` varchar(255) COLLATE utf8_bin NOT NULL,
`place` varchar(255) COLLATE utf8_bin,
`iptype` varchar(255) COLLATE utf8_bin NOT NULL,
`time` varchar(255) COLLATE utf8_bin NOT NULL,
PRIMARY KEY (`id`)
)ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin AUTO_INCREMENT=1 ;
""")
print("創建iptable表成功")
print("創建完成")
cur.close()

def xicidaili():#將西刺代理的數據入庫
nows=time.time()
home="http://www.xicidaili.com"
guonei="http://www.xicidaili.com/nn/"
guowai="http://www.xicidaili.com/wn/"
def getiplist(url):
count=0
sql="INSERT INTO temp(ip,port,place,iptype,time) VALUES (%s,%s,%s,%s,%s)"
headers_base = {
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
"Accept-Encoding": "gzip, deflate, sdch",
"Accept-Language": "en-US,en;q=0.8,zh-CN;q=0.6,zh;q=0.4,zh-TW;q=0.2",
"Connection": "keep-alive",
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/43.0.2357.130 Safari/537.36",
}
conn=pymysql.connect(host="127.0.0.1",user="root",passwd="root",db="proxy",charset="utf8")
cur=conn.cursor()
s = requests.session()
res = s.get(url, headers=headers_base)
ipsoup=BeautifulSoup(res.content.decode("utf8"), "html.parser")
nextpagetgae=ipsoup.find_all(attrs={"class":"next_page"})
while nextpagetgae!=[]:
iptable=ipsoup.find(attrs={"id":"ip_list"})
iplist=iptable.find_all("tr")
for i in iplist[1:]:
ipinfo=i.find_all(name="td")
ip=ipinfo[1].string
port=ipinfo[2].string
place=ipinfo[3].string
iptype=ipinfo[5].string
updatetime="20"+ipinfo[9].string
t=(int(updatetime[:4]),int(updatetime[5:7]),int(updatetime[8:10]),int(updatetime[11:13]),int(updatetime[14:16]),0,0,0,0)
updatetimes=time.mktime(t)
if nows+28800-updatetimes&>86400*1:#如果更新時間與現時間差超過X天則丟棄,28800是為了糾正東八區的8小時時差,86400是一天
conn.commit()
cur.close()
print("導入%d條數據"%count)
return
if place==None:
place=ipinfo[3].a.string
cur.execute(sql,[ip,port,place.strip(),iptype,updatetime])
count+=1
conn.commit()
if nextpagetgae[0].get("href")==None:
break
nextpage=home+nextpagetgae[0].get("href")
ipsoup=BeautifulSoup(s.get(nextpage,headers=headers_base).content.decode("utf8"), "html.parser")
nextpagetgae=ipsoup.find_all(attrs={"class":"next_page"})
cur.close()
print("導入%d條數據"%count)
getiplist(guonei)
getiplist(guowai)

dbstatus("root","root")
xicidaili()

這是結果

最近初學編程(python)，踩了許多坑，天天在error的海洋中的遨遊…第一次做代理池,比上面簡陋多了，記錄一下哈哈

  • 實現功能：運行腳本後抓取網頁上的免費代理IP到mysql，並且mysql自動更新，過濾掉無效IP
  • 準備工作：用mysql workbench新建了名為"hello"的schema，在其下建立了名為"ip_agent"的table,添加"ip"和"port"的column

  • 思路流程：
  1. 抓取網頁上的免費代理IP
  2. 抓取mysql表中的數據
  3. 整合去重
  4. 篩選出有效數據
  5. 存入mysql
  • 代碼部分（python3）：

---- 實踐得出python2和python3幾點區別

  1. urllib2和urllib.request
  2. python2中html=page.read()就可以了，python3後面得加上.decode("utf-8",errors="ignore")，不然會報錯
  3. python2隻能用MySQLdb(這個大小寫很坑有沒有)控制mysql,python3隻能用pymysql，用法差不多

#-*- coding:utf-8 -*-
import urllib.request
import re
import requests
import pymysql

#抓取三個網頁上比較新的免費代理ip和port到_crawl
def _gaoni():
urls=["http://www.ip3366.net/free/?stype=1page=1",
"http://www.ip3366.net/free/?stype=1page=2"]
re_gaoni=re.compile(r"(d{2,3}.d{2,3}.d{2,3}.d{2,3})&s*?
s*?&(d{2,5})")
list_gaoni=[]
maxtrynum=10
for url in urls:
for tries in range(maxtrynum):
try:
response=urllib.request.Request(url)
page=urllib.request.urlopen(response)
except:
if tries &< (maxtrynum-1): continue else: print ("failed") html=page.read().decode("utf-8",errors="ignore")#python3 list_gaoni=list_gaoni+re_gaoni.findall(html) return list_gaoni def _66ip(): urls=["http://www.66ip.cn/index.html", "http://www.66ip.cn/2.html"] re_66ip=re.compile(r"(d{2,3}.d{2,3}.d{2,3}.d{2,3})&&(d{2,5})")
list_66ip=[]
maxtrynum=10
for url in urls:
for tries in range(maxtrynum):
try:
response=urllib.request.Request(url)
page=urllib.request.urlopen(response)
except:
if tries &< (maxtrynum-1): continue else: print ("failed") html=page.read().decode("utf-8",errors="ignore")#python3 list_66ip=list_66ip+re_66ip.findall(html) return list_66ip def _httpdaili(): re_httpdaili=re.compile(r"(d{2,3}.d{2,3}.d{2,3}.d{2,3})&s*?
s*?&(d{2,5})")
list_httpdaili=[]
maxtrynum=10
for tries in range(maxtrynum):
try:
response=urllib.request.Request("http://www.httpdaili.com/mfdl/")
page=urllib.request.urlopen(response)
except:
if tries &< (maxtrynum-1): continue else: print ("failed") html=page.read().decode("utf-8",errors="ignore")#python3 list_httpdaili=list_httpdaili+re_httpdaili.findall(html) return list_httpdaili _crawl=_gaoni()+_66ip()+_httpdaili() #抓取資料庫中ip_agent表的數據到_MySQLdata,清空ip_agent host="127.0.0.1" dbName="hello" user="root" password="password" db=pymysql.connect(host,user,password,dbName,charset="utf8") cur=db.cursor() try: cur.execute("SELECT * FROM ip_agent") _MySQLdata=cur.fetchall() except: print ("can"t fetch MySQLdata") db.commit() try: cur.execute("DELETE FROM ip_agent") db.commit() except: db.rollback() #整合去重 _all=[] for m in _crawl: if m not in _all: _all.append(m) for n in _MySQLdata: if n not in _all: _all.append(n) #篩選出有效數據 _useful=[] for i in _all: proxies={"http":"http://"+i[0]+":"+i[1]} url="http://www.baidu.com" try: requests.get(url,proxies) except: pass else: if i not in _useful: _useful.append(i) #存入資料庫 for m in _useful: param=(m[0],m[1]) sql="INSERT INTO ip_agent VALUES (%s,%s)" cur.execute(sql,param) cur.close() db.commit() db.close()

  • 成果

目前抓了三個網頁，一次大約能抓到五十幾個數據，代碼功能還很簡陋只能滿足基本需求，以後根據需要再完善吧= =

http://code.lujq.me/2016/09/11/如何構建我自己的ip代理池/

用多個tor實例

穩定性要求不高的話，抓代理吧。可以不用自己造輪子，github上有很多現成的爬蟲。
穩定性要求較高的話，還是買代理吧。網上的免費代理，單個請求的成功率能達到50%就很不錯了，有些付費代理的成功率能達到80%以上。
如果是單機運行又不想抓代理或購買代理，可以試試tor，就是請求時間太久了。

買爬蟲是性價比最高的做法。兩三千塊錢夠買一個月ip池三五萬，並發ip數百的資源。可以用錢解決的時候沒必要去做抓免費爬蟲這種費力不討好的事。
自己開發去抓取免費爬蟲的策略是可以用來玩，生產環境根本不夠用。
至於爬蟲有效性校驗，實際生產環境需要的是間隔幾秒就要並發檢測數千個代理的有效性，阻塞的方案全部都不能滿足實際並發的業務。
tornado這種支持非同步的框架比較適合，編寫比較簡單，asyncHTTPClient發請求加個回調處理結果，就可以，如果代理需要輸入帳號密碼，簡單的解決方案是自己把帳號密碼拼成"帳號:密碼"用base64轉碼加到header的Proxy-Authentication裡面。認真查手冊可以看到配置項文檔，proxy_username proxy_password設置進去即可。
AsyncHTTPClient.configure(

None, defaults=dict(proxy_username="username",proxy_password="password"))
# or with force_instance:
client = AsyncHTTPClient(force_instance=True,
defaults=dict(proxy_username="username",proxy_password="password"))

考慮到javascript的天生非同步機制，nodejs也可以加入備選方案。

買ip代理已經接近黑產的邊緣了，在玩黑產的圈裡面，自己養ros撥號是最常見的，簡單的就是用來養微信、淘寶、微博這種帳號賣量。你稍微進圈問問這波人，手上沒有三五百並發的穩定ip都沒法自己養帳號。
隨便搜：什麼高匿代理，動態vpn，刷量。多問幾家就能摸到圈子的邊緣了。

一般免費的肯定都是公共的IP，可以寫腳本去實時更新的到資料庫里，不就成一個ip池了。

之前遇到IP限制我都是從幾大代理平台爬最新的IP出來，由於IP質量不好，還需要用目標網站做測試，能正常打開的放進自己的IP庫，爬蟲再去IP庫請求出錯次數(flag)最少的IP使用，請求出錯將該IP的flag+1，被封則標記一個較大的flag。確保程序不會再使用它。
以上只是在Java爬蟲里實現的，Python還沒試過，或者可能有更簡單的解決方案。

如果爬蟲對穩定性比較高的話，還是買代理ip吧，

http://www.free-proxy-list.net

不夠就買（其實各大高校的ip。。。你懂的

再寫一個爬蟲抓取某些網站的代理ip信息

推薦閱讀：

TAG:Python | IP地址 | Python開發 |