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 ) 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: # c += 1 # print(c) 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) ) if self.limit_flag: return 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 ) 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 ) 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()