""" 博清科技-祝福中老年之视频 @author luojunhui @date 2024-02-06 """ import os import sys import json import time import uuid import random import asyncio import aiohttp 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 BoQingZhuFu(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(time.time()) 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["id"]) item.add_video_info("video_title", video_obj["title"]) # item.add_video_info("publish_time_str", video_obj["create"]) 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["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()))) 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': 'api.newboqing.top', 'Content-Type': 'application/json', 'Accept-Language': 'zh-cn', 'Accept': '*/*', 'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 11_3 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E217 MicroMessenger/6.8.0(0x16080000) NetType/WIFI Language/en Branch/Br_trunk MiniProgramEnv/Mac', 'Referer': 'https://servicewechat.com/wxacce796175899acd/9/page-frame.html', 'token': 'eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpYXQiOjE3MDI0MzcyMDgsIm5iZiI6MTcwMjQzNzIwOCwiZXhwIjoxNzAyNDQ0NDA4LCJkYXRhIjp7InVzZXJfaWQiOjU3OTUyNzM0fX0.WhnR4quSvRGkTCr5HIY0b7_mUcbzNXLY5Y-iTdp2QuM', 'ik': 'b326b5062b2f0e69046810717534cb09' } url = "https://api.newboqing.top/index.php" # ?s=mobile/Video/getList&cid=1&page={}&api_version=4&appid=wxacce796175899acd&version=1.9.0&env_version=release&scene=1008&from_uid=57769568&share_time=1701661552497&referer_vid=231783".format( # i) params = { "s": "mobile/Video/getList", 'cid': 1, 'page': page_index, 'api_version': 4, 'appid': "wxacce796175899acd", 'version': "1.9.0", 'env_version': "release", 'scene': "1008" } await asyncio.sleep(5) async with session.get( url, headers=headers, params=params ) as response: response_text = await response.text() response_json = json.loads(response_text) # print(json.dumps(response_json, ensure_ascii=False, indent=4)) for index, video_obj in enumerate(response_json["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 page in range(1, 20): 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: # print(e) self.aliyun_log.logging( code="3000", message="抓取第{}页时候出现错误, 报错信息是{}".format(page, e), ) # if __name__ == '__main__': # BQ = BoQingZhuFu( # platform="BoQingZhuFu", # mode="recommend", # rule_dict={}, # user_list=[{"uid": "12345", "nick_name": "luojunhui"}], # ) # loop = asyncio.get_event_loop() # loop.run_until_complete(BQ.run())