[python]图片公式多线程验证ip

发表于 | 分类于 | | 阅读次数:

通过前两篇文章
[Python]40行代码实现公式转换成图片，手把手教你从模仿到实现
[Python]公式转图片程序的反爬虫改进
我们的半自动获取ip并验证功能已经实现了，如果你运行了，你就会发现很棒但是很慢，最慢的就是验证ip，今天我们就解决这个问题。

最好的方式是：用多线程方式，使用代理去访问某个网站，然后输出可用的代理。
python 爬虫 ip池怎么做？ - Kaito的回答 - 知乎
https://www.zhihu.com/question/47464143/answer/124035743

嗯知道关键词了，搜索「多线程验证ip」，非常多结果，选择自己能看懂思路的，大概知道哪一部分是干什么的就行。

我选择了一个http://www.cnblogs.com/sjzh/p/5990152.html

这是他的源程序，模仿的第一步是分析这个源程序，看看这个源程序方不方便改成我要的功能。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
import urllib.request
import urllib
import re
import time
import random
import socket
import threading
#抓取代理IP
ip_totle=[]
for page in range(2,6):
    url='http://ip84.com/dlgn/'+str(page)
    #url='http://www.xicidaili.com/nn/'+str(page) #西刺代理
    headers={"User-Agent":"Mozilla/5.0 (Windows NT 10.0; WOW64)"}
    request=urllib.request.Request(url=url,headers=headers)
    response=urllib.request.urlopen(request)
    content=response.read().decode('utf-8')
    print('get page',page)
    pattern=re.compile('<td>(\d.*?)</td>')  #截取<td>与</td>之间第一个数为数字的内容
    ip_page=re.findall(pattern,str(content))
    ip_totle.extend(ip_page)
    time.sleep(random.choice(range(1,3)))
#打印抓取内容
print('代理IP地址     ','\t','端口','\t','速度','\t','验证时间')
for i in range(0,len(ip_totle),4):
    print(ip_totle[i],'    ','\t',ip_totle[i+1],'\t',ip_totle[i+2],'\t',ip_totle[i+3])
#整理代理IP格式
proxys = []
for i in range(0,len(ip_totle),4):
    proxy_host = ip_totle[i]+':'+ip_totle[i+1]
    proxy_temp = {"http":proxy_host}
    proxys.append(proxy_temp)

proxy_ip=open('proxy_ip.txt','w')  #新建一个储存有效IP的文档
lock=threading.Lock()  #建立一个锁
#验证代理IP有效性的方法
def test(i):
    socket.setdefaulttimeout(5)  #设置全局超时时间
    url = "http://quote.stockstar.com/stock"  #打算爬取的网址
    try:
        proxy_support = urllib.request.ProxyHandler(proxys[i])
        opener = urllib.request.build_opener(proxy_support)
        opener.addheaders=[("User-Agent","Mozilla/5.0 (Windows NT 10.0; WOW64)")]
        urllib.request.install_opener(opener)
        res = urllib.request.urlopen(url).read()
        lock.acquire()     #获得锁
        print(proxys[i],'is OK')        
        proxy_ip.write('%s\n' %str(proxys[i]))  #写入该代理IP
        lock.release()     #释放锁
    except Exception as e:
        lock.acquire()
        print(proxys[i],e)
        lock.release()
#单线程验证
'''for i in range(len(proxys)):
    test(i)'''
#多线程验证    
threads=[]
for i in range(len(proxys)):
    thread=threading.Thread(target=test,args=[i])
    threads.append(thread)
    thread.start()
#阻塞主进程，等待所有子线程结束
for thread in threads:
    thread.join()
    
proxy_ip.close()  #关闭文件

看下结构就行，程序到大概是两部分，第一部分抓取ip，第二部分验证ip。我们自己有抓取ip的程序所以前面可以不要，列表proxys[]用来存储爬取的ip并且传递给第二部分使用，那把我们爬取的ip复制到这里就行了，注意看看格式，proxy_temp = {“http”:proxy_host}人家的格式是字典形式的，OK为了避免改第二部分的程序，我们把ip也改成这种格式，对后面的程序就没有影响了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
import urllib.request
import urllib
import re
import time
import random
import socket
import threading
#抓取代理IP

