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.chrome.service import Service
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')
            chrome_options.add_argument("--no-sandbox")

            # driver初始化
            Common.logger(log_type).info('初始化 webdriver')
            # driver = webdriver.Chrome(desired_capabilities=ca, options=chrome_options)
            # driver = webdriver.Chrome(desired_capabilities=ca, options=chrome_options, service=Service('/Users/wangkun/Downloads/chromedriver_v107/chromedriver'))
            # driver = webdriver.Chrome(desired_capabilities=ca, options=chrome_options, service=Service('/Users/lieyunye/Downloads/chromedriver_v107/chromedriver'))
            driver = webdriver.Chrome(desired_capabilities=ca, options=chrome_options, service=Service('/Users/piaoquan/Downloads/chromedriver_v108/chromedriver'))

            driver.implicitly_wait(10)
            Common.logger(log_type).info('打开个人主页')
            driver.get('https://www.ixigua.com/home/100909070373/')
            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/100909070373' 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(f'url:{url}\n')
                    Common.logger(log_type).info(f'signature:{signature}\n')

                    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(f'signature写入飞书成功:{signature}\n')
                    cls.del_signature(log_type)
        except Exception as e:
            Common.logger(log_type).error(f'get_signature异常:{e}\n')

    @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(f'del_signature异常:{e}\n')


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