IP代理软件,http代理,API提取-黑洞HTTP

黑洞HTTP

您的位置: 首页 > 新闻资讯 > HTTP代理 > 正文

如何自建爬虫代理IP池?

发布时间:2019年11月07日 来源:互联网

    用Python抓取免费的代理IP并检验其有效性。这篇文章就是在其基础上尝试完整实现了一个代理IP池:

如何自建爬虫代理IP池?

    后端用的是sqlite3,因为之前爬网易云音乐评论时后端就是用的它,最后存了大概一万条评论,读取写入时间都很不错,我估计代理池里的IP最后也只会是这个量级。还有就是sqlite3用起来非常方便简单。不过我仍然在Sqlite3api.py里对其进行了封装。


    除了sqlite3,程序里还维护了一个名为iplist的列表,它的大小由配置项iplist_pool_size决定。这个列表的作用就是对db的cache,里面只会存储检验有效的IP。用户通过API方法proxy_get_one_ip()获取表头位置的IP,然后再将这个IP添加到列表末尾,实现循环利用。这里我的思路是proxy_get_one_ip()只会尽量保证返回一个有效的IP,而用户如果发现IP无效,可以通过proxy_report_invalid_ip()告知代理池,代理池会将这个IP从iplist列表和DB中同时删除。


    如何保证iplist列表和DB中内容的一致性?这里参考了Cache的同步理念,就是读取的时候只读取cache里的内容,写入的时候要同时写入list和db。


    目前,我采用的IP抓取方式是从大象代理的API直接获取,之前我花5块钱买了10000个IP,现在才用了2000个不到。虽然这里也能抓取很多免费代理IP,但是我想说的是,即使是我花钱买的IP,有效率也才百分之二三十的样子,有效时间还不知道。抓取免费代理IP的方法前面的文章里有,就不赘述了。我个人还是推荐花5块钱买一下,反正又不贵。


    程序还提供了另一个API方法proxy_fill_db()用来抓取并往数据库里写入有效IP,这个方法主要作用是扩充db里的IP数量。


    当proxy_get_one_ip()被调用时,代理池先检测iplist是否有内容,有直接返回第一个元素,没有的话会调用内部函数__refresh():


    __refresh()方法的最终目的是将iplist列表里IP数量补足到至少iplist_pool_size的大小。它会先从数据库里抓取需要的IP数量;如果数据库里的IP不够,则会通过__get_ip_thr_api()从代理API处获取数量为N的IP值。


    关于N 值的确定,首先先做一次采样,先从代理API处获取10个IP,并检查它们的有效性,假设10个里面有4个能用,此时会计算一个ratio = 4/10 = 0.4 ,之后假设我们还需要missing_ip_num个IP,N的值则是missing_ip_num/ratio*factor,这里factor可以是任何整数,因为这个ratio值不一定每次都这么高,据我测试大多数时候是0.2到0.3的样子,有时也会是0.0。总之,factor越大,则有可能获得更多的有效IP,同时消耗的时间也更长。最后,代理池会对每个IP做校验,如果是有效IP,则会写入iplist和db中。


    这里我还想提一点,就是我在之前的文章里写了一种设想,就是为了保证IP有效率,在代理池给出IP时,概率性的选择要不要检验一下这个IP,比如说这里,如果说p1=0.4, N=5000, p2=0.9,则x/N=0.83,就是说83%的概率在给出IP前要检查,如果检查失败的话,就要重新获取IP。因为p1只有0.4,甚至更低,所以一般情况下可能要2-3次的检查才能保证这个IP是有效的。我觉得两种方法都是可以的,等待时间其实是一样的,只不过一个是在初始化的时候等,一个是在返回IP的时候等。


    检查一个IP是否有效,目前就是用这个IP访问百度页面,检查其返回码是不是200 OK。


    另外还包含了一些测试代码。


    这里还想补充一个关于unittest的库,因为之前写C代码,要写很多测试代码,而我一般命名测试函数为unittest,后来我发现Python里也有类似的函数,一开始我以为和C差不多,最近我才知道原来有一个叫unittest的测试框架,可以用来编写测试用例,实现自动化测试。不过这里我就不展开说了。


    问题


    目前最大的问题就是缺少多线程的支持,我会尽快补上的。


    如果有需要,我也会补上获取免费代理IP的部分。


    这里没有对IP做重复性检测,但是我想了一下,这种情况在目前的实现里是几乎不可能出现的。之后如果增加多个source,写入db之前还需要做重复性检查。


    还有一点是,因为我之前一直习惯写C代码,所有很少用在Python里很少用类,我不确定这是不是有问题。


    代码


    Configure.py


    # proxy related


    daxiang_proxy_tid = #order number


    iplist_pool_size = 10


    #iplist_thread_num = 1


    # DB


    db_name = 'proxy.db'


    Sqlite3api.py


    import sqlite3


    import os


    import Configure


    def sqlite3_init():


    try:


    conn = sqlite3.connect(Configure.db_name)


    except Exception as e:


    print ('sqlite3 init fail.')


    print (e)


    return conn


    def sqlite3_execute(conn, sql, args = None):


    data = None


    try:


    cur = conn.cursor()


    if args:


    cur.execute(sql, args)


    else:


    cur.execute(sql)


    data = cur.fetchall()


    except Exception as e:


    print (e, "[SQL]:" + sql.strip())


    conn.rollback()


    conn.commit()


    if data:


    return data


    return None


    def sqlite3_close(conn):


    conn.close()


    def unittest():


    conn = sqlite3_init()


    sqlite3_execute(conn, "CREATE TABLE stocks (date text, trans text, symbol text, qty real, price real)")


    sqlite3_execute(conn, "INSERT INTO stocks VALUES ('2006-01-05','BUY','RHAT',100,35.14)")


    sqlite3_execute(conn, "INSERT INTO stocks VALUES ('2006-03-28', 'BUY', 'IBM', 1000, 45.00)")


    sqlite3_execute(conn, "INSERT INTO stocks VALUES ('2006-04-05', 'BUY', 'MSFT', 1000, 72.00)")


    sqlite3_execute(conn, "INSERT INTO stocks VALUES ('2006-04-06', 'SELL', 'IBM', 500, 53.00)")


    assert 4 == sqlite3_execute(conn, "SELECT count(*) FROM stocks")[0][0]


    sqlite3_execute(conn, "DROP TABLE stocks")


    sqlite3_close(conn)


    if __name__ == '__main__':


    unittest()


    proxy.py


    import requests


    import ast


    from random import choice


    import os


    import time


    import Configure


    import Sqlite3api as sqlite3


    iplist = []


    def proxy_init():


    conn = sqlite3.sqlite3_init()


    sqlite3.sqlite3_execute(conn, "CREATE TABLE proxy (t real, valid real, ip text)")


    sqlite3.sqlite3_close(conn)


    # main entry, return an ip with certain protocols(either http or https)


    def proxy_get_one_ip(protocol='all'):


    global iplist


    if not iplist:


    __refresh(protocol)


    # pop() return and remove the last element in the list by default, amazing!


    # here I need op on the first element


    ip = iplist.pop(0)


    iplist.append(ip)


    return ip


    def proxy_report_invalid_ip(ip, protocol=None):


    # sync write op


    # remove it from list


    iplist.remove(ip)


    # set false in db


    conn = sqlite3.sqlite3_init()


    sqlite3.sqlite3_execute(conn, "DELETE FROM proxy WHERE ip = '{0:s}'".format(ip))


    sqlite3.sqlite3_close(conn)


    return proxy_get_one_ip(protocol) if protocol else None


    def proxy_fill_db(num = 20):


    samples = __get_ip_thr_api(num)


    cnt = 0


    conn = sqlite3.sqlite3_init()


    for sample in samples:


    ip = "{0:s}:{1:d}".format(sample.get('host'), sample.get('port'))


    if __validation(ip):


    sqlite3.sqlite3_execute(conn, "INSERT INTO proxy VALUES ({0:d},1,'{1:s}')".format(int(time.time()*1000), ip))


    #print ("Found one: {0:s}.".format(ip))


    cnt += 1


    sqlite3.sqlite3_close(conn)


    return cnt


    # get ip through api from http://www.daxiangdaili.com


    def __get_ip_thr_api(num=1, protocol='all'):


    tid = Configure.daxiang_proxy_tid


    url = "http://tvp.daxiangdaili.com/ip/?tid={0:d}&num={1:d}&delay=3&category=2&sortby=time&filter=on&format=json".format(tid, num)


    if protocol == 'https':


    url += "&protocol=https"


    try:


    response = requests.get(url)


    content = None


    if response.status_code == requests.codes.ok:


    content = response.text


    except Exception as e:


    print (e)


    return ast.literal_eval(content.strip())


    # validate ip,addr format: [ip:port]


    def __validation(addr):


    proxies = {


    "http": "http://{0}".format(addr),


    "https": "http://{0}".format(addr)


    }


    header = {}


    header['user-agent'] = choice(Configure.FakeUserAgents)


    try:


    response = requests.get("https://www.baidu.com", headers=header, proxies=proxies, timeout=5)


    except Exception as e:


    #print (e)


    return False


    else:


    if response.status_code == requests.codes.ok:


    return True


    def __cal_sample_ratio(sampe_size, protocol):


    # collect sample to calculate validity ratio


    cnt, valid = sampe_size, 0


    addrs = __get_ip_thr_api(cnt, protocol)


    conn = sqlite3.sqlite3_init()


    for addr in addrs:


    ip = "{0:s}:{1:d}".format(addr.get('host'), addr.get('port'))


    if __validation(ip):


    valid += 1


    # Logically, here I should not add this ip to db nor iplist


    # because this func is for ratio calculation


    # However, it's not easy to get valid ip, I don't any waste here


    iplist.append(ip)


    sqlite3.sqlite3_execute(conn, "INSERT INTO proxy VALUES ({0:d},1,'{1:s}')".format(int(time.time()*1000), ip))


    sqlite3.sqlite3_close(conn)


    # calculate validity ratio


    ratio = float(valid/cnt)


    print ("ratio is {0}".format(ratio))


    return ratio


    def __refresh(protocol='all'):


    print ("[Start updating pool]")


    global iplist


    # need this number of valid ips to fill the pool


    missing_ip_num = Configure.iplist_pool_size - len(iplist)


    # (1) get them from db


    #       normally the db should return enough valid ips


    conn = sqlite3.sqlite3_init()


    data = sqlite3.sqlite3_execute(conn, "SELECT ip, valid FROM proxy where valid = 1 order by t limit {0:d}".format(missing_ip_num))


    sqlite3.sqlite3_close(conn)


    for item in data:


    l = list(item)


    iplist.append(l[0])


    # check the results


    if len(iplist) >= Configure.iplist_pool_size:


    print ("Enough valid ips in pool. Refresh finished.")


    return


    else:


    missing_ip_num = Configure.iplist_pool_size - len(iplist)


    print ("Still need {0:d} ips after retrieving from db.".format(missing_ip_num))


    # (2) this means even used all valid ips in db, pool is still not full


    #       then get more valid ips from api or other sources


    # calculate ratio


    ratio = __cal_sample_ratio(10, protocol)


    # Since inside __cal_sample_ratio() would add new ips to db


    # I did one more time check here


    if len(iplist) >= Configure.iplist_pool_size:


    print ("Enough valid ips in pool. Refresh finished.")


    return


    # higher means more valid ips, but slower to process


    factor = 1


    # __sample_size__ means based on the validity ratio, this number of ips are needed


    # to filter enough valid ips, approximately


    if ratio == 0.00:


    # extremely bad sample


    sample_size = missing_ip_num * factor * 2


    else:


    sample_size = int(1/ratio * missing_ip_num * factor)


    print ("Need to collect {0:d} ips for validation test".format(sample_size))


    samples = __get_ip_thr_api(sample_size, protocol)


    # TODO multi threads


    # if sample_size > a certain number, do multi threads


    # else:


    conn = sqlite3.sqlite3_init()


    for sample in samples:


    ip = "{0:s}:{1:d}".format(sample.get('host'), sample.get('port'))


    if __validation(ip):


    # sync write operation


    iplist.append(ip)


    sqlite3.sqlite3_execute(conn, "INSERT INTO proxy VALUES ({0:d},1,'{1:s}')".format(int(time.time()*1000), ip))


    print ("Found one: {0:s}.".format(ip))


    if len(iplist) >= Configure.iplist_pool_size:


    print (len(iplist))


    break


    sqlite3.sqlite3_close(conn)


    # Note here I didn't check the list length again.


    # so the list may be still not full because there are


    # not enough valid ips in the sample


    # Generally, this should not happen often, one should


    # call proxy_fill_db() enough times to make sure


    # db has enough valid ips already for use


    print ("[pool updated.]")


    def unittest():


    global iplist


    assert 1 == len(__get_ip_thr_api(1, 'https'))


    assert 1 == len(__get_ip_thr_api(1))


    proxy_init()


    conn = sqlite3.sqlite3_init()


    cnt = sqlite3.sqlite3_execute(conn, "SELECT count(*) FROM proxy")[0][0]


    res = proxy_fill_db(10)


    assert (cnt + res) == sqlite3.sqlite3_execute(conn, "SELECT count(*) FROM proxy")[0][0]


    if res > 0:


    ip = proxy_get_one_ip()


    cnt1 = sqlite3.sqlite3_execute(conn, "SELECT count(*) FROM proxy")[0][0]


    proxy_report_invalid_ip(ip)


    cnt2 = sqlite3.sqlite3_execute(conn, "SELECT count(*) FROM proxy")[0][0]


    assert 1 == (cnt1 - cnt2)


    assert 0 == sqlite3.sqlite3_execute(conn, "SELECT count(*) FROM proxy where ip = '{0:s}'".format(ip))[0][0]


    sqlite3.sqlite3_close(conn)


    if __name__ == '__main__':


    unittest()


    作者:小温侯


    链接:https://www.jianshu.com/p/64980c31036c


    来源:简书


    著作权归作者所有。商业转载请联系作者获得授权,非商业转载请注明出处。