#整理代理IP格式
ip_list = ['27.40.146.105:61234', '123.163.154.134:808', '110.200.155.21:8118', '116.192.26.17:1080', '39.72.54.28:8118', '112.93.236.224:9797', '115.46.77.233:8123', '114.231.243.208:30544', '14.153.54.167:3128', '117.66.167.30:8118', '110.73.8.147:8123', '219.136.173.59:9999', '116.31.2.113:8118']
all_ip=[]
proxys=[]
for i in ip_list:
     proxy = {'http':i}
     proxys.append(proxy)
#保证的格式和原来一样
# proxy_ip=open('proxy_ip.txt','w')  #新建一个储存有效IP的文档
lock=threading.Lock()  #建立一个锁
#验证代理IP有效性的方法
def test(i):
    socket.setdefaulttimeout(5)  #设置全局超时时间
    url = "http://quote.stockstar.com/stock"  #打算爬取的网址
    try:
        proxy_support = urllib.request.ProxyHandler(proxys[i])
        opener = urllib.request.build_opener(proxy_support)
        opener.addheaders=[("User-Agent","Mozilla/5.0 (Windows NT 10.0; WOW64)")]
        urllib.request.install_opener(opener)
        res = urllib.request.urlopen(url).read()
        lock.acquire()     #获得锁
        print(proxys[i],'is OK')        
        # proxy_ip.write('%s\n' %str(proxys[i]))  #写入该代理IP
        lock.release()     #释放锁
    except Exception as e:
        lock.acquire()
        print(proxys[i],e)
        lock.release()
#单线程验证
'''for i in range(len(proxys)):
    test(i)'''
#多线程验证    
threads=[]
for i in range(len(proxys)):
    thread=threading.Thread(target=test,args=[i])
    threads.append(thread)
    thread.start()
#阻塞主进程，等待所有子线程结束
for thread in threads:
    thread.join()
    
# proxy_ip.close()  #关闭文件

把我们的ip_list复制过来，修改格式，然后把一下读写操作都去掉，我们不要读写。运行一下

OK，有错误，没事，有错误就对了，通常我们找到的程序都不能直接用，我们试试把提示的错误改掉，看看能不能解决。
搜索「fixture ‘i’ not found」

我搜索了很多结果，都没有直接告诉我答案，我都准备换一个代码了，我突然想到搜索过程中错误提示是跟一个叫pytest的关键词一起出现，会不会是函数名不能要test？

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
import urllib.request
import urllib
import re
import time
import random
import socket
import threading
#抓取代理IP

#整理代理IP格式
ip_list = ['27.40.146.105:61234', '123.163.154.134:808', '110.200.155.21:8118', '116.192.26.17:1080', '39.72.54.28:8118', '112.93.236.224:9797', '115.46.77.233:8123', '114.231.243.208:30544', '14.153.54.167:3128', '117.66.167.30:8118', '110.73.8.147:8123', '219.136.173.59:9999', '116.31.2.113:8118']
proxys=[]
for i in ip_list:
     proxy = {'http':i}
     proxys.append(proxy)
#保证的格式和原来一样
# proxy_ip=open('proxy_ip.txt','w')  #新建一个储存有效IP的文档
lock=threading.Lock()  #建立一个锁
#验证代理IP有效性的方法
def iptest(i):
    socket.setdefaulttimeout(5)  #设置全局超时时间
    url = "http://quote.stockstar.com/stock"  #打算爬取的网址
    try:
        proxy_support = urllib.request.ProxyHandler(proxys[i])
        opener = urllib.request.build_opener(proxy_support)
        opener.addheaders=[("User-Agent","Mozilla/5.0 (Windows NT 10.0; WOW64)")]
        urllib.request.install_opener(opener)
        res = urllib.request.urlopen(url).read()
        lock.acquire()     #获得锁
        print(proxys[i],'is OK')        
        # proxy_ip.write('%s\n' %str(proxys[i]))  #写入该代理IP
        lock.release()     #释放锁
    except Exception as e:
        lock.acquire()
        print(proxys[i],e)
        lock.release()
