增加新小程序

application/config/topic_group_queue.py

@@ -22,7 +22,11 @@ class TopicGroup(object):
             ('zfqztjl', 'recommend', 'zhufuquanzituijianliu'),
             ('bszf', 'recommend', 'benshanzhufu'),
             ('zfnnsxjx', 'recommend', 'zhufuniannianshunxinjixiang'),
-            ('ttjfq', 'recommend', 'tiantianjufuqi')
+            ('ttjfq', 'recommend', 'tiantianjufuqi'),
+            ('zfkz', 'recommend', 'zhufukuaizhuan'),
+            ('lpld', 'recommend', 'lepaoledong'),
+            ('pqgs', 'recommend', 'piaoquangushi'),
+            ('wqsp', 'recommend', 'weiquanshipin'),
         ]
 
     def produce(self):

application/pipeline/pipeline.py

@@ -129,7 +129,7 @@ class PiaoQuanPipeline(object):
         视频是否重复
         :return:
         """
-        if self.platform == "zhufuniannianshunxinjixiang" and self.mode == "recommend":
+        if self.platform == "zhufuniannianshunxinjixiang" or  self.platform == "weiquanshipin" or  self.platform == "piaoquangushi" or  self.platform == "lepaoledong" or  self.platform == "zhufukuaizhuan":
             return True
         if self.platform == "zhuwanwufusunew" and self.mode == "recommend":
             return True

spider/crawler_online/lepaoledong.py

@@ -0,0 +1,143 @@
+import os
+import random
+import sys
+import time
+import uuid
+import json
+
+
+import cv2
+import requests
+
+from application.common.mysql.sql import Sql
+from application.common.redis.xng_redis import xng_in_video_data
+
+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 LPLDRecommend(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_recommend_list(self):
+        print("乐跑乐动开始")
+
+        """
+        获取推荐页视频
+        """
+        headers = {
+            'Content-Type': 'application/json'
+        }
+        url = "http://8.217.192.46:8889/crawler/le_pao_le_dong/recommend"
+        while True:
+            payload = json.dumps({
+                "cursor": ""
+            })
+            response = requests.request("POST", url, headers=headers, data=payload)
+            response = response.json()
+            if response['code'] != 0:
+                self.aliyun_log.logging(
+                    code="3000",
+                    message="抓取单条视频失败,请求失败"
+                ),
+                return
+            data = response['data']['data']
+            if len(data) == 0:
+                return
+            for index, video_obj in enumerate(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(1, 5))
+
+    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()
+        item.add_video_info("video_id", video_obj["uuid"])
+        item.add_video_info("video_title", video_obj["title"])
+        item.add_video_info("play_cnt", 0)
+        item.add_video_info("publish_time_stamp", int(time.time()))
+        item.add_video_info("out_user_id", video_obj["uuid"])
+        item.add_video_info("cover_url", video_obj["cover_url"])
+        item.add_video_info("like_cnt", 0)
+        item.add_video_info("share_cnt", 0)
+        item.add_video_info("comment_cnt", 0)
+        item.add_video_info("video_url", video_obj["urls"][0])
+        item.add_video_info("out_video_id", video_obj["uuid"])
+        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 = LPLDRecommend(
+        platform="lepaoledong",
+        mode="recommend",
+        rule_dict={},
+        user_list=[{'uid': "123456", 'nick_name': "xiaoxiao"}],
+
+    )
+    J.get_recommend_list()
+    # J.logic()

spider/crawler_online/piaoquangushi.py

@@ -0,0 +1,172 @@
+import os
+import random
+import sys
+import time
+import uuid
+import json
+
+
+import cv2
+import requests
+
+from application.common.mysql.sql import Sql
+from application.common.redis.xng_redis import xng_in_video_data
+
+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 PQGSRecommend(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_recommend_list(self):
+        print("篻圈故事")
+
+        """
+        获取推荐页视频
+        """
+        headers = {
+            'Content-Type': 'application/json'
+        }
+        cursor  = ""
+        url = "http://8.217.192.46:8889/crawler/xi_que_bo/recommend"
+
+        while True:
+            payload = json.dumps({
+                "cursor": cursor
+            })
+            response = requests.request("POST", url, headers=headers, data=payload)
+            response = response.json()
+            if response['code'] != 0:
+                self.aliyun_log.logging(
+                    code="3000",
+                    message="抓取单条视频失败,请求失败"
+                ),
+                return
+            cursor = response['data']["next_cursor"]
+            data = response['data']['data']
+            if len(data) == 0:
+                return
+            for index, video_obj in enumerate(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(1, 5))
+
+    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)
+        video_url = self.get_video_url(video_obj["id"])
+        if video_url:
+            item = VideoItem()
+            item.add_video_info("video_id", video_obj["id"])
+            item.add_video_info("video_title", video_obj["title"])
+            item.add_video_info("play_cnt", 0)
+            item.add_video_info("publish_time_stamp", int(time.time()))
+            item.add_video_info("out_user_id", video_obj["id"])
+            item.add_video_info("cover_url", video_obj["cover"])
+            item.add_video_info("like_cnt", 0)
+            item.add_video_info("share_cnt", 0)
+            item.add_video_info("comment_cnt", 0)
+            item.add_video_info("video_url", video_url)
+            item.add_video_info("out_video_id", video_obj["id"])
+            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 get_video_url(self, vid):
+        url = "http://8.217.192.46:8889/crawler/piao_quan_gu_shi/detail"
+
+        payload = json.dumps({
+            "content_id": f"{vid}"
+        })
+        headers = {
+            'Content-Type': 'application/json'
+        }
+        try:
+            response = requests.request("POST", url, headers=headers, data=payload)
+            response = response.json()
+            if response['code'] != 0:
+                self.aliyun_log.logging(
+                    code="3000",
+                    message="获取视频链接失败"
+                ),
+                return None
+            video_url = response['data']['data']['video_url_list'][0]['video_url']
+            return video_url
+        except Exception as e:
+            return None
+
+    def run(self):
+        self.get_recommend_list()
+
+
+if __name__ == '__main__':
+    J = PQGSRecommend(
+        platform="piaoquangushi",
+        mode="recommend",
+        rule_dict={},
+        user_list=[{'uid': "123456", 'nick_name': "xiaoxiao"}],
+
+    )
+    J.get_recommend_list()
+    # J.logic()

spider/crawler_online/weiquanshipin.py

@@ -0,0 +1,171 @@
+import os
+import random
+import sys
+import time
+import uuid
+import json
+
+
+import cv2
+import requests
+
+from application.common.mysql.sql import Sql
+from application.common.redis.xng_redis import xng_in_video_data
+
+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 WQSPRecommend(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_recommend_list(self):
+        print("微圈视频")
+
+        """
+        获取推荐页视频
+        """
+        headers = {
+            'Content-Type': 'application/json'
+        }
+        cursor  = ""
+        url = "http://8.217.192.46:8889/crawler/wei_quan_shi_pin/recommend"
+        while True:
+            payload = json.dumps({
+                "cursor": cursor
+            })
+            response = requests.request("POST", url, headers=headers, data=payload)
+            response = response.json()
+            if response['code'] != 0:
+                self.aliyun_log.logging(
+                    code="3000",
+                    message="抓取单条视频失败,请求失败"
+                ),
+                return
+            cursor = response['data']["next_cursor"]
+            data = response['data']['data']
+            if len(data) == 0:
+                return
+            for index, video_obj in enumerate(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(1, 5))
+
+    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)
+        video_url = self.get_video_url(video_obj["vid"])
+        if video_url:
+            item = VideoItem()
+            item.add_video_info("video_id", video_obj["vid"])
+            item.add_video_info("video_title", video_obj["vtitle"])
+            item.add_video_info("play_cnt", 0)
+            item.add_video_info("publish_time_stamp", int(time.time()))
+            item.add_video_info("out_user_id", video_obj["vid"])
+            item.add_video_info("cover_url", video_obj["poster"])
+            item.add_video_info("like_cnt", 0)
+            item.add_video_info("share_cnt", 0)
+            item.add_video_info("comment_cnt", 0)
+            item.add_video_info("video_url", video_url)
+            item.add_video_info("out_video_id", video_obj["vid"])
+            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 get_video_url(self, vid):
+        url = "http://8.217.192.46:8889/crawler/wei_quan_shi_pin/detail"
+
+        payload = json.dumps({
+            "content_id": f"{vid}"
+        })
+        headers = {
+            'Content-Type': 'application/json'
+        }
+        try:
+            response = requests.request("POST", url, headers=headers, data=payload)
+            response = response.json()
+            if response['code'] != 0:
+                self.aliyun_log.logging(
+                    code="3000",
+                    message="获取视频链接失败"
+                ),
+                return None
+            video_url = response['data']['data']['video_url_list'][0]['video_url']
+            return video_url
+        except Exception as e:
+            return None
+
+    def run(self):
+        self.get_recommend_list()
+
+
+if __name__ == '__main__':
+    J = WQSPRecommend(
+        platform="weiquanshipin",
+        mode="recommend",
+        rule_dict={},
+        user_list=[{'uid': "123456", 'nick_name': "xiaoxiao"}],
+
+    )
+    J.get_recommend_list()
+    # J.logic()

spider/crawler_online/zhufukuaizhuan.py

@@ -0,0 +1,173 @@
+import os
+import random
+import sys
+import time
+import uuid
+import json
+
+
+import cv2
+import requests
+
+from application.common.mysql.sql import Sql
+from application.common.redis.xng_redis import xng_in_video_data
+
+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 ZFKZRecommend(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_recommend_list(self):
+        print("祝福快转")
+
+        """
+        祝福快转
+        """
+        headers = {
+            'Content-Type': 'application/json'
+        }
+        cursor  = ""
+        url = "http://8.217.192.46:8889/crawler/zhu_fu_kuai_zhuan/recommend"
+
+
+        while True:
+            payload = json.dumps({
+                "cursor": cursor
+            })
+            response = requests.request("POST", url, headers=headers, data=payload)
+            response = response.json()
+            if response['code'] != 0:
+                self.aliyun_log.logging(
+                    code="3000",
+                    message="抓取单条视频失败,请求失败"
+                ),
+                return
+            cursor = response['data']["next_cursor"]
+            data = response['data']['data']
+            if len(data) == 0:
+                return
+            for index, video_obj in enumerate(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(1, 5))
+
+    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()
+        item.add_video_info("video_id", video_obj["id"])
+        item.add_video_info("video_title", video_obj["title"])
+        item.add_video_info("play_cnt", 0)
+        item.add_video_info("publish_time_stamp", int(time.time()))
+        item.add_video_info("out_user_id", video_obj["id"])
+        item.add_video_info("cover_url", video_obj["video_cover"])
+        item.add_video_info("like_cnt", 0)
+        item.add_video_info("share_cnt", 0)
+        item.add_video_info("comment_cnt", 0)
+        item.add_video_info("video_url", video_obj["video_url"])
+        item.add_video_info("out_video_id", video_obj["id"])
+        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 get_video_url(self, vid):
+        url = "http://8.217.192.46:8889/crawler/wei_quan_shi_pin/detail"
+
+        payload = json.dumps({
+            "content_id": f"{vid}"
+        })
+        headers = {
+            'Content-Type': 'application/json'
+        }
+        try:
+            response = requests.request("POST", url, headers=headers, data=payload)
+            response = response.json()
+            if response['code'] != 0:
+                self.aliyun_log.logging(
+                    code="3000",
+                    message="获取视频链接失败"
+                ),
+                return None
+            video_url = response['data']['data']['video_url_list'][0]['video_url']
+            return video_url
+        except Exception as e:
+            return None
+
+
+    def run(self):
+        self.get_recommend_list()
+
+
+if __name__ == '__main__':
+    J = ZFKZRecommend(
+        platform="zhufukuaizhuan",
+        mode="recommend",
+        rule_dict={},
+        user_list=[{'uid': "123456", 'nick_name': "xiaoxiao"}],
+
+    )
+    J.get_recommend_list()
+    # J.logic()

spider/spider_map.py

@@ -7,7 +7,11 @@ value是爬虫封装好的类
 from spider.crawler_online import *
 from spider.crawler_online.benshanzhufu import BSZHRecommend
 from spider.crawler_online.dakaiyinghaoyun import DKYHYRecommend
+from spider.crawler_online.lepaoledong import LPLDRecommend
+from spider.crawler_online.piaoquangushi import PQGSRecommend
+from spider.crawler_online.weiquanshipin import WQSPRecommend
 from spider.crawler_online.xiaoniangaotuijianliu import XNGTJLRecommend
+from spider.crawler_online.zhufukuaizhuan import ZFKZRecommend
 from spider.crawler_online.zhufuniannianshunxinjixiang import ZFNNSXJXRecommend
 from spider.crawler_online.zhufuquanzituijianliu import ZFQZTJLRecommend
 
@@ -84,6 +88,22 @@ spider_map = {
     "zhufuquanzituijianliu": {
         "recommend": ZFQZTJLRecommend
     },
+    # 乐跑乐动
+    "lepaoledong": {
+        "recommend": LPLDRecommend
+    },
+    # 篻圈故事
+    "piaoquangushi": {
+        "recommend": PQGSRecommend
+    },
+    # 微圈视频
+    "weiquanshipin": {
+        "recommend": WQSPRecommend
+    },
+    # 祝福快转
+    "zhufukuaizhuan": {
+        "recommend": ZFKZRecommend
+    },
     # 祝福年年顺心吉祥
     "zhufuniannianshunxinjixiang": {
         "recommend": ZFNNSXJXRecommend