youlegaoxiaoxiaoshipin_scheduling.py 7.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192
  1. # -*- coding: utf-8 -*-
  2. # @Author: luojunhui
  3. # @Time: 2023/10/23
  4. import json
  5. import os
  6. import sys
  7. import time
  8. import random
  9. import requests
  10. sys.path.append(os.getcwd())
  11. from common.mq import MQ
  12. from common.aliyun_log import AliyunLogger
  13. from common.pipeline import PiaoQuanPipeline
  14. from common.public import clean_title
  15. class YLGXXSPScheduling:
  16. def __init__(self, platform, mode, rule_dict, env, our_uid):
  17. self.platform = platform
  18. self.mode = mode
  19. self.rule_dict = rule_dict
  20. self.env = env
  21. self.our_uid = our_uid
  22. self.mq = MQ(topic_name="topic_crawler_etl_" + self.env)
  23. self.download_count = 0
  24. # 获取视频id_list
  25. def get_videoList(self, page_id):
  26. time.sleep(random.randint(5, 10))
  27. AliyunLogger.logging(
  28. code="1000",
  29. platform=self.platform,
  30. mode=self.mode,
  31. env=self.env,
  32. data={},
  33. message="开始抓取第{}页".format(page_id),
  34. )
  35. headers = {
  36. "Host": "cpu.baidu.com",
  37. "xweb_xhr": "1",
  38. "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.4(0x13080410)XWEB/31009",
  39. "Accept": "*/*",
  40. "Sec-Fetch-Site": "cross-site",
  41. "Sec-Fetch-Mode": "cors",
  42. "Sec-Fetch-Dest": "empty",
  43. "Referer": "https://servicewechat.com/wx38382a240eab7214/4/page-frame.html",
  44. "Accept-Language": "en-US,en;q=0.9",
  45. }
  46. channel_id_dict = {
  47. '1059': "搞笑",
  48. "1058": "音乐",
  49. "1061": "娱乐",
  50. "1063": "社会",
  51. "1066": "生活",
  52. "1064": "猎奇"
  53. }
  54. for channel_id in channel_id_dict:
  55. data = {
  56. "channelId": channel_id,
  57. "needHybrid": "1",
  58. "pageNo": str(page_id),
  59. "pageSize": "10",
  60. }
  61. response = requests.post(
  62. "https://cpu.baidu.com/1033/a16a67fe", headers=headers, data=data
  63. )
  64. result = response.json()
  65. channel_name = channel_id_dict[channel_id]
  66. if "data" not in result or response.status_code != 200:
  67. AliyunLogger.logging(
  68. code="2000",
  69. platform=self.platform,
  70. mode=self.mode,
  71. env=self.env,
  72. data={},
  73. message="{}抓取第{}页失败,无数据".format(channel_name, page_id),
  74. )
  75. return
  76. elif len(result["data"]["result"]) == 0:
  77. AliyunLogger.logging(
  78. code="2001",
  79. platform=self.platform,
  80. mode=self.mode,
  81. env=self.env,
  82. data={},
  83. message="{}抓取d到第{}页, 没有更多数据了".format(channel_name, page_id),
  84. )
  85. return
  86. else:
  87. data_list = result["data"]["result"]
  88. for index, video_obj in enumerate(data_list):
  89. try:
  90. AliyunLogger.logging(
  91. code="1001",
  92. platform=self.platform,
  93. mode=self.mode,
  94. env=self.env,
  95. data={},
  96. message="{}成功扫描到一条视频, 该视频位于第{}页{}条".format(channel_name, page_id, index + 1),
  97. )
  98. self.process_video_obj(video_obj)
  99. except Exception as e:
  100. AliyunLogger.logging(
  101. code="3000",
  102. platform=self.platform,
  103. mode=self.mode,
  104. env=self.env,
  105. data=video_obj,
  106. message="{}抓取单条视频异常, 报错原因是: {}, 该视频位于第{}页{}条".format(
  107. channel_name, e, page_id, index + 1
  108. ),
  109. )
  110. AliyunLogger.logging(
  111. code="1000",
  112. platform=self.platform,
  113. mode=self.mode,
  114. env=self.env,
  115. data={},
  116. message="{}完成抓取第{}页".format(channel_name, page_id),
  117. )
  118. def process_video_obj(self, video_obj):
  119. video_id = video_obj.get("data", {}).get("id", 0)
  120. video_title = clean_title(video_obj.get("data", {}).get("title", "no title"))
  121. video_time = video_obj["data"]["duration"]
  122. publish_time_stamp = int(video_obj["data"]["clusterTime"])
  123. publish_time_str = time.strftime(
  124. "%Y-%m-%d %H:%M:%S", time.localtime(publish_time_stamp)
  125. )
  126. user_name = video_obj["data"]["source"]
  127. video_dict = {
  128. "video_title": video_title,
  129. "video_id": video_id,
  130. "duration": video_time,
  131. "play_cnt": int(video_obj["data"].get("playbackCount", 0)),
  132. "like_cnt": int(video_obj.get("likeCount", 0)),
  133. "comment_cnt": int(video_obj.get("commentCounts", 0)),
  134. "share_cnt": 0,
  135. "user_name": user_name,
  136. "publish_time_stamp": publish_time_stamp,
  137. "publish_time_str": publish_time_str,
  138. "video_width": 0,
  139. "video_height": 0,
  140. "profile_id": 0,
  141. "profile_mid": 0,
  142. "session": f"youlegaoxiaoxiaoshipin-{int(time.time())}",
  143. }
  144. rule_pipeline = PiaoQuanPipeline(
  145. platform=self.platform,
  146. mode=self.mode,
  147. rule_dict=self.rule_dict,
  148. env=self.env,
  149. item=video_dict,
  150. )
  151. flag = rule_pipeline.process_item()
  152. if flag:
  153. video_dict["out_user_id"] = video_obj["data"].get("ownerId", 0)
  154. video_dict["platform"] = self.platform
  155. video_dict["strategy"] = self.mode
  156. video_dict["out_video_id"] = str(video_dict["video_id"])
  157. video_dict["width"] = video_dict["video_width"]
  158. video_dict["height"] = video_dict["video_height"]
  159. video_dict["crawler_rule"] = json.dumps(self.rule_dict)
  160. video_dict["user_id"] = self.our_uid
  161. video_dict["publish_time"] = video_dict["publish_time_str"]
  162. video_dict["video_url"] = video_obj["data"]["url"]
  163. video_dict["avatar_url"] = "http:" + video_obj["data"]["avatar"]
  164. self.download_count += 1
  165. self.mq.send_msg(video_dict)
  166. AliyunLogger.logging(
  167. code="1002",
  168. platform=self.platform,
  169. mode=self.mode,
  170. env=self.env,
  171. data=video_dict,
  172. message="成功发送 MQ 至 ETL",
  173. )
  174. if __name__ == "__main__":
  175. ZL = YLGXXSPScheduling(
  176. platform="ylgxxsp",
  177. mode="recommend",
  178. rule_dict={},
  179. our_uid="luojunhuihaoshuai",
  180. env="prod",
  181. )
  182. for i in range(5):
  183. ZL.get_videoList(page_id=i + 1)
  184. print(ZL.download_count)