LongArticlesJob/coldStartTasks/crawler/weixin_account_crawler.py

"""
@author: luojunhui
"""
import time
import traceback
from typing import List, Set, Dict, Tuple

from tqdm import tqdm
from pymysql.cursors import DictCursor

from applications import WeixinSpider, longArticlesMySQL, log, bot, aiditApi
from applications.const import WeixinVideoCrawlerConst
from applications.functions import Functions

const = WeixinVideoCrawlerConst()
function = Functions()


def get_inner_account_gh_id() -> Set[str]:
    """
    获取内部账号名称
    :return:
    """
    accounts = aiditApi.get_publish_account_from_aigc()
    gh_id_list = [i['ghId'] for i in accounts]
    return set(gh_id_list)


class WeixinAccountCrawler(object):
    """
    账号抓取
    """

    def __init__(self):
        self.db_client = longArticlesMySQL()
        self.spider = WeixinSpider()
        self.crawler_account_count = 0

    def get_crawler_articles(self) -> List[Dict]:
        """
        获取已经抓取到的文章，判断其是否有链接账号，若有则继续抓账号
        :return:
        """
        sql = f"""
            SELECT id, article_url
            FROM publish_single_video_source
            WHERE source_account = {const.NEED_SCAN_SOURCE_ACCOUNT};
        """
        article_url_list = self.db_client.select(sql, cursor_type=DictCursor)
        return article_url_list

    def update_crawler_article_status(self, article_id_tuple: Tuple[int, ...]) -> int:
        """
        :param article_id_tuple:
        :return:
        """
        sql = """
            UPDATE publish_single_video_source
            SET source_account = %s
            WHERE id IN %s;
            """
        affected_rows = self.db_client.update(sql, (const.DO_NOT_NEED_SOURCE_ACCOUNT, article_id_tuple))
        return affected_rows

    def get_seed_titles(self, run_date) -> List[str]:
        """
        :return:
        """
        publish_timestamp_threshold = int(run_date.timestamp()) - const.STAT_PERIOD
        sql = f"""
            SELECT distinct title
            FROM datastat_sort_strategy
            WHERE read_rate > {const.READ_AVG_MULTIPLE} and view_count > {const.MIN_READ_COUNT} and publish_timestamp > {publish_timestamp_threshold}
            ORDER BY read_rate DESC;
        """
        title_list = self.db_client.select(sql, cursor_type=DictCursor)
        title_list = [i['title'] for i in title_list]
        return title_list

    def is_original(self, article_url: str) -> bool:
        """
        判断视频是否是原创
        :return:
        """
        response = self.spider.get_article_text(article_url)
        data = response['data']['data']
        return data['is_original']

    def insert_account(self, gh_id: str, account_name: str) -> int:
        """
        插入账号
        :param account_name:
        :param gh_id:
        :return:
        """
        init_date = time.strftime("%Y-%m-%d", time.localtime())
        sql = """
            INSERT IGNORE INTO weixin_account_for_videos
            (gh_id, account_name, account_init_date)
            VALUES 
            (%s, %s, %s);
        """
        insert_rows = self.db_client.update(sql, (gh_id, account_name, init_date))
        return insert_rows

    def process_search_result(self, response: Dict, inner_account_set: Set[str]):
        """
        处理搜索结果
        :param response:
        :param inner_account_set:
        :return:
        """
        if response['code'] != const.REQUEST_SUCCESS:
            return

        article_list = response['data']['data']
        if article_list:
            for article in article_list:
                try:
                    # 先判断账号是否内部账号
                    article_url = article['url']
                    account_detail = self.spider.get_account_by_url(article_url)
                    account_detail = account_detail['data']['data']
                    account_name = account_detail['account_name']
                    gh_id = account_detail['wx_gh']
                    if gh_id in inner_account_set:
                        continue
                    # 判断搜索结果是否原创
                    if self.is_original(article_url):
                        continue

                    # 判断是否有视频链接
                    try:
                        video_url = function.get_video_url(article_url)
                    except Exception as e:
                        continue

                    if not video_url:
                        continue

                    # 将账号抓取进来
                    insert_rows = self.insert_account(gh_id=gh_id, account_name=account_name)
                    if insert_rows:
                        log(
                            task="account_crawler_v1",
                            function="process_search_result",
                            message="insert account success",
                            data={
                                "gh_id": gh_id,
                                "account_name": account_name
                            }
                        )
                        self.crawler_account_count += 1
                except Exception as e:
                    log(
                        task="account_crawler_v1",
                        function="process_search_result",
                        message="insert account error",
                        data={
                            "error": str(e),
                            "traceback": traceback.format_exc(),
                            "data": article
                        }
                    )

    def search_title_in_weixin(self, title: str, inner_account_set: Set[str]) -> None:
        """
        调用搜索接口，在微信搜索
        :param inner_account_set:
        :param title:
        :return:
        """
        for page_index in tqdm(range(1, const.MAX_SEARCH_PAGE_NUM + 1), desc='searching: {}'.format(title)):
            try:
                response = self.spider.search_articles(title, page=str(page_index))
                self.process_search_result(response, inner_account_set)
                time.sleep(const.SLEEP_SECONDS)
            except Exception as e:
                log(
                    task="account_crawler_v1",
                    function="search_title_in_weixin",
                    message="search title error",
                    data={
                        "error": str(e),
                        "traceback": traceback.format_exc(),
                        "title": title
                    }
                )

    def run(self, run_date) -> None:
        """
        入口函数
        :return:
        """
        # get seed titles
        title_list = self.get_seed_titles(run_date)
        # get inner accounts set
        inner_account_gh_id_set = get_inner_account_gh_id()

        start_time = time.time()
        for title in tqdm(title_list, desc="search each title"):
            self.search_title_in_weixin(title, inner_account_gh_id_set)

        # 通知
        bot(
            title="微信账号抓取V1完成",
            detail={
                "总更新账号数量": self.crawler_account_count,
                "总耗时": time.time() - start_time,
                "种子标题数量": len(title_list)
            },
            mention=False
        )

    def run_v2(self) -> None:
        """
        入口函数
        :return:
        """
        # get article list
        crawler_article_list = self.get_crawler_articles()
        article_id_list = []
        insert_account_count = 0
        for crawler_article_obj in tqdm(crawler_article_list, desc="crawler article list"):
            try:
                article_id = crawler_article_obj['id']
                # 记录处理过的id
                article_id_list.append(int(article_id))

                article_url = crawler_article_obj['article_url']
                # 判断文章是否原创
                if self.is_original(article_url):
                    continue
                try:
                    source_account_info = function.get_source_account(article_url)
                except Exception as e:
                    continue
                if not source_account_info:
                    continue
                if source_account_info:
                    account_name = source_account_info['name']
                    gh_id = source_account_info['gh_id']
                    affected_rows = self.insert_account(gh_id=gh_id, account_name=account_name)
                    insert_account_count += affected_rows
                else:
                    continue

            except Exception as e:
                print(e)
                print(traceback.format_exc())

        if article_id_list:
            article_id_tuple = tuple(article_id_list)
            affected_rows = self.update_crawler_article_status(article_id_tuple)

        bot(
            title="微信账号抓取V2完成",
            detail={
                "扫描文章数量": len(crawler_article_list),
                "新增账号数量": insert_account_count
            },
            mention=False
        )