import datetime import json import os import random import sys import time import uuid import requests import urllib3 from selenium.webdriver import DesiredCapabilities from selenium.webdriver.chrome.service import Service from selenium.webdriver.common.by import By from selenium import webdriver sys.path.append(os.getcwd()) from common.mq import MQ from common.common import Common from common.feishu import Feishu from common.scheduling_db import MysqlHelper from common.public import get_config_from_mysql, download_rule, title_like, task_unbind from common import AliyunLogger, PiaoQuanPipeline class GongzhonghaoAuthor: platform = "公众号" # 获取 token @classmethod def get_token(cls, log_type, crawler, token_index, env): select_sql = f""" select * from crawler_config where source="{crawler}" and title LIKE "%公众号_{token_index}%";""" configs = MysqlHelper.get_values(log_type, crawler, select_sql, env, action="") if len(configs) == 0: Feishu.bot(log_type, crawler, f"公众号_{token_index}:未配置token") time.sleep(60) return None token_dict = { "token_id": configs[0]["id"], "title": configs[0]["title"].strip(), "token": dict(eval(configs[0]["config"]))["token"].strip(), "cookie": dict(eval(configs[0]["config"]))["cookie"].strip(), "update_time": time.strftime( "%Y-%m-%d %H:%M:%S", time.localtime(int(configs[0]["update_time"] / 1000)), ), "operator": configs[0]["operator"].strip(), } return token_dict # 获取用户 fakeid @classmethod def get_user_info(cls, log_type, crawler, task_dict, user_dict, token_index, env): Common.logger(log_type, crawler).info(f"获取站外用户信息:{user_dict['link']}") Common.logging(log_type, crawler, env, f"获取站外用户信息:{user_dict['link']}") AliyunLogger.logging( code="1000", platform=crawler, mode=log_type, env=env, message=f"获取站外用户信息:{user_dict['link']}", ) while True: token_dict = cls.get_token(log_type, crawler, token_index, env) url = "https://mp.weixin.qq.com/cgi-bin/searchbiz?" headers = { "accept": "*/*", "accept-encoding": "gzip, deflate, br", "accept-language": "zh-CN,zh;q=0.9", "referer": "https://mp.weixin.qq.com/cgi-bin/appmsg?" "t=media/appmsg_edit_v2&action=edit&isNew=1" "&type=77&createType=5&token=1011071554&lang=zh_CN", "sec-ch-ua": '" Not A;Brand";v="99", "Chromium";v="100", "Google Chrome";v="100"', "sec-ch-ua-mobile": "?0", "sec-ch-ua-platform": '"Windows"', "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "same-origin", "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" " (KHTML, like Gecko) Chrome/100.0.4896.127 Safari/537.36", "x-requested-with": "XMLHttpRequest", "cookie": token_dict["cookie"], } params = { "action": "search_biz", "begin": "0", "count": "5", "query": str(user_dict["link"]), "token": token_dict["token"], "lang": "zh_CN", "f": "json", "ajax": "1", } urllib3.disable_warnings() r = requests.get(url=url, headers=headers, params=params, verify=False) r.close() if r.json()["base_resp"]["err_msg"] == "invalid session": Common.logger(log_type, crawler).warning( f"status_code:{r.status_code}, get_fakeid:{r.text}\n" ) Common.logging( log_type, crawler, env, f"status_code:{r.status_code}, get_fakeid:{r.text}\n", ) AliyunLogger.logging( code="2000", platform=crawler, mode=log_type, env=env, message=f"status_code:{r.status_code}, get_fakeid:{r.text}\n", ) if 20 >= datetime.datetime.now().hour >= 10: Feishu.bot( log_type, crawler, f"{token_dict['title']}\n操作人:{token_dict['operator']}\n更换日期:{token_dict['update_time']} \n过期啦，请扫码更换token\nhttps://mp.weixin.qq.com/", ) time.sleep(60 * 15) continue if r.json()["base_resp"]["err_msg"] == "freq control": Common.logger(log_type, crawler).warning( f"status_code:{r.status_code}, get_fakeid:{r.text}\n" ) Common.logging( log_type, crawler, env, f"status_code:{r.status_code}, get_fakeid:{r.text}\n", ) AliyunLogger.logging( code="2000", platform=crawler, mode=log_type, env=env, message=f"status_code:{r.status_code}, get_fakeid:{r.text}\n", ) if 20 >= datetime.datetime.now().hour >= 10: Feishu.bot( log_type, crawler, f"{token_dict['title']}\n操作人:{token_dict['operator']}\n更换日期:{token_dict['update_time']} \n频控啦，请扫码更换其他公众号token\nhttps://mp.weixin.qq.com/", ) time.sleep(60 * 15) continue if r.json()["base_resp"]["err_msg"] == "ok" and len(r.json()["list"]) == 0: Common.logger(log_type, crawler).warning( f"status_code:{r.status_code}, get_fakeid:{r.text}\n" ) Common.logging( log_type, crawler, env, f"status_code:{r.status_code}, get_fakeid:{r.text}\n", ) AliyunLogger.logging( code="2000", platform=crawler, mode=log_type, env=env, message=f"status_code:{r.status_code}, get_fakeid:{r.text}\n", ) unbind_msg = task_unbind( log_type=log_type, crawler=crawler, taskid=task_dict["id"], uids=str(user_dict["uid"]), env=env, ) if unbind_msg == "success": if 20 >= datetime.datetime.now().hour >= 10: Feishu.bot( log_type, crawler, f"公众号:{user_dict['link']}, 站内昵称:{user_dict['nick_name']}\n抓取异常, 已取消抓取该公众号\n", ) Common.logging( log_type, crawler, env, f"公众号:{user_dict['link']}, 站内昵称:{user_dict['nick_name']}\n抓取异常, 已取消抓取该公众号\n", ) AliyunLogger.logging( code="2000", platform=crawler, mode=log_type, env=env, message=f"公众号:{user_dict['link']}, 站内昵称:{user_dict['nick_name']}\n抓取异常, 已取消抓取该公众号\n", ) else: Common.logger(log_type, crawler).warning(f"unbind_msg:{unbind_msg}") Common.logging(log_type, crawler, env, f"unbind_msg:{unbind_msg}") AliyunLogger.logging( code="2000", platform=crawler, mode=log_type, env=env, message=f"unbind_msg: {unbind_msg}", ) return None user_info_dict = { "user_name": r.json()["list"][0]["nickname"], "user_id": r.json()["list"][0]["fakeid"], "avatar_url": r.json()["list"][0]["round_head_img"], } return user_info_dict # 获取腾讯视频下载链接 @classmethod def get_tencent_video_url(cls, video_id): url = "https://h5vv.video.qq.com/getinfo?vid={}&platform=101001&charge=0&otype=json&defn=shd".format( video_id ) 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", } 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"] video_url = folder + name + "?vkey=" + key time.sleep(random.randint(1, 5)) return video_url @classmethod def get_video_url(cls, article_url, env): # 打印请求配置 ca = DesiredCapabilities.CHROME ca["goog:loggingPrefs"] = {"performance": "ALL"} # 不打开浏览器运行 chrome_options = webdriver.ChromeOptions() chrome_options.add_argument("headless") chrome_options.add_argument( f"user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.79 Safari/537.36" ) chrome_options.add_argument("--no-sandbox") # driver初始化 if env == "prod": driver = webdriver.Chrome(desired_capabilities=ca, options=chrome_options) else: driver = webdriver.Chrome( desired_capabilities=ca, options=chrome_options, service=Service( "/Users/wangkun/Downloads/chromedriver/chromedriver_v113/chromedriver" ), ) driver.implicitly_wait(10) driver.get(article_url) time.sleep(1) if ( len( driver.find_elements( By.XPATH, '//div[@class="js_video_poster video_poster"]/*[2]' ) ) != 0 ): video_url = driver.find_element( By.XPATH, '//div[@class="js_video_poster video_poster"]/*[2]' ).get_attribute("src") elif ( len( driver.find_elements( By.XPATH, '//span[@class="js_tx_video_container"]/*[1]' ) ) != 0 ): iframe = driver.find_element( By.XPATH, '//span[@class="js_tx_video_container"]/*[1]' ).get_attribute("src") video_id = iframe.split("vid=")[-1].split("&")[0] video_url = cls.get_tencent_video_url(video_id) else: video_url = 0 driver.quit() if "mpvideo.qpic.cn" in str(video_url): time.sleep(random.randint(1, 3)) return video_url # 获取文章列表 @classmethod def get_videoList( cls, log_type, crawler, task_dict, token_index, rule_dict, user_dict, env ): mq = MQ(topic_name="topic_crawler_etl_" + env) user_info_dict = cls.get_user_info( log_type=log_type, crawler=crawler, task_dict=task_dict, user_dict=user_dict, token_index=token_index, env=env, ) if user_info_dict is None: return user_dict["user_id"] = user_info_dict["user_id"] user_dict["user_name"] = user_info_dict["user_name"] user_dict["avatar_url"] = user_info_dict["avatar_url"] begin = 0 while True: token_dict = cls.get_token(log_type, crawler, token_index, env) url = "https://mp.weixin.qq.com/cgi-bin/appmsg?" headers = { "accept": "*/*", "accept-encoding": "gzip, deflate, br", "accept-language": "zh-CN,zh;q=0.9", "referer": "https://mp.weixin.qq.com/cgi-bin/appmsg?" "t=media/appmsg_edit_v2&action=edit&isNew=1" "&type=77&createType=5&token=" + str(token_dict["token"]) + "&lang=zh_CN", "sec-ch-ua": '" Not A;Brand";v="99", "Chromium";v="100", "Google Chrome";v="100"', "sec-ch-ua-mobile": "?0", "sec-ch-ua-platform": '"Windows"', "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "same-origin", "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" " (KHTML, like Gecko) Chrome/100.0.4896.127 Safari/537.36", "x-requested-with": "XMLHttpRequest", "cookie": token_dict["cookie"], } params = { "action": "list_ex", "begin": str(begin), "count": "5", "fakeid": user_dict["user_id"], "type": "9", "query": "", "token": str(token_dict["token"]), "lang": "zh_CN", "f": "json", "ajax": "1", } urllib3.disable_warnings() r = requests.get(url=url, headers=headers, params=params, verify=False) r.close() if r.json()["base_resp"]["err_msg"] == "invalid session": Common.logger(log_type, crawler).warning( f"status_code:{r.status_code}, get_videoList:{r.text}\n" ) Common.logging( log_type, crawler, env, f"status_code:{r.status_code}, get_videoList:{r.text}\n", ) AliyunLogger.logging( code="2000", platform=crawler, mode=log_type, env=env, message=f"status_code:{r.status_code}, get_videoList:{r.text}\n", ) if 20 >= datetime.datetime.now().hour >= 10: Feishu.bot( log_type, crawler, f"{token_dict['title']}\n操作人:{token_dict['operator']}\n更换日期:{token_dict['update_time']}\n过期啦，请扫码更换token\nhttps://mp.weixin.qq.com/", ) time.sleep(60 * 15) continue if r.json()["base_resp"]["err_msg"] == "freq control": Common.logger(log_type, crawler).warning( f"status_code:{r.status_code}, get_videoList:{r.text}\n" ) Common.logging( log_type, crawler, env, f"status_code:{r.status_code}, get_videoList:{r.text}\n", ) AliyunLogger.logging( code="2000", platform=crawler, mode=log_type, env=env, message=f"status_code:{r.status_code}, get_videoList:{r.text}\n", ) if 20 >= datetime.datetime.now().hour >= 10: Feishu.bot( log_type, crawler, f"{token_dict['title']}\n操作人:{token_dict['operator']}\n更换日期:{token_dict['update_time']} \n频控啦，请扫码更换其他公众号token\nhttps://mp.weixin.qq.com/", ) time.sleep(60 * 15) continue if ( r.json()["base_resp"]["err_msg"] == "invalid args" and r.json()["base_resp"]["ret"] == 200002 ): Common.logger(log_type, crawler).warning( f"status_code:{r.status_code}, get_videoList:{r.text}\n" ) Common.logging( log_type, crawler, env, f"status_code:{r.status_code}, get_videoList:{r.text}\n", ) AliyunLogger.logging( code="2000", platform=crawler, mode=log_type, env=env, message=f"status_code:{r.status_code}, get_videoList:{r.text}\n", ) task_unbind( log_type=log_type, crawler=crawler, taskid=task_dict["id"], uids=str(user_dict["uid"]), env=env, ) if 20 >= datetime.datetime.now().hour >= 10: Feishu.bot( log_type, crawler, f"公众号:{user_dict['link']}, 站内昵称:{user_dict['nick_name']}\n抓取异常, 已取消抓取该公众号\n", ) return if "app_msg_list" not in r.json(): Common.logger(log_type, crawler).warning( f"status_code:{r.status_code}, get_videoList:{r.text}\n" ) Common.logging( log_type, crawler, env, f"status_code:{r.status_code}, get_videoList:{r.text}\n", ) AliyunLogger.logging( code="2000", platform=crawler, mode=log_type, env=env, message=f"status_code:{r.status_code}, get_videoList:{r.text}\n", ) if 20 >= datetime.datetime.now().hour >= 10: Feishu.bot( log_type, crawler, f"{token_dict['title']}\n操作人:{token_dict['operator']}\n更换日期:{token_dict['update_time']}\n频控啦，请扫码更换其他公众号token\nhttps://mp.weixin.qq.com/", ) time.sleep(60 * 15) continue if len(r.json()["app_msg_list"]) == 0: Common.logger(log_type, crawler).info("没有更多视频了\n") Common.logging(log_type, crawler, env, "没有更多视频了\n") AliyunLogger.logging( code="2000", platform=crawler, mode=log_type, env=env, message="没有更多视频了\n", ) return else: begin += 5 app_msg_list = r.json()["app_msg_list"] for article in app_msg_list: try: trace_id = crawler + str(uuid.uuid1()) create_time = article.get("create_time", 0) update_time = article.get("update_time", 0) publish_time_stamp = int(create_time) update_time_stamp = int(update_time) publish_time_str = time.strftime( "%Y-%m-%d %H:%M:%S", time.localtime(publish_time_stamp) ) article_url = article.get("link", "") video_dict = { "video_id": article.get("aid", ""), "video_title": article.get("title", "") .replace(" ", "") .replace('"', "") .replace("'", ""), "publish_time_stamp": publish_time_stamp, "publish_time_str": publish_time_str, "user_name": user_dict["user_name"], "play_cnt": 0, "comment_cnt": 0, "like_cnt": 0, "share_cnt": 0, "user_id": user_dict["user_id"], "avatar_url": user_dict["avatar_url"], "cover_url": article.get("cover", ""), "article_url": article.get("link", ""), "video_url": cls.get_video_url(article_url, env), "session": f"gongzhonghao-author1-{int(time.time())}", } for k, v in video_dict.items(): Common.logger(log_type, crawler).info(f"{k}:{v}") Common.logging( log_type, crawler, env, f"video_dict:{video_dict}" ) AliyunLogger.logging( code="1001", trace_id=trace_id, platform=crawler, mode=log_type, env=env, message="扫描到一条视频", data=video_dict, ) if ( int(time.time()) - publish_time_stamp > 3600 * 24 * int(rule_dict.get("period", {}).get("max", 1000)) ) and ( int(time.time()) - update_time_stamp > 3600 * 24 * int(rule_dict.get("period", {}).get("max", 1000)) ): Common.logger(log_type, crawler).info( f"发布时间超过{int(rule_dict.get('period', {}).get('max', 1000))}天\n" ) Common.logging( log_type, crawler, env, f"发布时间超过{int(rule_dict.get('period', {}).get('max', 1000))}天\n", ) AliyunLogger.logging( code="2004", trace_id=trace_id, platform=crawler, mode=log_type, env=env, data=video_dict, message="发布时间超过{}天".format( int(rule_dict.get("period", {}).get("max", 1000)) ), ) return if ( video_dict["article_url"] == 0 or video_dict["video_url"] == 0 ): Common.logger(log_type, crawler).info("文章涉嫌违反相关法律法规和政策\n") Common.logging(log_type, crawler, env, "文章涉嫌违反相关法律法规和政策\n") AliyunLogger.logging( code="2005", trace_id=trace_id, platform=crawler, mode=log_type, env=env, data=video_dict, message="无效文章或视频", ) # 标题敏感词过滤 elif ( any( str(word) if str(word) in video_dict["video_title"] else False for word in get_config_from_mysql( log_type=log_type, source=crawler, env=env, text="filter", action="", ) ) is True ): Common.logger(log_type, crawler).info("标题已中过滤词\n") Common.logging(log_type, crawler, env, "标题已中过滤词\n") AliyunLogger.logging( code="2003", trace_id=trace_id, platform=crawler, mode=log_type, env=env, data=video_dict, message="标题已中过滤词\n", ) # 已下载判断 elif ( cls.repeat_video( log_type, crawler, video_dict["video_id"], video_dict["video_title"], env, ) != 0 ): Common.logger(log_type, crawler).info("视频已下载\n") Common.logging(log_type, crawler, env, "视频已下载\n") AliyunLogger.logging( code="2002", trace_id=trace_id, platform=crawler, mode=log_type, env=env, data=video_dict, message="视频已下载", ) # 标题相似度 # elif title_like(log_type, crawler, video_dict['video_title'], cls.platform, env) is True: # Common.logger(log_type, crawler).info(f'标题相似度>=80%:{video_dict["video_title"]}\n') # Common.logging(log_type, crawler, env, f'标题相似度>=80%:{video_dict["video_title"]}\n') else: video_dict["out_user_id"] = video_dict["user_id"] video_dict["platform"] = crawler video_dict["strategy"] = log_type video_dict["out_video_id"] = video_dict["video_id"] video_dict["width"] = 0 video_dict["height"] = 0 video_dict["crawler_rule"] = json.dumps(rule_dict) video_dict["user_id"] = user_dict[ "uid" ] # 站内 UID？爬虫获取不到了（随机发布到原 5 个账号中） video_dict["publish_time"] = video_dict["publish_time_str"] mq.send_msg(video_dict) AliyunLogger.logging( code="1002", trace_id=trace_id, platform=crawler, mode=log_type, env=env, data=video_dict, message="成功发送 MQ 至 ETL", ) time.sleep(random.randint(1, 8)) except Exception as e: Common.logger(log_type, crawler).error(f"抓取单条视频异常:{e}\n") Common.logging(log_type, crawler, env, f"抓取单条视频异常:{e}\n") AliyunLogger.logging( code="3000", platform=crawler, mode=log_type, env=env, message=f"抓取单条视频异常:{e}\n", ) Common.logger(log_type, crawler).info("休眠 60 秒\n") Common.logging(log_type, crawler, env, "休眠 60 秒\n") time.sleep(60) @classmethod def repeat_video(cls, log_type, crawler, video_id, video_title, env): # sql = f""" select * from crawler_video where platform="公众号" and out_video_id="{video_id}"; """ sql = f""" select * from crawler_video where platform in ("{crawler}","{cls.platform}") and out_video_id="{video_id}" ; """ # sql = f"""select * from crawler_video where platform = "{crawler}" and (out_video_id="{video_id}" or video_title="{video_title}") ;""" repeat_video = MysqlHelper.get_values(log_type, crawler, sql, env) return len(repeat_video) @classmethod def get_all_videos( cls, log_type, crawler, task_dict, token_index, rule_dict, user_list, env ): total_s = 8 * 60 * 60 # 每个爬虫每天抓取的时间是12h(8h等待+4h抓取） wait_average_time = int((total_s / len(user_list))) for user_dict in user_list: Common.logger(log_type, crawler).info(f'抓取公众号:{user_dict["nick_name"]}\n') Common.logging(log_type, crawler, env, f'抓取公众号:{user_dict["nick_name"]}\n') AliyunLogger.logging( code="1003", platform=crawler, mode=log_type, env=env, message="开始抓取公众号: {}".format(user_dict["nick_name"]), ) try: cls.get_videoList( log_type=log_type, crawler=crawler, task_dict=task_dict, token_index=token_index, rule_dict=rule_dict, user_dict=user_dict, env=env, ) sleep_time = random.randint( wait_average_time - 120, wait_average_time - 60 ) Common.logger(log_type, crawler).info("休眠 {} 秒\n".format(sleep_time)) Common.logging(log_type, crawler, env, "休眠 {} 秒\n".format(sleep_time)) time.sleep(sleep_time) except Exception as e: Common.logger(log_type, crawler).info( f'抓取公众号:{user_dict["nick_name"]}时异常:{e}\n' ) Common.logging( log_type, crawler, env, f'抓取公众号:{user_dict["nick_name"]}时异常:{e}\n' ) AliyunLogger.logging( code="3000", platform=crawler, mode=log_type, env=env, message="抓取公众号: {} 时异常".format(user_dict["nick_name"]), ) AliyunLogger.logging( code="1004", platform=crawler, mode=log_type, env=env, message="完成抓取公众号: {}".format(user_dict["nick_name"]), ) # 判断该账号10天内是否有新进到数据 if __name__ == "__main__": pass