崔庆才flask+scrapy维护cookies过程及代码讲解

浅浅的花香味﹌ 2021-11-22 11:04 198阅读 0赞

源码请访问其github，[https://github.com/Python3WebSpider/CookiesPool][https_github.com_Python3WebSpider_CookiesPool]

下面开始一步一步来讲解代码：

首先看run.py看一下程序的入口

from cookiespool.scheduler import Scheduler
    
    def main():
        s = Scheduler()
        s.run()
    
    if __name__ == '__main__':
        main()

程序入口十分简单，看来是调用了Scheduler()的run函数。

def run(self):
            if API_PROCESS:#如果config.py里API接口服务是True意味着要开启API接口服务
                api_process = Process(target=Scheduler.api)#Process是开启一个进程
                api_process.start()
            
            if GENERATOR_PROCESS:#同理 产生器
                generate_process = Process(target=Scheduler.generate_cookie)
                generate_process.start()
            
            if VALID_PROCESS:#同理 验证器
                valid_process = Process(target=Scheduler.valid_cookie)
                valid_process.start()

看来，崔神是根据config.py里对于API\_PROCESS 等三个变量的设置进行了相应的操作。顾名思义，产生器是抽取cookies的，验证器是检验cookies是否可用的。先看第二个，也就是产生器吧。

@staticmethod
        #产生器
        def generate_cookie(cycle=CYCLE):
            while True:
                print('Cookies生成进程开始运行')
                try:
                    for website, cls in GENERATOR_MAP.items():
                        generator = eval(cls + '(website="' + website + '")')#即WeiboCookiesGenerator(website="weibo")
                        generator.run()
                        print('Cookies生成完成')
                        generator.close()
                        time.sleep(cycle)
                except Exception as e:
                    print(e.args)

看来是读取了静态配置数据，然后调用了相应的函数。我们看下是啥。

# 产生器类，如扩展其他站点，请在此配置
    GENERATOR_MAP = {
        'weibo': 'WeiboCookiesGenerator'
    }

原来是调用了WeiboCookiesGenerator(website="weibo")这个函数。我们去generator.py看一下他干了嘛。

class WeiboCookiesGenerator(CookiesGenerator):
        def __init__(self, website='weibo'):
            """
            初始化操作
            :param website: 站点名称
            :param browser: 使用的浏览器
            """
            CookiesGenerator.__init__(self, website)
            self.website = website
        
        def new_cookies(self, username, password):
            """
            生成Cookies
            :param username: 用户名
            :param password: 密码
            :return: 用户名和Cookies
            """
            return WeiboCookies(username, password, self.browser).main()

CookiesGenerator.__init__(self, website)

我们看上面这一句，找到相应代码

class CookiesGenerator(object):
        def __init__(self, website='default'):
            """
            父类, 初始化一些对象
            :param website: 名称
            :param browser: 浏览器, 若不使用浏览器则可设置为 None
            """
            self.website = website
            self.cookies_db = RedisClient('cookies', self.website)#type=cookies,website=weibo self.name={'cookies':'weibo'}
            self.accounts_db = RedisClient('accounts', self.website)#type=cookies,website=weibo self.name={'accounts':'weibo'}
            self.init_browser()
    
        def __del__(self):
            self.close()
        
        def init_browser(self):
            """
            通过browser参数初始化全局浏览器供模拟登录使用
            :return:
            """
            if BROWSER_TYPE == 'PhantomJS':
                caps = DesiredCapabilities.PHANTOMJS
                caps[
                    "phantomjs.page.settings.userAgent"] = 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/57.0.2987.133 Safari/537.36'
                self.browser = webdriver.PhantomJS(desired_capabilities=caps)
                self.browser.set_window_size(1400, 500)
            elif BROWSER_TYPE == 'Chrome':
                self.browser = webdriver.Chrome()
    ……………………………………该类未完

根据传过来的website="weibo"从而 初始化了self.cookies和accounts\_db，这两个都是数据库操作文件，没什么好看的，想看就自己看db.py吧，这里不解释了。

继续看init\_browse函数，就是用selenium初始化了浏览器而已，因为用了phantomjs所以电脑不会以窗口形式打开浏览器，而是后台模拟。

到这里generator = eval(cls + '(website="' + website + '")')\#即WeiboCookiesGenerator(website="weibo")这句话走完了，在继续看。

generator.run()

def run(self):
            """
            运行, 得到所有账户, 然后顺次模拟登录
            :return:
            """
            accounts_usernames = self.accounts_db.usernames() #获取name对应的hash中所有键值对的key  self.name={'accounts':'weibo'}
            cookies_usernames = self.cookies_db.usernames()#获取name对应的hash中所有键值对的key  self.name={'cookies':'weibo'}
            
            for username in accounts_usernames:
                if not username in cookies_usernames:
                    password = self.accounts_db.get(username)#在name对应的hash中获取根据key获取value
                    print('正在生成Cookies', '账号', username, '密码', password)
                    result = self.new_cookies(username, password)
                    # 成功获取
                    if result.get('status') == 1:
                        cookies = self.process_cookies(result.get('content'))
                        print('成功获取到Cookies', cookies)
                        if self.cookies_db.set(username, json.dumps(cookies)):
                            print('成功保存Cookies')
                    # 密码错误，移除账号
                    elif result.get('status') == 2:
                        print(result.get('content'))
                        if self.accounts_db.delete(username):
                            print('成功删除账号')
                    else:
                        print(result.get('content'))
            else:
                print('所有账号都已经成功获取Cookies')

不断从redis数据库获得账号密码来模拟登录从而获取cookies，我们看result = self.new\_cookies(username, password)，对应函数代码如下：

def new_cookies(self, username, password):
            """
            生成Cookies
            :param username: 用户名
            :param password: 密码
            :return: 用户名和Cookies
            """
            return WeiboCookies(username, password, self.browser).main()

接着找weiboCookies，锁定到了cookies.py，咱们看一下这个文件是干嘛的,首先，参数初始化

class WeiboCookies():
        def __init__(self, username, password, browser):
            self.url = 'https://passport.weibo.cn/signin/login?entry=mweibo&r=https://m.weibo.cn/'
            self.browser = browser
            self.wait = WebDriverWait(self.browser, 20)
            self.username = username
            self.password = password

然后执行main()

def main(self):
            """
            破解入口
            :return:
            """
            self.open()
            if self.password_error():
                return {
                    'status': 2,
                    'content': '用户名或密码错误'
                }
            # 如果不需要验证码直接登录成功
            if self.login_successfully():
                cookies = self.get_cookies()
                return {
                    'status': 1,
                    'content': cookies
                }
            # 获取验证码图片
            image = self.get_image('captcha.png')
            numbers = self.detect_image(image)
            self.move(numbers)
            if self.login_successfully():
                cookies = self.get_cookies()
                return {
                    'status': 1,
                    'content': cookies
                }
            else:
                return {
                    'status': 3,
                    'content': '登录失败'
                }

main执行了open()

def open(self):
            """
            打开网页输入用户名密码并点击
            :return: None
            """
            self.browser.delete_all_cookies()
            self.browser.get(self.url)
            username = self.wait.until(EC.presence_of_element_located((By.ID, 'loginName')))
            password = self.wait.until(EC.presence_of_element_located((By.ID, 'loginPassword')))
            submit = self.wait.until(EC.element_to_be_clickable((By.ID, 'loginAction')))
            username.send_keys(self.username)
            password.send_keys(self.password)
            time.sleep(1)
            submit.click()

看来open函数就是模拟了打开网页模拟登录的操作

再回过头看if语句

if self.password_error():
                return {
                    'status': 2,
                    'content': '用户名或密码错误'
                }
            # 如果不需要验证码直接登录成功
            if self.login_successfully():
                cookies = self.get_cookies()
                return {
                    'status': 1,
                    'content': cookies
                }
            # 获取验证码图片
            image = self.get_image('captcha.png')
            numbers = self.detect_image(image)
            self.move(numbers)
            if self.login_successfully():
                cookies = self.get_cookies()
                return {
                    'status': 1,
                    'content': cookies
                }
            else:
                return {
                    'status': 3,
                    'content': '登录失败'
                }

self.password\_error函数挺简单，不看了，判断是不是登录成功了，第二个if是如果不需要验证码，则直接返回，不然的话开来是需要验证码了。后面是执行验证码破解的操作，我想单独写一个博客，这里不赘述了。总之返回了一个字典，包含status和content。

到这里，回到起点 run函数里的result = self.new\_cookies(username, password)已经执行完了，继续往下走

# 成功获取
                    if result.get('status') == 1:
                        cookies = self.process_cookies(result.get('content'))
                        print('成功获取到Cookies', cookies)
                        if self.cookies_db.set(username, json.dumps(cookies)):#json.dumps()用于#将字典形式的数据转化为字符串，json.loads()用于将字符串形式的数据转化为字典，
                            print('成功保存Cookies')
                    # 密码错误，移除账号
                    elif result.get('status') == 2:
                        print(result.get('content'))
                        if self.accounts_db.delete(username):
                            print('成功删除账号')
                    else:
                        print(result.get('content'))
            else:
                print('所有账号都已经成功获取Cookies')

到这里，run函数也执行完了，产生器结束。再看验证器。

if VALID_PROCESS:#同理 验证器
                valid_process = Process(target=Scheduler.valid_cookie)
                valid_process.start()

看一下valid\_cookie函数

@staticmethod
        #验证器
        def valid_cookie(cycle=CYCLE):
            while True:
                print('Cookies检测进程开始运行')
                try:
                    for website, cls in TESTER_MAP.items():
                        tester = eval(cls + '(website="' + website + '")')
                        tester.run()
                        print('Cookies检测完成')
                        del tester
                        time.sleep(cycle)
                except Exception as e:
                    print(e.args)

我们再看一下TESTER\_MAP是个啥

TESTER_MAP = {
        'weibo': 'WeiboValidTester'
    }

tester=eval(cls+'(website="'+website='")')意思就是调用WeiboValidTester(website="weibo")函数。

接着看看这个函数

class WeiboValidTester(ValidTester):
        def __init__(self, website='weibo'):
            ValidTester.__init__(self, website)

接着看ValidTester.\_\_init\_\_(self,'weibo')函数

class ValidTester(object):
        def __init__(self, website='default'):
            self.website = website
            self.cookies_db = RedisClient('cookies', self.website)
            self.accounts_db = RedisClient('accounts', self.website)

再看tester.run()

def run(self):
            cookies_groups = self.cookies_db.all()##取出name对应的hash中所有的键值对
            for username, cookies in cookies_groups.items():
                self.test(username, cookies)

再看test函数：

def test(self, username, cookies):
            print('正在测试Cookies', '用户名', username)
            try:
                cookies = json.loads(cookies)
            except TypeError:
                print('Cookies不合法', username)
                self.cookies_db.delete(username)
                print('删除Cookies', username)
                return
            try:
                test_url = TEST_URL_MAP[self.website]
                response = requests.get(test_url, cookies=cookies, timeout=5, allow_redirects=False)
                if response.status_code == 200:
                    print('Cookies有效', username)
                else:
                    print(response.status_code, response.headers)
                    print('Cookies失效', username)
                    self.cookies_db.delete(username)
                    print('删除Cookies', username)
            except ConnectionError as e:
                print('发生异常', e.args)

根据代码可以看出他是在检验取出的cookies，就提的检验方法看第二个try，

TEST_URL_MAP = {
        'weibo': 'https://m.weibo.cn/'
    }

就是请求了这个网站，看返回的状态码。如果失效就删除。到这里就完成了对于cookies的检测，代码还是比较简单的。

最后咱们看一下api吧

@staticmethod
        def api():
            print('API接口开始运行')
            app.run(host=API_HOST, port=API_PORT)

就是一个flask框架生成web接口，还是不看了吧，没啥意思，会flask的一看就懂，不用讲。

总结：

稍微总结一下这个项目吧，核心就是这个scheduler调度器，里面调用了其他工具类和工具函数。

工具之一：产生器generator.py,主要负责就是模拟用户登录来获取cookies

工具之二：验证器tester.py，用于检验redis里的cookies是否还有效。

两个工具都用到了db.py，也就是封装了一些对于redis数据库的增删改查操作，相当基础。

另外config.py里面写死了一些用到的参数。