相关文章内容简介

1 如何自建爬虫代理IP池?

    用Python抓取免费的代理IP并检验其有效性。这篇文章就是在其基础上尝试完整实现了一个代理IP池:    后端用的是sqlite3,因为之前爬网易云音乐评论时后端就是用的它,最后存了大概一万条评论,读取写入时间都很不错,我估计代理池里的IP最后也只会是这个量级。还有就是sqlite3用起来非常方便简单。不过我仍然在Sqlite3api.py里对其进行了封装。    除了sqlite3,程序里还维护了一个名为iplist的列表,它的大小由配置项iplist_pool_size决定。这个列表的作用就是对db的cache,里面只会存储检验有效的IP。用户通过API方法proxy_get_one_ip()获取表头位置的IP,然后再将这个IP添加到列表末尾,实现循环利用。这里我的思路是proxy_get_one_ip()只会尽量保 [阅读全文]

热门标签

最新标签

推荐阅读

  1. 30

    2019-03

    爬虫过程中的代理ip使用

    在实际的爬虫抓取的过程中,由于会存在恶意采集或者恶意攻击的情况,很多网站都会设置相应的防爬取机制,通常防爬程序都是通过ip来识别机器人用户的,因此充足可用的ip信息可以为我们

  2. 05

    2019-08

    移动http代理服务器地址

    http代理服务器在现代生活中的应用越来越多。总有人说IP代理HTTP代理,那么什么是真正的http代理服务器呢简单的说,HTTP代理和我们现实中的各种代理商是一样的,它就是我们在访问目标网络

  3. 15

    2019-06

    怎么获取国内http代理ip?

    我们在上网时候常会看见“www.***.com”的网站域名,无论是看新闻或是闲逛贴吧、论坛,或是上网购物还是资源下载,以“www”开头的域名是最为常用的。但是这类网站用的协议是http协议,这是

  4. 04

    2019-04

    网站判断爬虫在采集数据时,可以试用代理IP吗?

    我们在用Python爬虫进行信息的采集时,经常被禁,有时提示访问过于频繁,有时返回一些错误码等等,那么网站是如何掌握爬虫在进行信息采集的呢?

  5. 26

    2019-11

    代理ip如何突破限制访问?

    有了 IP 我们就必须验证一下,因为网络上很多的 代理IP 实际上是不能用的,及时专业的代理服务商也是一样。在更换完IP后,我们可以上百度搜索IP,如果显示出来与你自己不一样的 IP地址 ,

  6. 28

    2019-08

    如何利用http代理IP为好友投票

    在互联网应用逐渐广泛的今天,http代理IP给我们的工作生活带来了很大的方便。所以,http代理IP的产品代理商也是逐渐增加,很多用户陷入了选择困难症中,不知道应该怎样选择。