piaoquan_crawler/gongzhonghao/gongzhonghao_author/gongzhonghao_author.py

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'], 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, 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}"; """
        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):
        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)
                Common.logger(log_type, crawler).info('休眠 60 秒\n')
                Common.logging(log_type, crawler, env, '休眠 60 秒\n')
                time.sleep(60)
            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'])
            )


if __name__ == "__main__":
    pass