|
@@ -0,0 +1,394 @@
|
|
|
+import os
|
|
|
+import json
|
|
|
+import random
|
|
|
+import sys
|
|
|
+import time
|
|
|
+import uuid
|
|
|
+import requests
|
|
|
+from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
|
|
+from cryptography.hazmat.backends import default_backend
|
|
|
+
|
|
|
+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
|
|
|
+from application.common.mysql import MysqlHelper
|
|
|
+
|
|
|
+
|
|
|
+class AES(object):
|
|
|
+ """
|
|
|
+ 祝万物复苏
|
|
|
+ """
|
|
|
+
|
|
|
+ def __init__(self):
|
|
|
+ self.key = b"50102fa64073ad76" # 用适当的方式转换或直接定义为字节串
|
|
|
+ self.iv = b"173d023138824bb0" # 同上
|
|
|
+
|
|
|
+ def aes_encrypt(self, data):
|
|
|
+ """
|
|
|
+ aes 加密
|
|
|
+ :param data:
|
|
|
+ :return:
|
|
|
+ """
|
|
|
+ cipher = Cipher(
|
|
|
+ algorithms.AES(self.key), modes.CBC(self.iv), backend=default_backend()
|
|
|
+ )
|
|
|
+ encryptor = cipher.encryptor()
|
|
|
+ ct = encryptor.update(self._pad(data).encode()) + encryptor.finalize()
|
|
|
+ return ct.hex().upper()
|
|
|
+
|
|
|
+ def aes_decrypt(self, data):
|
|
|
+ """
|
|
|
+ 解密
|
|
|
+ :param data:
|
|
|
+ :return:
|
|
|
+ """
|
|
|
+ cipher = Cipher(
|
|
|
+ algorithms.AES(self.key), modes.CBC(self.iv), backend=default_backend()
|
|
|
+ )
|
|
|
+ decryptor = cipher.decryptor()
|
|
|
+ decrypted_data = decryptor.update(bytes.fromhex(data)) + decryptor.finalize()
|
|
|
+ return self._unpad(decrypted_data).decode()
|
|
|
+
|
|
|
+ def _pad(self, s):
|
|
|
+ """
|
|
|
+ 填充
|
|
|
+ :param s:
|
|
|
+ :return:
|
|
|
+ """
|
|
|
+ return s + (16 - len(s) % 16) * chr(16 - len(s) % 16)
|
|
|
+
|
|
|
+ def _unpad(self, s):
|
|
|
+ """
|
|
|
+ 非填充
|
|
|
+ :param s:
|
|
|
+ :return:
|
|
|
+ """
|
|
|
+ return s[: -ord(s[len(s) - 1:])]
|
|
|
+
|
|
|
+
|
|
|
+class ZhuWanWuFuSuRecommend(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.cryptor = AES()
|
|
|
+ self.aliyun_log = AliyunLogger(mode=self.mode, platform=self.platform)
|
|
|
+ self.mysql = MysqlHelper(mode=self.mode, platform=self)
|
|
|
+
|
|
|
+ def get_recommend_list(self):
|
|
|
+ """
|
|
|
+ 获取推荐页视频
|
|
|
+ """
|
|
|
+ url = "https://api.lidongze.cn/jeecg-boot/ugc/getVideoListsEn2"
|
|
|
+ headers = {
|
|
|
+ "Host": "api.lidongze.cn",
|
|
|
+ "xweb_xhr": "1",
|
|
|
+ "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",
|
|
|
+ "token": "",
|
|
|
+ "content-type": "application/json",
|
|
|
+ "accept": "*/*",
|
|
|
+ "referer": "https://servicewechat.com/wx0afdc2669ed8df2f/3/page-frame.html",
|
|
|
+ "accept-language": "en-US,en;q=0.9",
|
|
|
+ }
|
|
|
+ page_index = 1
|
|
|
+ total_page = 2
|
|
|
+ while page_index <= total_page:
|
|
|
+ try:
|
|
|
+ query = {
|
|
|
+ "pageNo": page_index,
|
|
|
+ "pageSize": 10,
|
|
|
+ "groupId": "1650323161797439489", # 推荐流的 ID
|
|
|
+ "vn": 1,
|
|
|
+ "gx": 1,
|
|
|
+ "appid": "wx0afdc2669ed8df2f",
|
|
|
+ "type": 0,
|
|
|
+ }
|
|
|
+ params = {"v": self.cryptor.aes_encrypt(data=json.dumps(query))}
|
|
|
+ response = requests.get(
|
|
|
+ url, headers=headers, params=params, proxies=tunnel_proxies()
|
|
|
+ )
|
|
|
+ result = json.loads(self.cryptor.aes_decrypt(response.text))
|
|
|
+ total_page = result["list"]["pages"]
|
|
|
+ page_index = result["list"]["current"] + 1
|
|
|
+ for index, video_obj in enumerate(result["list"]["records"], 1):
|
|
|
+ try:
|
|
|
+ self.aliyun_log.logging(
|
|
|
+ code="1001", message="扫描到一条视频", data=video_obj
|
|
|
+ )
|
|
|
+ self.process_video_obj(video_obj)
|
|
|
+ # print(json.dumps(video_obj, ensure_ascii=False, indent=4))
|
|
|
+ except Exception as e:
|
|
|
+ self.aliyun_log.logging(
|
|
|
+ code="3000",
|
|
|
+ message="抓取单条视频失败, 该视频位于第{}页第{}条报错原因是{}".format(
|
|
|
+ page_index, index, e
|
|
|
+ ),
|
|
|
+ )
|
|
|
+ except Exception as e:
|
|
|
+ self.aliyun_log.logging(
|
|
|
+ code="3000", message="抓取第{}页的时候失败, 报错原因是{}".format(page_index, e)
|
|
|
+ )
|
|
|
+ time.sleep(random.randint(5, 10))
|
|
|
+
|
|
|
+ def get_user_videos(self, user_id):
|
|
|
+ """
|
|
|
+ 在抓取完推荐页之后,去抓每一个用户的主页视频
|
|
|
+ """
|
|
|
+ url = "https://api.lidongze.cn/jeecg-boot/ugc/getAuthVideoList"
|
|
|
+ headers = {
|
|
|
+ "Host": "api.lidongze.cn",
|
|
|
+ "xweb_xhr": "1",
|
|
|
+ "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",
|
|
|
+ "token": "",
|
|
|
+ "content-type": "application/json",
|
|
|
+ "accept": "*/*",
|
|
|
+ "referer": "https://servicewechat.com/wx0afdc2669ed8df2f/3/page-frame.html",
|
|
|
+ "accept-language": "en-US,en;q=0.9",
|
|
|
+ }
|
|
|
+ page_index = 1
|
|
|
+ total_page = 1
|
|
|
+ while page_index <= total_page:
|
|
|
+ query = {"pageNo": page_index, "pageSize": 10, "authid": user_id}
|
|
|
+ params = {"v": self.cryptor.aes_encrypt(data=json.dumps(query))}
|
|
|
+ response = requests.request(
|
|
|
+ "GET", url, headers=headers, params=params, proxies=tunnel_proxies()
|
|
|
+ )
|
|
|
+ result = json.loads(self.cryptor.aes_decrypt(response.text))
|
|
|
+ total_page = result["list"]["pages"]
|
|
|
+ page_index = result["list"]["current"] + 1
|
|
|
+ for index, video_temp in enumerate(result["list"]["records"]):
|
|
|
+ video_id = video_temp["id"]
|
|
|
+ detail_query = {"videoId": video_id}
|
|
|
+ detail_params = {
|
|
|
+ "v": self.cryptor.aes_encrypt(data=json.dumps(detail_query))
|
|
|
+ }
|
|
|
+ url = "https://api.lidongze.cn/jeecg-boot/ugc/getVideosDataEn"
|
|
|
+ headers = {
|
|
|
+ "Host": "api.lidongze.cn",
|
|
|
+ "xweb_xhr": "1",
|
|
|
+ "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",
|
|
|
+ "token": "",
|
|
|
+ "content-type": "application/json",
|
|
|
+ "accept": "*/*",
|
|
|
+ "referer": "https://servicewechat.com/wx0afdc2669ed8df2f/3/page-frame.html",
|
|
|
+ "accept-language": "en-US,en;q=0.9",
|
|
|
+ }
|
|
|
+ detail_response = requests.request(
|
|
|
+ "GET",
|
|
|
+ url,
|
|
|
+ headers=headers,
|
|
|
+ params=detail_params,
|
|
|
+ proxies=tunnel_proxies(),
|
|
|
+ )
|
|
|
+ detail_video = json.loads(
|
|
|
+ self.cryptor.aes_decrypt(detail_response.text)
|
|
|
+ )
|
|
|
+ if detail_video["success"]:
|
|
|
+ try:
|
|
|
+ self.aliyun_log.logging(
|
|
|
+ code="1001", message="扫描到一条视频", data=detail_video["data"]
|
|
|
+ )
|
|
|
+ self.process_video_obj(detail_video["data"])
|
|
|
+ except Exception as e:
|
|
|
+ self.aliyun_log.logging(
|
|
|
+ code="3000",
|
|
|
+ message="抓取单条视频失败, 该视频位于第{}条报错原因是{}".format(index, e),
|
|
|
+ )
|
|
|
+
|
|
|
+ def process_video_obj(self, video_obj):
|
|
|
+ """
|
|
|
+ 处理视频
|
|
|
+ :param video_obj:
|
|
|
+ """
|
|
|
+ time.sleep(random.randint(3, 8))
|
|
|
+ trace_id = self.platform + str(uuid.uuid1())
|
|
|
+ if video_obj.get("playnum"):
|
|
|
+ play_cnt = (
|
|
|
+ int(video_obj["playnum"].replace("万+", "0000"))
|
|
|
+ if "万+" in video_obj["playnum"]
|
|
|
+ else int(video_obj["playnum"])
|
|
|
+ )
|
|
|
+ else:
|
|
|
+ play_cnt = 0
|
|
|
+ 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["vname"])
|
|
|
+ item.add_video_info("play_cnt", play_cnt)
|
|
|
+ item.add_video_info("publish_time_stamp", int(time.time()))
|
|
|
+ item.add_video_info("out_user_id", video_obj["authid"])
|
|
|
+ item.add_video_info("cover_url", video_obj["shareimg"])
|
|
|
+ item.add_video_info("like_cnt", int(video_obj["likenum"]))
|
|
|
+ item.add_video_info("video_url", video_obj["videoaddr"])
|
|
|
+ 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"])
|
|
|
+ # 把扫描到的账号存到 accounts 表中
|
|
|
+ self.manage_auth_id(
|
|
|
+ out_user_id=video_obj["authid"], out_user_name=video_obj["authname"]
|
|
|
+ )
|
|
|
+ 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 manage_auth_id(self, out_user_id, out_user_name):
|
|
|
+ """
|
|
|
+ out_user_id: 外站视频的用户 id
|
|
|
+ out_user_name: 外站视频用户名字
|
|
|
+ 逻辑: 对新扫描到的视频的用户 id 进行判断,若用户 id 不存在,则把视频 id 存到表中,
|
|
|
+ 如果用户 id 存在,则判断用户是否修改名字,若名字修改则更新名字
|
|
|
+ """
|
|
|
+ select_user_sql = f"""select name, name_id from accounts where name_id = "{out_user_id}" and platform = "{self.platform}" and useful = 1 limit 1"""
|
|
|
+ out_user_info = self.mysql.select(sql=select_user_sql)
|
|
|
+ if out_user_info:
|
|
|
+ name, name_id = out_user_info[0]
|
|
|
+ if name == out_user_name:
|
|
|
+ return
|
|
|
+ else:
|
|
|
+ update_sql = f"""update accounts set name = "{out_user_name}" where name_id = "{out_user_id}";"""
|
|
|
+ self.mysql.update(sql=update_sql)
|
|
|
+ else:
|
|
|
+ insert_sql = f"""INSERT INTO accounts (name, name_id, platform, useful) values ("{out_user_name}", "{out_user_id}", "{self.platform}", 1 )"""
|
|
|
+ self.mysql.update(sql=insert_sql)
|
|
|
+
|
|
|
+ def get_user_list(self):
|
|
|
+ """
|
|
|
+ 获取用户列表
|
|
|
+ :return:
|
|
|
+ """
|
|
|
+ select_user_sql = f"""select name_id from accounts where platform = "{self.platform}" and useful = 1"""
|
|
|
+ out_user_info = self.mysql.select(
|
|
|
+ sql=select_user_sql,
|
|
|
+ )
|
|
|
+ if out_user_info:
|
|
|
+ result = []
|
|
|
+ for i in out_user_info:
|
|
|
+ result.append(i[0])
|
|
|
+ return result
|
|
|
+ else:
|
|
|
+ return []
|
|
|
+
|
|
|
+ def get_detail_video_list(self):
|
|
|
+ """
|
|
|
+ 获取推荐列视频
|
|
|
+ :return:
|
|
|
+ """
|
|
|
+ url = "https://api.lidongze.cn/jeecg-boot/ugc/getDetailVideoListsEn2"
|
|
|
+ headers = {
|
|
|
+ "Host": "api.lidongze.cn",
|
|
|
+ "xweb_xhr": "1",
|
|
|
+ "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.5(0x13080510)XWEB/1100",
|
|
|
+ "token": "",
|
|
|
+ "referer": "https://servicewechat.com/wx0afdc2669ed8df2f/3/page-frame.html",
|
|
|
+ "accept-language": "en-US,en;q=0.9",
|
|
|
+ }
|
|
|
+ page_index = 1
|
|
|
+ total_page = 2
|
|
|
+ while page_index <= total_page:
|
|
|
+ try:
|
|
|
+ if self.limit_flag:
|
|
|
+ self.aliyun_log.logging(code="2000", message="本轮已经抓取足够数量的视频")
|
|
|
+ return
|
|
|
+ else:
|
|
|
+ query = {
|
|
|
+ "groupId": "1650323161797439489",
|
|
|
+ "pageNo": page_index,
|
|
|
+ "pageSize": 10,
|
|
|
+ "appid": "wx0afdc2669ed8df2f",
|
|
|
+ "type": 3,
|
|
|
+ "hxid": "1556555457243828666",
|
|
|
+ }
|
|
|
+ params = {"v": self.cryptor.aes_encrypt(data=json.dumps(query))}
|
|
|
+ response = requests.request(
|
|
|
+ "GET", url, headers=headers, params=params
|
|
|
+ )
|
|
|
+ result = json.loads(self.cryptor.aes_decrypt(response.text))
|
|
|
+ total_page = result["list"]["pages"]
|
|
|
+ page_index = result["list"]["current"] + 1
|
|
|
+ for index, video_obj in enumerate(result["list"]["records"], 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(
|
|
|
+ page_index, index, e
|
|
|
+ ),
|
|
|
+ )
|
|
|
+ except Exception as e:
|
|
|
+ self.aliyun_log.logging(
|
|
|
+ code="3000", message="抓取第{}页的时候失败, 报错原因是{}".format(page_index, e)
|
|
|
+ )
|
|
|
+ time.sleep(random.randint(5, 10))
|
|
|
+
|
|
|
+ def run(self):
|
|
|
+ """
|
|
|
+ 先抓取推荐列表的视频, 等待 2 分钟后抓取 detail 页面,等待 5 分钟后,抓取账号视频
|
|
|
+ """
|
|
|
+ self.get_recommend_list()
|
|
|
+ if self.limit_flag:
|
|
|
+ return
|
|
|
+ time.sleep(2 * 60)
|
|
|
+ self.get_detail_video_list()
|
|
|
+ if self.limit_flag:
|
|
|
+ return
|
|
|
+ time.sleep(5 * 60)
|
|
|
+ self.mode = "author"
|
|
|
+ user_list = self.get_user_list()
|
|
|
+ if user_list:
|
|
|
+ for index, user_id in enumerate(user_list):
|
|
|
+ try:
|
|
|
+ if self.limit_flag:
|
|
|
+ self.aliyun_log.logging(code="2000", message="本轮已经抓取足够数量的视频")
|
|
|
+ return
|
|
|
+ self.get_user_videos(user_id=user_id)
|
|
|
+ except Exception as e:
|
|
|
+ self.aliyun_log.logging(
|
|
|
+ code="3000",
|
|
|
+ message="抓取账号视频出现异常,账号 id 是{}, 报错原因是{}".format(user_id, e),
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+# if __name__ == '__main__':
|
|
|
+# Z = ZhuWanWuFuSuRecommend(
|
|
|
+# platform="zhuwanwufusu2",
|
|
|
+# mode="recommend",
|
|
|
+# rule_dict={},
|
|
|
+# user_list=[{'uid': "123456", 'nick_name': "luojunhui"}],
|
|
|
+#
|
|
|
+# )
|
|
|
+# Z.get_recommend_list()
|