crawler_shipinhao/main/common.py

# -*- coding: utf-8 -*-
# @Author: wangkun
# @Time: 2022/8/30
"""
公共方法，包含：生成log / 删除log / 下载方法 / 读取文件 / 统计下载数
"""
from datetime import date, timedelta
from loguru import logger
import datetime
import os
import time
import requests
import urllib3
proxies = {"http": None, "https": None}


class Common:
    # 统一获取当前时间 <class 'datetime.datetime'>  2022-04-14 20:13:51.244472
    now = datetime.datetime.now()
    # 昨天 <class 'str'>  2022-04-13
    yesterday = (date.today() + timedelta(days=-1)).strftime("%Y-%m-%d")
    # 今天 <class 'datetime.date'>  2022-04-14
    today = date.today()
    # 明天 <class 'str'>  2022-04-15
    tomorrow = (date.today() + timedelta(days=1)).strftime("%Y-%m-%d")

    # 使用 logger 模块生成日志
    @staticmethod
    def logger(log_type):
        """
        使用 logger 模块生成日志
        """
        # 日志路径
        log_dir = r"./logs/"
        # log_dir = r"./logs/"
        log_path = os.getcwd() + os.sep + log_dir
        if not os.path.isdir(log_path):
            os.makedirs(log_path)

        # 日志文件名
        log_name = time.strftime("%Y-%m-%d", time.localtime(time.time())) + '-shipinhao-'+str(log_type)+'.log'

        # 日志不打印到控制台
        logger.remove(handler_id=None)

        # rotation="500 MB"，实现每 500MB 存储一个文件
        # rotation="12:00"，实现每天 12:00 创建一个文件
        # rotation="1 week"，每周创建一个文件
        # retention="10 days"，每隔10天之后就会清理旧的日志
        # 初始化日志
        logger.add(log_dir + log_name, level="INFO", rotation='00:00')

        return logger

    # 清除日志，保留最近 6 个文件
    @classmethod
    def del_logs(cls, log_type):
        """
        清除冗余日志文件
        :d_dir: 需要删除的 log 地址
        :return: 保留最近 6 个日志
        """
        logs_dir = r"./logs/"
        # logs_dir = r"./logs/"
        if not os.path.exists(logs_dir):
            os.mkdir(logs_dir)

        all_files = sorted(os.listdir(logs_dir))
        all_logs = []
        for log in all_files:
            name = os.path.splitext(log)[-1]
            if name == ".log":
                all_logs.append(log)

        if len(all_logs) <= 6:
            pass
        else:
            for file in all_logs[:len(all_logs) - 6]:
                os.remove(logs_dir + file)
        cls.logger(log_type).info("清除冗余日志成功\n")

    @classmethod
    def del_charles_files(cls, log_type):
        # 目标文件夹下所有文件
        all_file = sorted(os.listdir("./chlsfiles/"))
        # for file in all_file[0:-1]:
        for file in all_file:
            os.remove(r"./chlsfiles/" + file)
        cls.logger(log_type).info("删除 charles 缓存文件成功\n")

    # 封装下载视频或封面的方法
    @classmethod
    def download_method(cls, log_type, text, d_name, d_url):
        """
        下载封面：text == "cover" ； 下载视频：text == "video"
        需要下载的视频标题：d_title
        视频封面，或视频播放地址：d_url
        下载保存路径："./files/{d_title}/"
        """
        videos_dir = r"./videos/"
        # videos_dir = r"./videos/"
        if not os.path.exists(videos_dir):
            os.mkdir(videos_dir)
        # 首先创建一个保存该视频相关信息的文件夹
        video_dir = r"./videos/" + d_name + r"/"
        # video_dir = r"./videos/" + d_name + "/"
        if not os.path.exists(video_dir):
            os.mkdir(video_dir)

        # 下载视频
        if text == "video":
            # 需要下载的视频地址
            video_url = d_url
            # 视频名
            video_name = "video.mp4"

            # 下载视频
            urllib3.disable_warnings()
            response = requests.get(video_url, stream=True, proxies=proxies, verify=False)
            try:
                with open(video_dir + video_name, "wb") as f:
                    for chunk in response.iter_content(chunk_size=10240):
                        f.write(chunk)
                cls.logger(log_type).info("==========视频下载完成==========")
            except Exception as e:
                cls.logger(log_type).exception("视频下载失败：{}\n", e)

        # 下载封面
        elif text == "cover":
            # 需要下载的封面地址
            cover_url = d_url
            # 封面名
            cover_name = "image.jpg"
            # # 封面名
            # cover_name = d_name + ".jpg"

            # 下载封面
            urllib3.disable_warnings()
            response = requests.get(cover_url, proxies=proxies, verify=False)
            try:
                with open(video_dir + cover_name, "wb") as f:
                    f.write(response.content)
                cls.logger(log_type).info("==========封面下载完成==========")
            except Exception as e:
                cls.logger(log_type).exception("封面下载失败：{}\n", e)


if __name__ == "__main__":
    # url = 'https://finder.video.qq.com/251/20302/stodownload?encfilekey=6xykWLEnztKcKCJZcV0rWCM8ua7DibZkibqXGfPxf5lrq2MRcDiaJzbKp1HZLPibvcic4UPgLTV6SLChiaTDuctV3zOPo7yfMXeAtIB3QBxiasjqxDTvNeKDCyuDzX9iatjib1icSibQIeT7MiaNk6vs2hALzzialobvJXicllb71OEwmHQXXSiasI&token=x5Y29zUxcibAMWmUVOhVkYE4rmSj8FKfeN86Zqyq5OqjxITFTBsNEhTkjsywzcZibrAl4IqV6GicXE&idx=1&a=1&adaptivelytrans=0&bizid=1023&dotrans=3071&hy=SH&m=83157388b45cf2ceced9b8dd0b13f305&finderextscene=6&web=1&X-snsvideoflag=xV1&taskid=0'
    url = 'https://finder.video.qq.com/251/20302/stodownload?encfilekey=rjD5jyTuFrIpZ2ibE8T7YmwgiahniaXswqzXw1KgS84gF4JiaL7JFsb77RSNIaKzdG6py9J7FJGMNmpOGpkGD2FYmAgrVpOcInBdr6LGKfOz3eW2mDCeM2UCRg&token=x5Y29zUxcibAMWmUVOhVkYHa5BcMzcgZQldAN5L8BgQ529PYwRMnvibDO1hqqoXfmbIO0cGoeCAPk&idx=1&adaptivelytrans=0&bizid=1023&dotrans=3071&hy=SH&m=&finderextscene=6&web=1&X-snsvideoflag=xV6&taskid=0'
    Common.download_method('recommend', 'video', 'video', url)