Merge remote-tracking branch 'origin/master'

application/common/messageQueue/mq.py

@@ -16,6 +16,7 @@ class MQ(object):
         self.mq_client = MQClient("http://1894469520484605.mqrest.cn-qingdao-public.aliyuncs.com",
                                   "LTAI4G7puhXtLyHzHQpD6H7A",
                                   "nEbq3xWNQd1qLpdy2u71qFweHkZjSG")
+        topic_name = topic_name+"_v2"
         self.producer = self.mq_client.get_producer(self.instance_id, topic_name)
 
     def send_msg(self, video_dict):

application/common/redis/pyredis.py

@@ -14,7 +14,8 @@ class RedisClient(object):
 
     def __init__(self):
         self.pool = None
-        self.host = 'r-bp1mb0v08fqi4hjffu.redis.rds.aliyuncs.com'
+        # self.host = 'r-bp1mb0v08fqi4hjffu.redis.rds.aliyuncs.com'
+        self.host="r-bp1mb0v08fqi4hjffupd.redis.rds.aliyuncs.com",  # 外网地址
         self.port = 6379
         self.db = 2
         self.password = 'Wqsd@2019'

application/config/topic_group_queue.py

@@ -18,6 +18,9 @@ class TopicGroup(object):
             ('zhdsn', 'recommend', 'zuihaodesongni'),
             ('jrzf', 'recommend', 'jierizhufu'),
             ('dkyhy', 'recommend', 'dakaiyinghaoyun'),
+            ('xngtjl', 'recommend', 'xiaoniangaotuijianliu'),
+            ('zfqztjl', 'recommend', 'zhufuquanzituijianliu'),
+            ('bszf', 'recommend', 'benshanzhufu'),
             ('ttjfq', 'recommend', 'tiantianjufuqi')
         ]
 

application/pipeline/pipeline.py

@@ -7,7 +7,7 @@ import time
 sys.path.append(os.getcwd())
 
 from application.common import MysqlHelper, AliyunLogger
-from application.common.redis.pyredis import RedisClient
+# from application.common.redis.pyredis import RedisClient
 
 
 class PiaoQuanPipeline(object):
@@ -25,7 +25,7 @@ class PiaoQuanPipeline(object):
         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()
+        # self.red = RedisClient()
 
     def publish_time_flag(self):
         """
@@ -135,6 +135,8 @@ class PiaoQuanPipeline(object):
             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":
@@ -154,6 +156,12 @@ class PiaoQuanPipeline(object):
                 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,
@@ -164,26 +172,26 @@ class PiaoQuanPipeline(object):
             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 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):
         """
