crawler_xigua/main/get_signature.py

# -*- coding: utf-8 -*-
# @Author: wangkun
# @Time: 2022/9/21
import json
import os
import sys
import time
from selenium import webdriver
from selenium.webdriver.common.desired_capabilities import DesiredCapabilities
sys.path.append(os.getcwd())
from main.common import Common
from main.feishu import Feishu
proxies = {'http': None, 'https': None}


class GetSignature:
    @classmethod
    def get_signature(cls, log_type):
        try:
            # 打印请求配置
            ca = DesiredCapabilities.CHROME
            ca["goog:loggingPrefs"] = {"performance": "ALL"}

            # 不打开浏览器运行
            chrome_options = webdriver.ChromeOptions()
            chrome_options.add_argument("headless")
            chrome_options.add_argument(f'user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.79 Safari/537.36')

            # driver初始化
            Common.logger(log_type).info('初始化 webdriver')
            # driver = webdriver.Chrome(desired_capabilities=ca)
            driver = webdriver.Chrome(desired_capabilities=ca, options=chrome_options)

            driver.implicitly_wait(10)
            Common.logger(log_type).info('打开『北京旅游等着瞧』个人主页')
            driver.get('https://www.ixigua.com/home/2176672314697038')
            time.sleep(5)

            # 向上滑动 1000 个像素
            Common.logger(log_type).info('向上滑动页面')
            driver.execute_script('window.scrollBy(0, 2000)')
            Common.logger(log_type).info('休眠 5s')
            time.sleep(5)
            Common.logger(log_type).info('刷新页面')
            driver.refresh()
            Common.logger(log_type).info('休眠 1s')
            time.sleep(1)
            Common.logger(log_type).info('解析signature')
            logs = driver.get_log("performance")
            # Common.logger(log_type).info('已获取logs:{}\n', logs)
            Common.logger(log_type).info('退出浏览器')
            driver.quit()
            for line in logs:
                msg = json.loads(line['message'])
                if 'params' not in msg['message']:
                    pass
                elif 'documentURL' not in msg['message']['params']:
                    pass
                elif 'www.ixigua.com/home/2176672314697038' not in msg['message']['params']['documentURL']:
                    pass
                elif 'url' not in msg['message']['params']['request']:
                    pass
                elif '_signature' not in msg['message']['params']['request']['url']:
                    pass
                elif 'web/user/settings' not in msg['message']['params']['request']['url']:
                    pass
                else:
                    url = msg['message']['params']['request']['url']
                    signature = url.split('_signature=')[-1].split('&')[0]
                    Common.logger(log_type).info('url:{}\n', url)
                    Common.logger(log_type).info('signature:{}\n', signature)

                    get_signature_time = time.strftime('%Y/%m/%d %H:%M:%S', time.localtime(time.time()))
                    values = [[get_signature_time, signature]]
                    Feishu.insert_columns(log_type, 'xigua', '6tZHhs', 'ROWS', 1, 2)
                    time.sleep(0.5)
                    Feishu.update_values(log_type, 'xigua', '6tZHhs', 'A2:B2', values)
                    Common.logger(log_type).info('signature写入飞书成功:{}\n', signature)
                    cls.del_signature(log_type)

                # if 'message' in line \
                #         and 'message' in json.loads(line['message'])\
                #         and 'params' in json.loads(line['message'])['message']\
                #         and 'response' in json.loads(line['message'])['message']['params']\
                #         and 'url' in json.loads(line['message'])['message']['params']['response']\
                #         and '_signature' in json.loads(line['message'])['message']['params']['response']['url']\
                #         and 'web/user/settings' in json.loads(line['message'])['message']['params']['response']['url']:
                #     url = json.loads(line['message'])['message']['params']['response']['url']
                #     signature = url.split('_signature=')[-1]
                #     if len(signature) < 50:
                #         pass
                #     else:
                #         get_signature_time = time.strftime('%Y/%m/%d %H:%M:%S', time.localtime(time.time()))
                #         values = [[get_signature_time, signature]]
                #         Feishu.update_values(log_type, 'xigua', '6tZHhs', 'A2:B2', values)
                #         Common.logger(log_type).info('signature写入飞书成功:{}\n', signature)
                #         cls.del_signature(log_type)
                # else:
                #     Common.logger(log_type).info('未获取到signature:{}\n', line)

        except Exception as e:
            Common.logger(log_type).error('get_signature异常:{}', e)

    @classmethod
    def del_signature(cls, log_type):
        try:
            while True:
                signature_sht = Feishu.get_values_batch(log_type, 'xigua', '6tZHhs')
                if len(signature_sht) <= 25:
                    break
                else:
                    Feishu.dimension_range(log_type, 'xigua', '6tZHhs', 'ROWS', 26, 26)
            Common.logger(log_type).info('删除signature完成\n')
        except Exception as e:
            Common.logger(log_type).error('del_signature异常:{}', e)


if __name__ == '__main__':
    GetSignature.get_signature('follow')
    # GetSignature.del_signature('follow')
    pass