上线——愿你福气常在

新增组件——ETL

application/config/topic_group_queue.py

@@ -13,7 +13,8 @@ class TopicGroup(object):
             ('syzf', 'recommend', 'sharkzhufu'),
             ("xsdd", 'recommend', 'xishiduoduo'),
             ("jxxf", 'recommend', 'jixiangxingfu'),
-            ("xnght", 'recommend', 'xiaoniangaohuati')
+            ("xnght", 'recommend', 'xiaoniangaohuati'),
+            ('ynfqcz', 'recommend', 'yuannifuqichangzai')
         ]
 
     def produce(self):

application/etl/download.py

@@ -0,0 +1,134 @@
+"""
+下载视频
+"""
+import os
+import json
+import time
+import asyncio
+from hashlib import md5
+import datetime
+
+import httpx
+import requests
+
+
+class VideoDownloader(object):
+    """
+    视频下载功能
+    """
+
+    def __init__(self, video_obj):
+        self.platform = video_obj['platform']
+        self.video_id = video_obj['video_id']
+        self.video_url = video_obj['video_url']
+        self.cover_url = video_obj['cover_url']
+        self.proxy = {
+            "http://": "http://t17772369458618:5zqcjkmy@q796.kdltps.com:15818/",
+            "https://": "http://t17772369458618:5zqcjkmy@q796.kdltps.com:15818/"
+        }
+        self.max_retry = 5
+
+    def generate_video_path(self):
+        """
+        通过视频信息生成唯一视频地址
+        :return:
+        """
+        index = "{}-{}".format(self.platform, self.video_id)
+        index = md5(index.encode()).hexdigest()
+        temp_dir = "/Users/luojunhui/cyber/automatic_crawler"
+        file_name = "{}.mp4".format(index)
+        date_info = datetime.datetime.today().strftime("%Y%m%d")
+        video_path = os.path.join(temp_dir, date_info, file_name)
+        if os.path.exists(video_path):
+            return
+        else:
+            os.makedirs(os.path.dirname(video_path), exist_ok=True)
+        return video_path
+
+    async def download_video(self):
+        """
+        download video from the web
+        :return:
+        """
+        if self.platform == "fuqiwang":
+            download_path = self.generate_video_path()
+            if download_path:
+                headers = {
+                    'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36 MicroMessenger/6.8.0(0x16080000) NetType/WIFI MiniProgramEnv/Mac MacWechat/WMPF MacWechat/3.8.6(0x13080610) XWEB/1156',
+                    'Accept-Encoding': 'identity;q=1, *;q=0',
+                    'Accept': '*/*',
+                    'Sec-Fetch-Site': 'cross-site',
+                    'Sec-Fetch-Mode': 'no-cors',
+                    'Sec-Fetch-Dest': 'video',
+                    'Referer': 'https://servicewechat.com/wxa1431c6e7acdd32d/2/page-frame.html',
+                    'Accept-Language': 'en-US,en;q=0.9',
+                    'Range': 'bytes=0-',
+                }
+                async with httpx.AsyncClient(http2=True, proxies=self.proxy, headers=headers) as client:
+                    try:
+                        response = await client.get(self.video_url, headers=headers)
+                        if response.status_code == 206:
+                            with open(download_path, "wb") as f:
+                                f.write(response.content)
+                        else:
+                            for _ in range(self.max_retry):
+                                response = await client.get(self.video_url, headers=headers, follow_redirects=True)
+                                if response.status_code == 206:
+                                    with open(download_path, "wb") as f:
+                                        f.write(response.content)
+                                    break
+                    except httpx.HTTPError as e:
+                        print(f"An error occurred while downloading: {e}")
+            else:
+                print("视频已经存在")
+
+    def get_by_request(self):
+        """
+        req
+        :return:
+        """
+        download_path = self.generate_video_path()
+        headers = {
+            'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36 MicroMessenger/6.8.0(0x16080000) NetType/WIFI MiniProgramEnv/Mac MacWechat/WMPF MacWechat/3.8.6(0x13080610) XWEB/1156',
+            'Accept-Encoding': 'identity;q=1, *;q=0',
+            'Accept': '*/*',
+            'Sec-Fetch-Site': 'cross-site',
+            'Sec-Fetch-Mode': 'no-cors',
+            'Sec-Fetch-Dest': 'video',
+            'Referer': 'https://servicewechat.com/wxa1431c6e7acdd32d/2/page-frame.html',
+            'Accept-Language': 'en-US,en;q=0.9',
+            'Range': 'bytes=0-',
+        }
+        r = requests.get(
+            url=self.video_url,
+            headers=headers,
+            proxies=self.proxy
+        )
+        print(r.status_code)
+        with open("test.mp4", "wb") as f:
+            f.write(r.content)
+
+
+async def main(video_obj):
+    """
+    异步执行函数
+    :param video_obj:
+    :return:
+    """
+    downloader = VideoDownloader(video_obj)
+    await downloader.download_video()
+
+
+if __name__ == '__main__':
+    video_o = {
+        "update_time": 1709784300,
+        "platform": "fuqiwang",
+        "video_id": 142599,
+        "title": "🔴3·8妇女节，最美的祝福，送给全天下的女神！",
+        "type": 1,
+        "video_type": 2,
+        "cover_url": "https://znl-video-bos.cdn.bcebos.com/c6f12b49992ef638342065439f55b444/65e93632/picture/20240306/b8b0c1cc262c2394f111650c9f82e35a_thumb.jpg",
+        "video_url": "https://znl-video-bos.cdn.bcebos.com/e368801a814c548e443835086d37caaf/65e93632/video/20240306/820ee1498e3ed2a59d37aed54d39ae95_1.mp4",
+    }
+    VideoDownloader(video_obj=video_o).get_by_request()
+    # asyncio.run(main(video_obj=video_o))

