LongArticlesJob/coldStartTasks/crawler/weixin_video_crawler.py

"""
@author: luojunhui
抓取视频
"""
import json
import time
import traceback
from typing import List, Dict

from tqdm import tqdm
from pymysql.cursors import DictCursor

from config import apolloConfig
from applications import bot
from applications import log
from applications import Functions
from applications import WeixinSpider
from applications import longArticlesMySQL
from applications.const import WeixinVideoCrawlerConst
from coldStartTasks.filter import video_crawler_duplicate_filter

spider = WeixinSpider()
functions = Functions()
config = apolloConfig(env="prod")
const = WeixinVideoCrawlerConst()


class WeixinVideoCrawler(object):
    """
    微信视频抓取
    """

    def __init__(self):
        self.db_client = longArticlesMySQL()
        self.festival_list = json.loads(config.getConfigValue("festival"))

    def is_festival(self, title: str) -> bool:
        """
        判断是否为节假日
        :param title:
        :return:
        """
        for festival in self.festival_list:
            if festival in title:
                return True
        return False

    def get_title_status(self, title: str) -> int:
        """
        通过标题获取文章状态
        :param title:
        :return:
        """
        if self.is_festival(title):
            return const.TITLE_FESTIVAL_STATUS
        elif len(title) < const.TITLE_MIN_LENGTH:
            return const.TITLE_SHORT_STATUS
        else:
            return const.TITLE_DEFAULT_STATUS

    def update_account_latest_crawler_timestamp(self, gh_id: str) -> int:
        """
        更新最新抓取时间戳
        :param gh_id:
        :return:
        """
        update_sql = f"""
            UPDATE weixin_account_for_videos
            SET latest_crawler_timestamp = (
                SELECT max(publish_timestamp) 
                FROM publish_single_video_source 
                WHERE out_account_id = %s
                )
            WHERE gh_id = %s;
        """
        affected_rows = self.db_client.update(
            sql=update_sql,
            params=(gh_id, gh_id)
        )
        return affected_rows

    def get_crawler_accounts(self) -> List[Dict]:
        """
        获取微信公众号列表
        :return:
        """
        select_sql = f"""
            SELECT gh_id, account_name, latest_crawler_timestamp
            FROM weixin_account_for_videos
            WHERE status = {const.ACCOUNT_CRAWL_STATUS}
            ORDER BY latest_crawler_timestamp;
        """
        response = self.db_client.select(select_sql, DictCursor)
        return response

    def crawler_article_video_list(self, account_obj: Dict, cursor=None):
        """
        抓取单个账号的文章列表，获取视频
        :param cursor:
        :param account_obj:
        :return: 返回待下载的视频列表
        """
        gh_id = account_obj["gh_id"]
        account_name = account_obj["account_name"]
        latest_crawler_timestamp = account_obj["latest_crawler_timestamp"]
        if latest_crawler_timestamp is None:
            latest_crawler_timestamp = const.DEFAULT_TIMESTAMP
        # 调用爬虫接口
        response = spider.update_msg_list(gh_id, index=cursor)
        if response['code'] == const.REQUEST_SUCCESS:
            # 一般返回最近10天的msg_list
            msg_list = response.get('data', {}).get("data", [])
            if msg_list:
                last_msg = msg_list[-1]
                last_msg_base_info = last_msg['AppMsg']['BaseInfo']
                last_msg_create_timestamp = last_msg_base_info['CreateTime']
                self.insert_msg_list(account_name=account_name, gh_id=gh_id, msg_list=msg_list)
                if last_msg_create_timestamp > latest_crawler_timestamp:
                    next_cursor = response['data']['next_cursor']
                    return self.crawler_article_video_list(account_obj=account_obj, cursor=next_cursor)
            else:
                return []
        else:
            return []
        return []

    def is_downloaded(self, url_unique: str) -> bool:
        """
        判断该视频是否已经下载
        :param url_unique:
        :return:
        """
        select_sql = f"""
            SELECT id
            FROM publish_single_video_source
            WHERE url_unique_md5 = '{url_unique}';
        """
        response = self.db_client.select(select_sql)
        if response:
            return True
        else:
            return False

    def insert_msg_list(self, account_name, gh_id, msg_list: List[Dict]) -> None:
        """
        插入视频信息
        :param gh_id:
        :param account_name:
        :param msg_list:
        :return:
        """
        for info in msg_list:
            create_time = info.get("AppMsg", {}).get("BaseInfo", {}).get("CreateTime", None)
            publish_type = info.get("AppMsg", {}).get("BaseInfo", {}).get("Type", None)
            detail_article_list = info.get("AppMsg", {}).get("DetailInfo", [])
            if detail_article_list:
                for article in tqdm(detail_article_list, desc="{}: crawler_in_msg_list".format(account_name)):
                    article_url = article.get("ContentUrl", None)
                    url_unique = functions.generateGzhId(article_url)
                    # 判断该视频链接是否下载，若已经下载则直接跳过
                    if self.is_downloaded(url_unique):
                        print("url exists")
                        continue

                    title = article.get("Title", None)
                    if not title:
                        continue

                    # 判断标题是否重复
                    if video_crawler_duplicate_filter(title, self.db_client):
                        log(
                            task='weixin_video_crawler',
                            function="insert_msg_list",
                            message="标题去重",
                            data={"url": article_url}
                        )
                        continue

                    try:
                        download_path = functions.download_gzh_video(article_url)
                        if download_path:
                            oss_path = functions.upload_to_oss(local_video_path=download_path)
                            position = article.get("ItemIndex", None)
                            cover_url = article.get("CoverImgUrl", None)
                            show_desc = article.get("ShowDesc", None)
                            show_stat = functions.show_desc_to_sta(show_desc)
                            read_cnt = show_stat.get("show_view_count", 0)
                            like_cnt = show_stat.get("show_like_count", 0)
                            title_status = self.get_title_status(title)
                            insert_sql = f"""
                                INSERT INTO publish_single_video_source
                                (content_trace_id, article_title, out_account_id, out_account_name, read_cnt, like_cnt, article_index, article_publish_type, article_url, cover_url, video_oss_path, bad_status, publish_timestamp, crawler_timestamp, url_unique_md5)
                                values
                                (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s);
                            """
                            try:
                                self.db_client.update(
                                    sql=insert_sql,
                                    params=(
                                        "video" + url_unique,
                                        title,
                                        gh_id,
                                        account_name,
                                        read_cnt,
                                        like_cnt,
                                        position,
                                        publish_type,
                                        article_url,
                                        cover_url,
                                        oss_path,
                                        title_status,
                                        create_time,
                                        int(time.time()),
                                        url_unique
                                    )
                                )
                                log(
                                    task='weixin_video_crawler',
                                    function="insert_msg_list",
                                    message="插入一条视频",
                                    data={"account_name": account_name, "url": article_url}
                                )
                            except Exception as e:
                                try:
                                    update_sql = f"""
                                        UPDATE publish_single_video_source
                                        SET read_cnt = %s, like_cnt = %s
                                        WHERE url_unique_md5 = %s;
                                    """
                                    self.db_client.update(
                                        sql=update_sql,
                                        params=(read_cnt, like_cnt, functions.generateGzhId(article_url))
                                    )
                                except Exception as e:
                                    error_stack = traceback.format_exc()
                                    log(
                                        task='weixin_video_crawler',
                                        function="update_msg_list",
                                        status="fail",
                                        message="更新内容失败",
                                        data={"error": str(e), "error_stack": error_stack, "url": article_url}

                                    )
                        else:
                            continue
                    except Exception as e:
                        error_stack = traceback.format_exc()
                        log(
                            task='weixin_video_crawler',
                            function="update_msg_list",
                            status="fail",
                            message="更新内容失败",
                            data={"error": str(e), "error_stack": error_stack, "url": article_url}

                        )

    def crawler_task(self):
        """
        抓取任务
        :return:
        """
        account_list = self.get_crawler_accounts()
        for account_obj in tqdm(account_list, desc="crawler_video_for_each_account"):
            try:
                self.crawler_article_video_list(account_obj)
                self.update_account_latest_crawler_timestamp(gh_id=account_obj["gh_id"])
                time.sleep(const.SLEEP_SECONDS)
            except Exception as e:
                error_stack = traceback.format_exc()
                log(
                    task='weixin_video_crawler',
                    function="crawler_task",
                    status="fail",
                    message="抓取任务失败--单账号",
                    data={"error": str(e), "error_stack": error_stack, "account_name": account_obj["account_name"]}
                )

    def mention(self, start_timestamp):
        """
        飞书发送消息
        :param start_timestamp:
        :return:
        """
        sql = f"""select count(1) from publish_single_video_source where crawler_timestamp >= {start_timestamp};"""
        response = self.db_client.select(sql)
        new_articles_count = response[0][0]
        bot(
            title='微信抓取任务执行完成',
            detail={
                "新增视频数量": new_articles_count
            }
        )

    def run(self):
        """
        执行任务
        :return:
        """
        start_timestamp = int(time.time())
        self.crawler_task()
        self.mention(start_timestamp)