@@ -191,8 +199,8 @@ class PiaoQuanPipeline(object):
         :return:
         """
         # 判断该 mq 是否已经发了
-        if not self.mq_exists():
-            return False
+        # if not self.mq_exists():
+        #     return False
         if not self.publish_time_flag():
             # 记录相关日志
             return False

spider/crawler_online/benshanzhufu.py

@@ -0,0 +1,171 @@
+import os
+import random
+import sys
+import time
+import uuid
+import json
+
+from datetime import datetime
+
+import cv2
+import requests
+
+from application.common import Feishu
+
+sys.path.append(os.getcwd())
+
+from application.items import VideoItem
+from application.pipeline import PiaoQuanPipeline
+from application.common.messageQueue import MQ
+from application.common.log import AliyunLogger
+from application.common.mysql import MysqlHelper
+
+
+
+class BSZHRecommend(object):
+
+    """
+    本山祝福推荐流
+    """
+
+    def __init__(self, platform, mode, rule_dict, user_list, env="prod"):
+        self.limit_flag = False
+        self.platform = platform
+        self.mode = mode
+        self.rule_dict = rule_dict
+        self.user_list = user_list
+        self.env = env
+        self.download_cnt = 0
+        self.mq = MQ(topic_name="topic_crawler_etl_" + self.env)
+        self.expire_flag = False
+        self.aliyun_log = AliyunLogger(mode=self.mode, platform=self.platform)
+        self.mysql = MysqlHelper(mode=self.mode, platform=self)
+
+    def get_video_duration(self, video_link: str) -> int:
+        cap = cv2.VideoCapture(video_link)
+        if cap.isOpened():
+            rate = cap.get(5)
+            frame_num = cap.get(7)
+            duration = int(frame_num / rate)
+            return duration
+        return 0
+
+    def get_recommend_list(self):
+        print("本山祝福开始")
+
+        """
+        获取推荐页视频
+        """
+        url = "http://47.236.68.175:8889/crawler/ben_shan_zhu_fu/recommend"
+        next_cursor = 1
+        for i in range(1, 200):
+            payload = json.dumps({
+                "cursor": f"{next_cursor}"
+            })
+            headers = {
+                'Content-Type': 'application/json'
+            }
+            for j in range(3):
+                response = requests.request("POST", url, headers=headers, data=payload)
+                response = response.json()
+                if response['code'] != 0:
+                    time.sleep(2)
+                    continue
+                else:
+                    break
+            if response['code'] != 0:
+                self.aliyun_log.logging(
+                    code="3000",
+                    message="抓取单条视频失败,请求失败"
+                ),
+                return
+            for index, video_obj in enumerate(response['data']['data'], i):
+                try:
+                    self.aliyun_log.logging(
+                        code="1001", message="扫描到一条视频", data=video_obj
+                    )
+                    next_cursor = response['data']['next_cursor']
+                    self.process_video_obj(video_obj)
+                except Exception as e:
+                    self.aliyun_log.logging(
+                        code="3000",
+                        message="抓取单条视频失败, 该视频位于第{}页第{}条报错原因是{}".format(
+                            i, index, e
+                        ),
+                    )
+                if self.limit_flag:
+                    return
+                time.sleep(random.randint(5, 10))
+
+    def process_video_obj(self, video_obj):
+        """
+        处理视频
+        :param video_obj:
+        """
+        time.sleep(random.randint(3, 8))
+        trace_id = self.platform + str(uuid.uuid1())
+        our_user = random.choice(self.user_list)
+        item = VideoItem()
+        # id = uuid.uuid4()
+        item.add_video_info("video_id", video_obj["nid"])
+        item.add_video_info("video_title", video_obj["title"])
+        item.add_video_info("play_cnt", 0)
+        item.add_video_info("publish_time_stamp", int(video_obj["update_time"]))
+        item.add_video_info("out_user_id", video_obj["nid"])
+        item.add_video_info("cover_url", video_obj["video_cover"])
+        item.add_video_info("like_cnt", 0)
+        item.add_video_info("video_url", video_obj["video_url"])
+        item.add_video_info("out_video_id", video_obj["nid"])
+        item.add_video_info("platform", self.platform)
+        item.add_video_info("strategy", self.mode)
+        item.add_video_info("session", "{}-{}".format(self.platform, int(time.time())))
+        item.add_video_info("user_id", our_user["uid"])
+        item.add_video_info("user_name", our_user["nick_name"])
+        mq_obj = item.produce_item()
+        pipeline = PiaoQuanPipeline(
+            platform=self.platform,
+            mode=self.mode,
+            rule_dict=self.rule_dict,
+            env=self.env,
+            item=mq_obj,
+            trace_id=trace_id,
+        )
+        if pipeline.process_item():
+            self.download_cnt += 1
+            self.mq.send_msg(mq_obj)
+            self.aliyun_log.logging(code="1002", message="成功发送至 ETL", data=mq_obj)
+            if self.download_cnt >= int(
+                    self.rule_dict.get("videos_cnt", {}).get("min", 200)
+            ):
+                self.limit_flag = True
+
+
+
+    def run(self):
+        self.get_recommend_list()
+
+
+if __name__ == '__main__':
+    J = BSZHRecommend(
+        platform="benshanzhufu",
+        mode="recommend",
+        rule_dict={},
+        user_list=[{
+    "createTime": 1684311893899,
+    "id": 8,
+    "interval": 7200,
+    "machine": "aliyun",
+    "mode": "recommend",
+    "operator": "王雪珂",
+    "rule": "[{\"like_cnt\":{\"min\":0,\"max\":0}}]",
+    "source": "benshanzhufu",
+    "spiderName": "run_bszf_recommend",
+    "startTime": 1730452800000,
+    "status": 0,
+    "taskName": "本山祝福",
+    "updateTime": 1730452617817
+}],
+
+    )
+    J.get_recommend_list()
+    # J.logic()

spider/crawler_online/xiaoniangaotuijianliu.py

@@ -0,0 +1,151 @@
+import os
+import random
+import sys
+import time
+import uuid
+import json
+
+from datetime import datetime
+
+import cv2
+import requests
+
+from application.common import Feishu
+
+sys.path.append(os.getcwd())
+
+from application.items import VideoItem
+from application.pipeline import PiaoQuanPipeline
+from application.common.messageQueue import MQ
+from application.common.log import AliyunLogger
+from application.common.mysql import MysqlHelper
+
+
+
+class XNGTJLRecommend(object):
+
+    """
+    小年糕推荐流
+    """
+
+    def __init__(self, platform, mode, rule_dict, user_list, env="prod"):
+        self.limit_flag = False
+        self.platform = platform
+        self.mode = mode
+        self.rule_dict = rule_dict
+        self.user_list = user_list
+        self.env = env
+        self.download_cnt = 0
+        self.mq = MQ(topic_name="topic_crawler_etl_" + self.env)
+        self.expire_flag = False
+        self.aliyun_log = AliyunLogger(mode=self.mode, platform=self.platform)
+        self.mysql = MysqlHelper(mode=self.mode, platform=self)
+
+    def get_video_duration(self, video_link: str) -> int:
+        cap = cv2.VideoCapture(video_link)
+        if cap.isOpened():
+            rate = cap.get(5)
+            frame_num = cap.get(7)
+            duration = int(frame_num / rate)
+            return duration
+        return 0
+
+    def get_recommend_list(self):
+        print("小年糕推荐流开始")
+
+        """
+        获取推荐页视频
+        """
+        headers = {
+            'Content-Type': 'application/json'
+        }
+        for i in range(3):
+            url = "http://47.236.68.175:8889/crawler/xiao_nian_gao_plus/recommend"
+            payload = json.dumps({})
+            response = requests.request("POST", url, headers=headers, data=payload)
+            response = response.json()
+            if response['code'] != 0:
+                self.aliyun_log.logging(
+                    code="3000",
+                    message="抓取单条视频失败,请求失败"
+                ),
+                return
+            for index, video_obj in enumerate(response['data']['data'], 1):
+                try:
+                    self.aliyun_log.logging(
+                        code="1001", message="扫描到一条视频", data=video_obj
+                    )
+                    self.process_video_obj(video_obj)
+                except Exception as e:
+                    self.aliyun_log.logging(
+                        code="3000",
+                        message="抓取单条视频失败, 该视频位于第{}页第{}条报错原因是{}".format(
+                            1, index, e
+                        ),
+                    )
+                if self.limit_flag:
+                    return
+                time.sleep(random.randint(5, 10))
+
+    def process_video_obj(self, video_obj):
+        """
+        处理视频
+        :param video_obj:
+        """
+        time.sleep(random.randint(3, 8))
+        trace_id = self.platform + str(uuid.uuid1())
+        our_user = random.choice(self.user_list)
+        item = VideoItem()
+        url = video_obj["v_url"]
+        duration = self.get_video_duration(url)
+        item.add_video_info("video_id", video_obj["id"])
+        item.add_video_info("video_title", video_obj["title"])
+        item.add_video_info("play_cnt", int(video_obj["play_pv"]))
+        item.add_video_info("publish_time_stamp", int(int(video_obj["t"]) / 1000))
+        item.add_video_info("out_user_id", video_obj["id"])
+        item.add_video_info("cover_url", video_obj["url"])
+        item.add_video_info("like_cnt", 0)
+        item.add_video_info("share_cnt", int(video_obj["share"]))
+        item.add_video_info("comment_cnt", int(video_obj["comment_count"]))
+        item.add_video_info("video_url", video_obj["v_url"])
+        item.add_video_info("out_video_id", video_obj["id"])
+        item.add_video_info("duration", int(duration))
+        item.add_video_info("platform", self.platform)
+        item.add_video_info("strategy", self.mode)
+        item.add_video_info("session", "{}-{}".format(self.platform, int(time.time())))
+        item.add_video_info("user_id", our_user["uid"])
+        item.add_video_info("user_name", our_user["nick_name"])
+        mq_obj = item.produce_item()
+        pipeline = PiaoQuanPipeline(
+            platform=self.platform,
+            mode=self.mode,
+            rule_dict=self.rule_dict,
+            env=self.env,
+            item=mq_obj,
+            trace_id=trace_id,
+        )
+        if pipeline.process_item():
+            self.download_cnt += 1
+            self.mq.send_msg(mq_obj)
+            self.aliyun_log.logging(code="1002", message="成功发送至 ETL", data=mq_obj)
+            if self.download_cnt >= int(
+                    self.rule_dict.get("videos_cnt", {}).get("min", 200)
+            ):
+                self.limit_flag = True
+
+
+
+    def run(self):
+        self.get_recommend_list()
+
+
+if __name__ == '__main__':
+    J = XNGTJLRecommend(
+        platform="xiaonianggaotuijianliu",
+        mode="recommend",
+        rule_dict={},
+        user_list=[{'uid': "123456", 'nick_name': "xiaoxiao"}],
+
+    )
+    J.get_recommend_list()
+    # J.logic()

spider/crawler_online/zhufuquanzituijianliu.py

@@ -0,0 +1,152 @@
+import os
+import random
+import sys
+import time
+import uuid
+import json
+
+from datetime import datetime
+
+import cv2
+import requests
+
+from application.common import Feishu
+
+sys.path.append(os.getcwd())
+
+from application.items import VideoItem
+from application.pipeline import PiaoQuanPipeline
+from application.common.messageQueue import MQ
+from application.common.log import AliyunLogger
+from application.common.mysql import MysqlHelper
+
+
+
+class ZFQZTJLRecommend(object):
+    """
+    祝福圈子推荐流
+    """
+
+    def __init__(self, platform, mode, rule_dict, user_list, env="prod"):
+        self.limit_flag = False
+        self.platform = platform
+        self.mode = mode
+        self.rule_dict = rule_dict
+        self.user_list = user_list
+        self.env = env
+        self.download_cnt = 0
+        self.mq = MQ(topic_name="topic_crawler_etl_" + self.env)
+        self.expire_flag = False
+        self.aliyun_log = AliyunLogger(mode=self.mode, platform=self.platform)
+        self.mysql = MysqlHelper(mode=self.mode, platform=self)
+
+    def get_video_duration(self, video_link: str) -> int:
+        cap = cv2.VideoCapture(video_link)
+        if cap.isOpened():
+            rate = cap.get(5)
+            frame_num = cap.get(7)
+            duration = int(frame_num / rate)
+            return duration
+        return 0
+
+    def get_recommend_list(self):
+        print("祝福圈子推荐流开始")
+        """
+        获取推荐页视频
+        """
+
+        headers = {
+            'Content-Type': 'application/json'
+        }
+        for i in range(2):
+            url = "http://47.236.68.175:8889/crawler/zhu_fu_quan_zi/recommend"
+            payload = json.dumps({})
+            response = requests.request("POST", url, headers=headers, data=payload)
+            response = response.json()
+            if  response['code'] != 0:
+                self.aliyun_log.logging(
+                    code="3000",
+                    message="抓取单条视频失败,请求失败"
+                    ),
+                return
+            for index, video_obj in enumerate(response['data']['data'], 1):
+                try:
+                    self.aliyun_log.logging(
+                        code="1001", message="扫描到一条视频", data=video_obj
+                    )
+                    self.process_video_obj(video_obj)
+                except Exception as e:
+                    self.aliyun_log.logging(
+                        code="3000",
+                        message="抓取单条视频失败, 该视频位于第{}页第{}条报错原因是{}".format(
+                            1, index, e
+                        ),
+                    )
+                if self.limit_flag:
+                    return
+                time.sleep(random.randint(5, 10))
+
+    def process_video_obj(self, video_obj):
+        """
+        处理视频
+        :param video_obj:
+        """
+        time.sleep(random.randint(3, 8))
+        trace_id = self.platform + str(uuid.uuid1())
+        our_user = random.choice(self.user_list)
+        item = VideoItem()
+        url =  video_obj["v_url"]
+        duration = self.get_video_duration(url)
+        item.add_video_info("video_id", video_obj["id"])
+        item.add_video_info("video_title", video_obj["title"])
+        item.add_video_info("play_cnt", int(video_obj["play_pv"]))
+        item.add_video_info("publish_time_stamp", int(int(video_obj["t"])/1000))
+        item.add_video_info("out_user_id", video_obj["id"])
+        item.add_video_info("cover_url", video_obj["url"])
+        item.add_video_info("like_cnt", 0)
+        item.add_video_info("share_cnt", int(video_obj["share"]))
+        item.add_video_info("comment_cnt", int(video_obj["comment_count"]))
+        item.add_video_info("video_url", video_obj["v_url"])
+        item.add_video_info("out_video_id", video_obj["id"])
+        item.add_video_info("duration", int(duration))
+        item.add_video_info("platform", self.platform)
+        item.add_video_info("strategy", self.mode)
+        item.add_video_info("session", "{}-{}".format(self.platform, int(time.time())))
+        item.add_video_info("user_id", our_user["uid"])
+        item.add_video_info("user_name", our_user["nick_name"])
+
+        mq_obj = item.produce_item()
+        pipeline = PiaoQuanPipeline(
+            platform=self.platform,
+            mode=self.mode,
+            rule_dict=self.rule_dict,
+            env=self.env,
+            item=mq_obj,
+            trace_id=trace_id,
+        )
+        if pipeline.process_item():
+            self.download_cnt += 1
+            self.mq.send_msg(mq_obj)
+            self.aliyun_log.logging(code="1002", message="成功发送至 ETL", data=mq_obj)
+            if self.download_cnt >= int(
+                    self.rule_dict.get("videos_cnt", {}).get("min", 200)
+            ):
+                self.limit_flag = True
+
+
+
+
+    def run(self):
+        self.get_recommend_list()
+
+
+if __name__ == '__main__':
+    J = ZFQZTJLRecommend(
+        platform="zhufuquanzituijianliu",
+        mode="recommend",
+        rule_dict={},
+        user_list=[{"uid": 75590470, "link": "zfqz推荐流_接口1", "nick_name": "做你的尾巴"}, {"uid": 75590471, "link": "zfqz推荐流_接口2", "nick_name": "能够相遇"}, {"uid": 75590472, "link": "zfqz推荐流_接口3", "nick_name": "一别两宽各生欢喜"}, {"uid": 75590473, "link": "zfqz推荐流_接口4", "nick_name": "惹火"}, {"uid": 75590475, "link": "zfqz推荐流_接口5", "nick_name": "顾九"}, {"uid": 75590476, "link": "zfqz推荐流_接口6", "nick_name": "宠一身脾气惯一身毛病"}],
+
+    )
+    J.get_recommend_list()
+    # J.logic()

spider/spider_map.py

@@ -5,7 +5,10 @@ sub_key是recommend，author, value；
 value是爬虫封装好的类
 """
 from spider.crawler_online import *
+from spider.crawler_online.benshanzhufu import BSZHRecommend
 from spider.crawler_online.dakaiyinghaoyun import DKYHYRecommend
+from spider.crawler_online.xiaoniangaotuijianliu import XNGTJLRecommend
+from spider.crawler_online.zhufuquanzituijianliu import ZFQZTJLRecommend
 
 spider_map = {
     # 祝万物复苏
@@ -67,6 +70,18 @@ spider_map = {
     # 福气矩阵-天天聚福气
     "tiantianjufuqi": {
         "recommend": TTJFFQRecommend
+    },
+    # 小年糕推荐流
+    "xiaoniangaotuijianliu": {
+        "recommend": XNGTJLRecommend
+    },
+    # 本山祝福
+    "benshanzhufu": {
+        "recommend": BSZHRecommend
+    },
+    # 祝福圈子推荐流
+    "zhufuquanzituijianliu": {
+        "recommend": ZFQZTJLRecommend
     }
 
 }