spider/crawler_online/__init__.py

@@ -7,4 +7,5 @@ from .boqingzhufu import BoQingZhuFu
 from .shayuzhufu import SharkZhuFuRecommend
 from .jiajiezhufuxishiduoduo import XiShiDuoDuoRecommend
 from .jixiangxingfu import JXXFRecommend
-from .xiaoniangaohuati import XNGHTecommend
+from .xiaoniangaohuati import XNGHTecommend
+from .yuannifuqichangzai import YuanNiFuQiChangZai

spider/crawler_online/yuannifuqichangzai.py

@@ -0,0 +1,191 @@
+"""
+@author: luojunhui
+"""
+import os
+import sys
+import json
+import time
+import uuid
+import random
+import asyncio
+import aiohttp
+import datetime
+
+sys.path.append(os.getcwd())
+
+from application.items import VideoItem
+from application.pipeline import PiaoQuanPipeline
+from application.common.messageQueue import MQ
+from application.common.proxies import tunnel_proxies
+from application.common.log import AliyunLogger
+
+
+class YuanNiFuQiChangZai(object):
+    """
+    愿你福气常在——推荐爬虫
+    """
+
+    def __init__(self, platform, mode, rule_dict, user_list, env="prod"):
+        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(platform=self.platform, mode=self.mode)
+
+    async def process_video_obj(self, video_obj):
+        """
+        处理每一个视频内容
+        :return: None
+        """
+        trace_id = self.platform + str(uuid.uuid1())
+        our_user = random.choice(self.user_list)
+        publish_time_stamp = int(video_obj['update_time'])
+        publish_time_str = datetime.datetime.fromtimestamp(publish_time_stamp).strftime('%Y-%m-%d %H:%M:%S')
+        item = VideoItem()
+        item.add_video_info("user_id", our_user["uid"])
+        item.add_video_info("user_name", our_user["nick_name"])
+        item.add_video_info("video_id", video_obj["nid"])
+        item.add_video_info("video_title", video_obj["title"])
+        item.add_video_info("publish_time_str", publish_time_str)
+        item.add_video_info("publish_time_stamp", int(publish_time_stamp))
+        item.add_video_info("video_url", video_obj["video_url"])
+        item.add_video_info("cover_url", video_obj["video_cover"])
+        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())))
+        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)
+            # print(json.dumps(mq_obj, ensure_ascii=False, indent=4))
+            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.expire_flag = True
+
+    async def get_recommend_list(self, session, page_index):
+        """
+        获取推荐页面的video_list
+        :param session: aiohttp 的session
+        :param page_index: 页码
+        :return: None
+        """
+        if self.expire_flag:
+            self.aliyun_log.logging(
+                code="2000",
+                message="本轮已经抓取到足够的数据，自动退出\t{}".format(self.download_cnt),
+            )
+            return
+        headers = {
+            'Host': 'ynfqcz.jiabeijian.cn',
+            'content-time': str(int(time.time() * 1000)),
+            'cache-time': str(int(time.time() * 1000)),
+            'chatkey': 'wxa1431c6e7acdd32d',
+            'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36 MicroMessenger/6.8.0(0x16080000) NetType/WIFI MiniProgramEnv/Mac MacWechat/WMPF MacWechat/3.8.6(0x13080610) XWEB/1156',
+            'content-type': 'application/x-www-form-urlencoded',
+            'visitorkey': '17096322301026589978',
+            'xweb_xhr': '1',
+            'vision': '1.1.0',
+            'token': '',
+            'accept': '*/*',
+            'sec-fetch-site': 'cross-site',
+            'sec-fetch-mode': 'cors',
+            'sec-fetch-dest': 'empty',
+            'referer': 'https://servicewechat.com/wxa1431c6e7acdd32d/2/page-frame.html',
+            'accept-language': 'en-US,en;q=0.9'
+        }
+        po = {
+            "cid": "",
+            "page": page_index,
+            "is_ads": 1,
+            "model": random.choice(["Windows", "Mac", "HuaWei", "Xiaomi", "Xiaomi2", "Yandex", "Google", "iphone", "oppo"]),
+            "mini_version": "3.8.6",
+            "ini_id": "17096322301026589978"
+        }
+        params = {
+            "parameter": json.dumps(po)
+        }
+        url = "https://ynfqcz.jiabeijian.cn/index.php/v111/index/index"
+        await asyncio.sleep(5)
+        async with session.get(
+                url, headers=headers, params=params, proxy=tunnel_proxies()['https']
+        ) as response:
+            data = await response.json()
+            for index, video_obj in enumerate(data["data"]["list"], 1):
+                try:
+                    self.aliyun_log.logging(
+                        code="1001",
+                        message="扫描到一条视频",
+                        data=video_obj,
+                    )
+                    await self.process_video_obj(video_obj)
+                except Exception as e:
+                    self.aliyun_log.logging(
+                        code="3000",
+                        message="抓取第{}条的时候出现问题， 报错信息是{}".format(index, e),
+                    )
+
+    async def run(self):
+        """
+        执行代码
+        :return: None
+        """
+        async with aiohttp.ClientSession() as session:
+            # for i in range(0, 100, 20):
+            #     if self.expire_flag:
+            #         return
+            #     tasks = [self.get_recommend_list(session, index) for index in range(i, min(i + 20, 200))]
+            #     await asyncio.gather(*tasks)
+            #     done, pending = await asyncio.wait(
+            #         tasks, return_when=asyncio.FIRST_COMPLETED
+            #     )
+            #     # 取消所有剩余的任务
+            #     for task in pending:
+            #         task.cancel()
+
+            # tasks = [self.get_recommend_list(session, index) for index in range(1, 100)]
+            # await asyncio.gather(*tasks)
+            # done, pending = await asyncio.wait(
+            #     tasks, return_when=asyncio.FIRST_COMPLETED
+            # )
+            # # 取消所有剩余的任务
+            # for task in pending:
+            #     task.cancel()
+            for page in range(1, 30):
+                if self.expire_flag:
+                    self.aliyun_log.logging(
+                        code="2000",
+                        message="本轮已经抓取到足够的数据，自动退出\t{}".format(self.download_cnt),
+                    )
+                    # message = "本轮已经抓取到足够的数据，自动退出\t{}".format(self.download_cnt)
+                    # print(message)
+                    return
+                else:
+                    try:
+                        await self.get_recommend_list(session, page_index=page)
+                    except Exception as e:
+                        self.aliyun_log.logging(
+                            code="3000",
+                            message="抓取第{}页时候出现错误， 报错信息是{}".format(page, e),
+                        )
+                        # message = "抓取第{}页时候出现错误， 报错信息是{}".format(page, e)
+                        # print(message)
+

spider/spider_map.py

@@ -46,6 +46,10 @@ spider_map = {
     # 小年糕-话题
     "xiaoniangaohuati": {
         "recommend": XNGHTecommend
+    },
+    # 福气旺系列
+    "yuannifuqichangzai": {
+        "recommend": YuanNiFuQiChangZai
     }
 
 }