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":
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