# -*- coding: utf-8 -*- # @Author: luojunhui # @Time: 2023/10/10 import json import os import random import sys import time import requests from Crypto.Cipher import AES from Crypto.Hash import MD5 from Crypto.Util.Padding import pad, unpad from base64 import b64encode, b64decode from common.mq import MQ sys.path.append(os.getcwd()) from common.common import Common from common.scheduling_db import MysqlHelper from common.public import get_config_from_mysql, download_rule_v2 proxies = {"http": None, "https": None} def clean_title(strings): return ( strings.strip() .replace("\n", "") .replace("/", "") .replace("\r", "") .replace("#", "") .replace(".", "。") .replace("\\", "") .replace("&NBSP", "") .replace(":", "") .replace("*", "") .replace("？", "") .replace("?", "") .replace('"', "") .replace("<", "") .replace(">", "") .replace("|", "") .replace(" ", "") .replace('"', "") .replace("'", "") ) def decrypt(a, e, n): e = MD5.new(e.encode()).hexdigest() key = e[16:].encode() iv = e[:16].encode() cipher = AES.new(key, AES.MODE_CBC, iv) if n: encrypted_data = b64decode(a) # print(encrypted_data) decrypted_data = unpad(cipher.decrypt(encrypted_data), AES.block_size) return decrypted_data.decode() else: padded_data = pad(a.encode(), AES.block_size) encrypted_data = cipher.encrypt(padded_data) return b64encode(encrypted_data).decode() def find_tencent_url(tx_vid): headers = { "Host": "h5vv.video.qq.com", "xweb_xhr": "1", "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/98.0.4758.102 Safari/537.36 MicroMessenger/6.8.0(0x16080000) NetType/WIFI MiniProgramEnv/Mac MacWechat/WMPF XWEB/30817", "Content-Type": "application/x-www-form-urlencoded", "Accept": "*/*", "Sec-Fetch-Site": "cross-site", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Dest": "empty", "Referer": "https://servicewechat.com/wx5fcd817f3f80aece/3/page-frame.html", "Accept-Language": "en", } video_id = tx_vid url = "https://h5vv.video.qq.com/getinfo?vid={}&platform=101001&charge=0&otype=json&defn=shd".format( video_id ) response = requests.get(url, headers=headers) result = json.loads(response.text.replace("QZOutputJson=", "")[:-1]) vl = result["vl"]["vi"][0] key = vl["fvkey"] name = vl["fn"] folder = vl["ul"]["ui"][0]["url"] url = folder + name + "?vkey=" + key return url class GGDCScheduling: def __init__(self, log_type, crawler, rule_dict, env, our_uid): self.platform = "刚刚都传" self.log_type = log_type self.crawler = crawler self.rule_dict = rule_dict self.env = env self.our_uid = our_uid self.mq = MQ(topic_name="topic_crawler_etl_" + self.env) self.download_count = 0 def repeat_video(self, video_id): sql = f""" select * from crawler_video where platform in ("{self.crawler}","{self.platform}") and out_video_id="{video_id}"; """ repeat_video = MysqlHelper.get_values( self.log_type, self.crawler, sql, self.env ) return len(repeat_video) # 获取视频id_list def get_videoList(self, page_id): time.sleep(random.randint(5, 10)) url = "https://ganggangdouchuan2.mengniu99.com/api/getcatevideos" params = { "cateid": "video", "page": page_id, "timeline": 0, "version": "9.0.2", # "timestamp": 1697700674000, # "openid": "oNnpe5SwkfGtD5aJAaRbsIKIEdjc", } headers = { "Host": "ganggangdouchuan2.mengniu99.com", # "Authorization": "oNnpe5SwkfGtD5aJAaRbsIKIEdjc", "xweb_xhr": "1", "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/98.0.4758.102 Safari/537.36 MicroMessenger/6.8.0(0x16080000) NetType/WIFI MiniProgramEnv/Mac MacWechat/WMPF XWEB/30817", # "Sign": "88ab143a9401ebac3525562073248231", "Content-Type": "application/json", "Accept": "*/*", "Sec-Fetch-Site": "cross-site", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Dest": "empty", "Referer": "https://servicewechat.com/wx5fcd817f3f80aece/3/page-frame.html", "Accept-Language": "en", } response = requests.get(url, headers=headers, params=params) while True: try: decrypted_data = decrypt( response.json()["data"], response.json()["_yyy"], True ) result = json.loads(decrypted_data) break except: print("编码不对,解密失败, 等待10秒钟") time.sleep(10) if "totalCount" not in result or response.status_code != 200: Common.logger(self.log_type, self.crawler).info( f"get_videoList:{response.text}\n" ) Common.logging( self.log_type, self.crawler, self.env, f"get_videoList:{response.text}\n", ) return elif len(result["videos"]) == 0: Common.logger(self.log_type, self.crawler).info(f"没有更多数据啦~\n") Common.logging(self.log_type, self.crawler, self.env, f"没有更多数据啦~\n") return else: data_list = result["videos"] for video_obj in data_list: try: self.process_video_obj(video_obj) except Exception as e: Common.logger(self.log_type, self.crawler).error(f"抓取单条视频异常:{e}\n") Common.logging( self.log_type, self.crawler, self.env, f"抓取单条视频异常:{e}\n" ) def process_video_obj(self, video_obj): video_id = video_obj.get("videoid", 0) video_title = clean_title(video_obj.get("title", "no title")) video_time = video_obj.get("v_time", 0) publish_time_stamp = int(time.time()) publish_time_str = time.strftime( "%Y-%m-%d %H:%M:%S", time.localtime(publish_time_stamp) ) user_name = video_obj["nickname"] video_dict = { "video_title": video_title, "video_id": video_id, "duration": video_time, "play_cnt": 0, "like_cnt": 0, "comment_cnt": 0, "share_cnt": 0, "user_name": user_name, "publish_time_stamp": publish_time_stamp, "publish_time_str": publish_time_str, "video_width": 0, "video_height": 0, "profile_id": 0, "profile_mid": 0, "cover_url": video_obj["cover"], "session": f"ganggangdouchuan-{int(time.time())}", } for k, v in video_dict.items(): Common.logger(self.log_type, self.crawler).info(f"{k}:{v}") Common.logging(self.log_type, self.crawler, self.env, f"{video_dict}") # 过滤无效视频 if video_title == "" or video_dict["video_id"] == "": Common.logger(self.log_type, self.crawler).info("无效视频\n") Common.logging(self.log_type, self.crawler, self.env, "无效视频\n") # 抓取基础规则过滤 elif ( download_rule_v2( log_type=self.log_type, crawler=self.crawler, video_dict=video_dict, rule_dict=self.rule_dict, ) is False ): Common.logger(self.log_type, self.crawler).info("不满足抓取规则\n") Common.logging(self.log_type, self.crawler, self.env, "不满足抓取规则\n") elif ( any( str(word) if str(word) in video_dict["video_title"] else False for word in get_config_from_mysql( log_type=self.log_type, source=self.crawler, env=self.env, text="filter", action="", ) ) is True ): Common.logger(self.log_type, self.crawler).info("已中过滤词\n") Common.logging(self.log_type, self.crawler, self.env, "已中过滤词\n") elif self.repeat_video(video_dict["video_id"]) != 0: Common.logger(self.log_type, self.crawler).info("视频已下载\n") Common.logging(self.log_type, self.crawler, self.env, "视频已下载\n") else: video_dict["out_user_id"] = video_dict["profile_id"] video_dict["platform"] = self.crawler video_dict["strategy"] = self.log_type video_dict["out_video_id"] = str(video_dict["video_id"]) video_dict["width"] = video_dict["video_width"] video_dict["height"] = video_dict["video_height"] video_dict["crawler_rule"] = json.dumps(self.rule_dict) video_dict["user_id"] = self.our_uid video_dict["publish_time"] = video_dict["publish_time_str"] video_dict["video_url"] = find_tencent_url(video_obj["txvid"]) video_dict["avatar_url"] = video_obj["avatarurl"] video_dict["cover_url"] = video_obj["cover"] # print(json.dumps(video_dict, ensure_ascii=False, indent=4)) self.download_count += 1 self.mq.send_msg(video_dict) if __name__ == "__main__": ZL = GGDCScheduling( log_type="recommend", crawler="ggdc", rule_dict={}, our_uid="luojunhuihaoshuai", env="dev", ) for i in range(5): ZL.get_videoList(page_id=i + 1) print(ZL.download_count)