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 HHXXZFDRecommend(object): def __init__(self, platform, mode, rule_dict, user_list, env): 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.limit_flag = False self.cryptor = AES() self.aliyun_log = AliyunLogger(platform=self.platform, mode=self.mode) self.mysql = MysqlHelper(platform=self.platform, mode=self.mode) 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": "wx9a60184c443f39af", "type": 0 } 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_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 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): 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): 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": "wx9a60184c443f39af", "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) )