automatic_crawler/application/pipeline/pipeline.py

import hashlib
import re
import sys
import os
import time

from application.common.feishu.feishu_utils import FeishuUtils

sys.path.append(os.getcwd())
from datetime import datetime

from application.common import MysqlHelper, AliyunLogger
# from application.common.redis.pyredis import RedisClient


class PiaoQuanPipeline(object):
    """
    爬虫管道——爬虫规则判断
    """

    def __init__(self, platform, mode, rule_dict, env, item, trace_id, account=None):
        self.platform = platform
        self.mode = mode
        self.item = item
        self.rule_dict = rule_dict
        self.env = env
        self.trace_id = trace_id
        self.mysql = MysqlHelper(env=env, mode=mode, platform=platform)
        self.aliyun_log = AliyunLogger(platform=platform, mode=mode, env=env)
        self.account = account
        # self.red = RedisClient()

    def feishu_time_list(self):
        summary = FeishuUtils.get_values_batch("KsoMsyP2ghleM9tzBfmcEEXBnXg", "RuLK77")
        for row in summary[1:]:
            channel = row[0]
            day_count = row[1]
            if channel:
                if channel == self.platform:
                    return day_count
            else:
                return None
        return None

    def publish_time_flag(self):
        """
        判断发布时间是否过期
        :return: True or False
        """
        # 判断发布时间
        publish_time_stamp = self.item["publish_time_stamp"]
        update_time_stamp = self.item["update_time_stamp"]
        max_d = self.rule_dict.get("period", {}).get("max", 1000)
        min_d = self.rule_dict.get("period", {}).get("min", 1000)
        days = max_d if max_d > min_d else min_d
        days_time = self.feishu_time_list()
        if days_time:
            days = int(days_time)
        if self.platform == "gongzhonghao":
            if (
                    int(time.time()) - publish_time_stamp
                    > 3600 * 24 * days
            ) and (
                    int(time.time()) - update_time_stamp
                    > 3600 * 24 * days
            ):
                self.aliyun_log.logging(
                    code="2004",
                    trace_id=self.trace_id,
                    data=self.item,
                    message="发布时间超过{}天".format(days),
                )
                return False
        else:
            if days == 0:
                publish_time_stamp = int(time.time())  # 示例时间戳
                is_today = datetime.fromtimestamp(publish_time_stamp).date() == datetime.today().date()
                if not is_today:
                    return False

            elif (
                    int(time.time()) - publish_time_stamp
                    > 3600 * 24 * days
            ):
                self.aliyun_log.logging(
                    code="2004",
                    trace_id=self.trace_id,
                    data=self.item,
                    message="发布时间超过{}天".format(days),
                )
                return False
        return True

    def title_flag(self):
        """
        视频标题是否满足需求
        :return:
        """
        title = self.item["video_title"]
        cleaned_title = re.sub(r"[^\w]", " ", title)
        # 敏感词
        # 获取敏感词列表
        sensitive_words = []
        if any(word in cleaned_title for word in sensitive_words):
            self.aliyun_log.logging(
                code="2003",
                trace_id=self.trace_id,
                message="标题中包含敏感词",
                data=self.item,
                account=self.account
            )
            return False
        return True

    def download_rule_flag(self):
        """
        视频基础下载规则
        :return:
        """
        for key in self.item:
            if self.rule_dict.get(key):
                max_value = (
                    int(self.rule_dict[key]["max"])
                    if int(self.rule_dict[key]["max"]) > 0
                    else 999999999999999
                )
                if key == "peroid":  # peroid是抓取周期天数
                    continue
                else:
                    flag = int(self.rule_dict[key]["min"]) <= int(self.item[key]) <= max_value
                    if not flag:
                        self.aliyun_log.logging(
                            code="2004",
                            trace_id=self.trace_id,
                            data=self.item,
                            message="{}: {} <= {} <= {}, {}".format(
                                key,
                                self.rule_dict[key]["min"],
                                self.item[key],
                                max_value,
                                flag,
                            ),
                            account=self.account
                        )
                        return flag
            else:
                continue
        return True

    def feishu_list(self):
        summary = FeishuUtils.get_values_batch("KsoMsyP2ghleM9tzBfmcEEXBnXg", "letS93")
        for row in summary[1:]:
            channel = row[0]
            day_count = row[1]
            if channel:
                if channel == self.platform:
                    return day_count
            else:
                return None
        return None

    # 按照某个具体平台来去重
    def repeat_video(self):
        """
        视频是否重复
        :return:
        """
        out_id = self.item["out_video_id"]
        day_count = self.feishu_list()
        if day_count:
            sql_2 = f"""select create_time from crawler_video where platform = "{self.platform}" and  out_video_id="{out_id}" AND create_time >= DATE_SUB(NOW(), INTERVAL {int(day_count)} DAY);"""
            repeat_video = self.mysql.select(sql=sql_2)
            if repeat_video:
                self.aliyun_log.logging(
                    code="2002",
                    trace_id=self.trace_id,
                    message="重复的视频",
                    data=self.item,
                    account=self.account
                )
                return False
            else:
                return True

        if self.platform == "zhufuniannianshunxinjixiang" or  self.platform == "weiquanshipin" or  self.platform == "piaoquangushi" or  self.platform == "lepaoledong" or  self.platform == "zhufukuaizhuan" or self.platform == "linglingkuailezhufu" or self.platform == "lepaoledongdijie":
            return True
        if self.platform == "jierizhufuhuakaifugui" or self.platform == "yuannifuqimanman" or self.platform == "haoyunzhufuduo" or self.platform == "quzhuan" or self.platform == "zhufudewenhou" or self.platform == "jierizhufuxingfujixiang" or self.platform == "haoyoushipin" or self.platform == "xinshiquan" or self.platform == "laonianshenghuokuaile" or self.platform == "laonianquan":
            return True
        if self.platform == "zhuwanwufusunew" and self.mode == "recommend":
            return True
        if self.platform == "jixiangxingfu" and self.mode == "recommend":
            return True
        if self.platform == "yuannifuqichangzai" and self.mode == "recommend":
            return True
        if self.platform == "benshanzhufu" and self.mode == "recommend":
            return True
        if self.platform == "zuihaodesongni" and self.mode == "recommend":
            return True
        if self.platform == "tiantianjufuqi" and self.mode == "recommend":
            return True
        # 判断加上标题去重
        if self.mode == "recommend" and self.platform == "zhufuhaoyunbaofu":
            title = self.item["video_title"]
            sql = f""" select 1 from crawler_video where platform = "{self.platform}" and out_video_id="{out_id}" and video_title="{title}"; """
        else:
            sql = f""" select 1 from crawler_video where platform = "{self.platform}" and out_video_id="{out_id}"; """
        repeat_video = self.mysql.select(sql=sql)
        if repeat_video:
            # 喜事多多平台 4 天去重一次
            if self.platform == "xishiduoduo":
                sql_2 = f"""select create_time from crawler_video where out_video_id="{out_id}";"""
                video_time = self.mysql.select(sql=sql_2)[0][0].timestamp()
                if int(time.time()) - video_time >= 86400 * 4:
                    return True
            # 小年糕推荐流和祝福圈子推荐流 3 天去重一次
            elif self.platform == "xiaoniangaotuijianliu" or self.platform == "zhufuquanzituijianliu":
                sql_2 = f"""select create_time from crawler_video where out_video_id="{out_id}";"""
                video_time = self.mysql.select(sql=sql_2)[0][0].timestamp()
                if int(time.time()) - video_time >= 86400 * 3:
                    return True
            self.aliyun_log.logging(
                code="2002",
                trace_id=self.trace_id,
                message="重复的视频",
                data=self.item,
                account=self.account
            )
            return False
        return True

    # def mq_exists(self):
    #     """
    #     检测 mq 是否已经发送过了
    #     :return:
    #     """
    #     if self.red.connect():
    #         index_txt = "{}-{}".format(self.platform, self.item['video_id'])
    #         index_md5 = hashlib.md5(index_txt.encode()).hexdigest()
    #         if self.red.select(index_md5):
    #             self.aliyun_log.logging(
    #                 code="2007",
    #                 trace_id=self.trace_id,
    #                 message="该视频 mq 已经发送"
    #             )
    #             return False
    #         else:
    #             self.red.insert(index_md5, int(time.time()), 43200)
    #             return True
    #     else:
    #         return True

    def process_item(self):
        """
        全规则判断，符合规则的数据则return True
        :return:
        """
        # 判断该 mq 是否已经发了
        # if not self.mq_exists():
        #     return False
        if not self.publish_time_flag():
            # 记录相关日志
            return False
        if not self.title_flag():
            # 记录相关日志
            return False
        if not self.repeat_video():
            # 记录相关日志
            return False
        if not self.download_rule_flag():
            # 记录相关日志
            return False
        return True