Python搭建代理IP池实现接口设置与整体调度

2019-11-25 11:31:47

字体：大中小

来源：转载

供稿：网友

接口模块需要用 API 来提供对外服务的接口，当然也可以直接连数据库来取，但是这样就需要知道数据库的连接信息，不太安全，而且需要配置连接，所以一个比较安全和方便的方式就是提供一个 Web API 接口，通过访问接口即可拿到可用代理

代码地址：https://github.com/Stevengz/Proxy_pool

另外三篇：
Python搭建代理IP池（一）- 获取 IP
Python搭建代理IP池（二）- 存储 IP
Python搭建代理IP池（三）- 检测 IP

添加设置

添加模块开关变量

setting.py

# 数据库地址HOST = '127.0.0.1'# MySql端口MYSQL_PORT = 3306# MySQl用户名、密码MYSQL_USERNAME = '***'MYSQL_PASSWORD = '***'# 数据库名SQL_NAME = 'test'# 代理等级MAX_SCORE = 30MIN_SCORE = 0INITIAL_SCORE = 10VALID_STATUS_CODES = [200, 302]# 代理池数量界限POOL_UPPER_THRESHOLD = 1000# 检查周期TESTER_CYCLE = 20# 获取周期GETTER_CYCLE = 300# 测试API，建议抓哪个网站测哪个TEST_URL = 'http://www.baidu.com'# API配置API_HOST = '0.0.0.0'API_PORT = 5555# 开关TESTER_ENABLED = TrueGETTER_ENABLED = TrueAPI_ENABLED = True# 最大批测试量BATCH_TEST_SIZE = 10

TESTER_ENABLED、GETTER_ENABLED、API_ENABLED 都是布尔类型，True 或者 False。标明了测试模块、获取模块、接口模块的开关，如果为 True，则代表模块开启

定义接口

使用框架：Flask

api.py

from flask import Flask, gfrom db import MySqlClient__all__ = ['app']app = Flask(__name__)def get_conn(): if not hasattr(g, 'mysql'):  g.mysql = MySqlClient() return g.mysql@app.route('/')def index(): return '<h2>Welcome to Proxy Pool System</h2>'# 随机代理@app.route('/random')def get_proxy(): conn = get_conn() return conn.random()# 代理池总量@app.route('/count')def get_counts(): conn = get_conn() return str(conn.count())if __name__ == '__main__': app.run()

声明了一个 Flask 对象，定义了三个接口，分别是首页、随机代理页、获取数量页。

只需要访问对应的接口即可获取到可用代理：

调度模块

调用定义的获取、存储、检测三个模块，将这三个模块通过多进程的形式运行起来

scheduler.py

import timefrom multiprocessing import Processfrom api import appfrom getter import Getterfrom tester import Testerfrom db import MySqlClientfrom setting import *class Scheduler(): # 定时测试代理 def schedule_tester(self, cycle=TESTER_CYCLE):  tester = Tester()  while True:   print('测试器开始运行')   tester.run()   time.sleep(cycle)  # 定时获取代理 def schedule_getter(self, cycle=GETTER_CYCLE):  getter = Getter()  while True:   print('开始抓取代理')   getter.run()   time.sleep(cycle)  # 开启API def schedule_api(self):  app.run(API_HOST, API_PORT)  def run(self):  print('代理池开始运行')    if TESTER_ENABLED:   tester_process = Process(target=self.schedule_tester)   tester_process.start()    if GETTER_ENABLED:   getter_process = Process(target=self.schedule_getter)   getter_process.start()    if API_ENABLED:   api_process = Process(target=self.schedule_api)   api_process.start()if __name__ == "__main__": scheduler = Scheduler() scheduler.run()

启动入口是 run() 方法，分别判断了三个模块的开关，如果开启的话，就新建一个 Process 进程，设置好启动目标，然后调用 start() 方法运行，这样三个进程就可以并行执行，互不干扰