#单线程验证
'''for i in range(len(proxys)):
    test(i)'''
#多线程验证    
threads=[]
for i in range(len(proxys)):
    thread=threading.Thread(target=iptest,args=[i])
    threads.append(thread)
    thread.start()
#阻塞主进程，等待所有子线程结束
for thread in threads:
    thread.join()
    
# proxy_ip.close()  #关闭文件

终于解决了，如果没有想到我会怎么做？我会换一个程序继续修改。模仿会出现各种问题，学习也是在解决问题中进行的，所以遇到问题是好事，如果你复制粘贴后就实现了你的功能，那你其实学到的就很少，但是如果你遇到各种问题，并且解决了一部分，即使还没成功，你已经在进步了。这次我是再也不会忘记函数名不能叫test了。

最后只需要把我们需要的内容打印出来即可。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
import urllib.request
import urllib
import time
import socket
import threading

ip_list=['27.40.146.105:61234', '123.163.154.134:808', '110.200.155.21:8118', '116.192.26.17:1080', '39.72.54.28:8118', '112.93.236.224:9797', '115.46.77.233:8123', '114.231.243.208:30544', '14.153.54.167:3128', '117.66.167.30:8118', '110.73.8.147:8123', '219.136.173.59:9999', '116.31.2.113:8118']

# 验证代理IP有效性的方法
proxys=[]
all_ip=[]
for i in ip_list:
     proxy = {'http':i}
     # print(type(proxy))
     proxys.append(proxy)

lock = threading.Lock()  # 建立一个锁
#不可以使用test作为函数或者文件名
def iptest(i):
    socket.setdefaulttimeout(5)  # 设置全局超时时间
    url = "http://www.baidu.com/"  # 打算爬取的网址
    try:
          proxy_support = urllib.request.ProxyHandler(proxys[i])
          opener = urllib.request.build_opener(proxy_support)
          opener.addheaders = [("User-Agent", "Mozilla/5.0 (Windows NT 10.0; WOW64)")]
          urllib.request.install_opener(opener)
          res = urllib.request.urlopen(url).read()
         # 获取锁，用于线程同步
          lock.acquire()  # 获得锁
          print(proxys[i], 'is OK')
          all_ip.append(str(proxys[i]['http']))
          # 释放锁，开启下一个线程
          lock.release()  # 释放锁
    except Exception as e:
          lock.acquire()
          print(proxys[i], e)
          lock.release()
    # 单线程验证
# 多线程验证
threads = []
start = time.clock()
# print(len(proxys))
for i in range(len(proxys)):
    thread = threading.Thread(target=iptest, args=[i])
    threads.append(thread)
    thread.start()
# 阻塞主进程，等待所有子线程结束
for thread in threads:
    thread.join()


end = time.clock()
print(all_ip)
print("开始时间: %f s" % start)
print("结束时间: %f s" % end)
print("校验IP耗时: %f s" % (end - start))

完成了功能，高兴之余别忘了再看看程序，这个程序大概意思是，有多少个ip就创建多少个线程，然后每个线程都运行的是iptest(i)这个函数，就等于是同时运行着多个iptest(i)函数，所以虽然我还不知道线程的一些知识，是不是我修改iptest(i)函数，就可以得到并行运行的其他功能呢？又可以做很多有意思的改进了。

到这里我们的公式转图片程序的半自动化功能就全部实现了，你现在有了爬取ip，多线程验证等手段，你已经可以做一个你自己想要的功能了，还等什么燥起来吧。

关注微信公众号，回复【公式图片】即可获取源码，同时欢迎加我的个人微信进行交流哦。

zinghd wechat
期待您的关注
您的赞赏是最大的支持