LongArticlesJob/tasks/crawler_accounts_by_association.py

"""
@author: luojunhui
"""

from __future__ import annotations

import json
import datetime
import traceback

from pymysql.cursors import DictCursor
from tqdm import tqdm

from applications import log
from applications.db import DatabaseConnector
from applications.pipeline import scrape_account_entities_process
from applications.utils import Item
from applications.utils import insert_into_candidate_account_pool_table
from coldStartTasks.crawler.baidu import haokan_search_videos
from coldStartTasks.crawler.baidu.baidu_spider import baidu_account_video_crawler
from coldStartTasks.crawler.channels import search_in_wechat_channel
from coldStartTasks.crawler.channels import get_channel_account_videos
from coldStartTasks.crawler.toutiao import get_associated_recommendation
from coldStartTasks.crawler.toutiao import get_toutiao_account_video_list
from coldStartTasks.crawler.wechat import get_article_detail
from coldStartTasks.crawler.wechat import get_article_list_from_account
from coldStartTasks.crawler.wechat import get_source_account_from_article
from config import apolloConfig, long_articles_config

config = apolloConfig()
recommend_cookie = config.getConfigValue("toutiao_detail_recommend_cookie")
blogger_cookie = config.getConfigValue("toutiao_blogger_cookie")


class CrawlerAccounts:

    def __init__(self):
        self.db_client = DatabaseConnector(db_config=long_articles_config)
        self.db_client.connect()

    def get_seed_keys(self) -> list[str]:
        """
        get search keys from database
        """
        fetch_query = f"""
                select association_title
                from `article_pool_promotion_source`
                where association_status = 2 order by association_update_timestamp desc limit 100;
        """
        fetch_response = self.db_client.fetch(query=fetch_query, cursor_type=DictCursor)
        title_list = []
        for item in fetch_response:
            try:
                title_list += json.loads(item['association_title'])
            except Exception as e:
                print(e)
                continue
        return list(set(title_list))

    def insert_video_into_recommend_table(self, item: dict) -> None:
        # whether account exists
        final_item = scrape_account_entities_process(item, self.db_client)
        if not final_item:
            return
        else:
            # save to db
            insert_into_candidate_account_pool_table(
                db_client=self.db_client, account_item=final_item
            )

    def update_account_status(
        self, account_id_tuple: tuple, ori_status: int, new_status: int
    ) -> int:
        update_query = f"""
            update video_association
            set status = %s
            where id in %s and status = %s;
        """
        affected_rows = self.db_client.save(
            query=update_query, params=(new_status, account_id_tuple, ori_status)
        )
        return affected_rows


class ChannelsAccountCrawler(CrawlerAccounts):
    """
    crawler channel accounts
    strategy:
        1. try to get seed titles from database
        2. try to get hot_points from web
        2. use search api to get accounts
    """

    def process_channels_account(
        self, account_name: str, account_id: str, title_list_str: str
    ):
        """
        process video item and save to database
        """

        account_item = Item()
        account_item.add("account_name", account_name)
        account_item.add("account_id", account_id)
        account_item.add("title_list", title_list_str)
        account_item.add("crawler_date", datetime.datetime.today().strftime("%Y-%m-%d"))
        account_item.add("platform", "sph")
        # check item
        account_item.check(source="candidate_account")

        # save to db
        self.insert_video_into_recommend_table(account_item.item)

    def process_search_response(self, video: dict):
        """
        通过搜索视频的账号名称去搜索账号，并且抓取该账号下的第一页视频
        """
        account_name = video["items"][0]["source"]["title"]
        # search account detail
        search_account_response = search_in_wechat_channel(
            search_key=account_name, search_type=2
        )
        account_detail = search_account_response["data"]["data"][0]["items"][0]
        account_id = account_detail["jumpInfo"]["userName"]

        # fetch account video list for the first page
        search_video_response = get_channel_account_videos(account_id)
        search_response_data = search_video_response["data"]
        search_response_data_type = type(search_response_data)
        if search_response_data_type == dict:
            video_list = search_response_data["object"]
        elif search_response_data_type == list:
            video_list = search_response_data
        else:
            raise RuntimeError("search video response type error")

        title_list = [i["objectDesc"]["description"] for i in video_list]
        title_list_str = json.dumps(title_list, ensure_ascii=False)
        self.process_channels_account(account_name, account_id, title_list_str)

    def search_video_in_channels(self, title: str) -> None:
        """
        search
        """
        search_response = search_in_wechat_channel(search_key=title, search_type=1)
        video_list = search_response["data"]["data"][0]["subBoxes"]
        for video in tqdm(video_list, desc="crawler each video"):
            try:
                self.process_search_response(video)
            except Exception as e:
                log(
                    task="crawler_channels_account_videos",
                    function="process_search_response",
                    message="search by title failed",
                    data={
                        "video": video,
                        "error": str(e),
                        "traceback": traceback.format_exc(),
                    },
                )

    def deal(self):
        seed_title_list = self.get_seed_keys()
        for seed_title in tqdm(seed_title_list, desc="crawler each title"):
            try:
                self.search_video_in_channels(title=seed_title)
            except Exception as e:
                log(
                    task="crawler_channels_account_videos",
                    function="search_video_in_channels",
                    message="search video in channels failed",
                    data={
                        "title": seed_title,
                        "error": str(e),
                        "traceback": traceback.format_exc(),
                    },
                )


class ToutiaoAccountCrawler(CrawlerAccounts):

    def get_seed_videos(self):
        fetch_query = f"""
            select out_account_name, article_title, url_unique_md5 
            from publish_single_video_source
            where platform = 'toutiao' and video_pool_audit_status = 1 and bad_status = 0
            order by score desc limit 100;
        """
        seed_video_list = self.db_client.fetch(
            query=fetch_query, cursor_type=DictCursor
        )
        return seed_video_list

    def get_level_up_videos(self):
        fetch_query = f"""
            select out_account_name, article_title, url_unique_md5
            from publish_single_video_source
            where platform = 'toutiao' and (
                article_title in (
                    select distinct(title) from article_pool_promotion_source where status = 1 and deleted = 0
                )
                    or flow_pool_level < 4
                );
        """
        uplevel_video_list = self.db_client.fetch(
            query=fetch_query, cursor_type=DictCursor
        )
        return uplevel_video_list

    def process_toutiao_account(self, video):

        # process video item and save to database
        account_item = Item()
        user_info = video["user_info"]
        account_item.add("account_name", user_info["name"])
        account_item.add("account_id", user_info["user_id"])
        account_item.add("platform", "toutiao")
        account_item.add("crawler_date", datetime.datetime.today().strftime("%Y-%m-%d"))

        # fetch account video first page video list
        fetch_response = get_toutiao_account_video_list(
            account_id=user_info["user_id"], cookie=blogger_cookie
        )
        video_list = fetch_response["data"]
        title_list = [i["title"] for i in video_list]
        title_list_str = json.dumps(title_list, ensure_ascii=False)
        account_item.add("title_list", title_list_str)

        # check item
        account_item.check(source="candidate_account")

        # insert into database
        self.insert_video_into_recommend_table(account_item.item)

    def get_recommend_video_list(self, seed_video: dict):

        # get recommend videos for each video
        seed_video_id = seed_video["url_unique_md5"]
        recommend_response = get_associated_recommendation(
            seed_video_id, recommend_cookie
        )
        recommend_video_list = recommend_response["data"]
        for video in tqdm(recommend_video_list):
            try:
                self.process_toutiao_account(video)

            except Exception as e:
                log(
                    task="toutiao account crawler",
                    function="process_toutiao_video",
                    message="get recommend video failed",
                    data={
                        "video": video,
                        "error": str(e),
                        "traceback": traceback.format_exc(),
                    },
                )

    def get_category_recommend_list(self):
        """
        品类推荐流几乎无视频，暂时不做
        """
        return NotImplementedError()

    def deal(self):

        # start
        # seed_video_list = self.get_seed_videos()
        seed_video_list = self.get_level_up_videos()
        for seed_video in tqdm(seed_video_list, desc="get each video recommendation"):
            try:
                self.get_recommend_video_list(seed_video)
            except Exception as e:
                log(
                    task="toutiao_recommendation_crawler",
                    function="save_each_recommendation",
                    message="save recommendation failed",
                    data={
                        "error": str(e),
                        "traceback": traceback.format_exc(),
                        "seed_video": seed_video,
                    },
                )


class HaoKanAccountCrawler(CrawlerAccounts):

    def process_haokan_video(self, video: dict) -> None:
        """
        process_haokan_video
        """

        account_item = Item()
        account_item.add("account_name", video["author"])
        account_item.add("account_id", video["author_id"])
        account_item.add("platform", "hksp")
        account_item.add("crawler_date", datetime.datetime.today().strftime("%Y-%m-%d"))

        # fetch account video first page video list
        fetch_response = baidu_account_video_crawler(account_id=video["author_id"])
        video_list = fetch_response["results"]
        title_list = [i["content"]["title"] for i in video_list]
        title_list_str = json.dumps(title_list, ensure_ascii=False)
        account_item.add("title_list", title_list_str)

        # check item
        account_item.check(source="candidate_account")

        # insert into database
        self.insert_video_into_recommend_table(account_item.item)

    def search_videos_in_haokan_video(self, title: str) -> None:
        """
        search_
        """
        search_response = haokan_search_videos(title)
        video_list = search_response["data"]["list"]
        for video in tqdm(video_list, desc="search videos"):
            try:
                self.process_haokan_video(video)
            except Exception as e:
                log(
                    task="haokan_search_crawler",
                    function="process_haokan_video",
                    message="process haokan video failed",
                    data={
                        "video": video,
                        "error": str(e),
                        "traceback": traceback.format_exc(),
                    },
                )

    def deal(self):
        seed_title_list = self.get_seed_keys()
        for seed_title in tqdm(seed_title_list, desc="crawler each title"):
            try:
                self.search_videos_in_haokan_video(seed_title)
            except Exception as e:
                log(
                    task="haokan_search_crawler",
                    function="search_videos_in_haokan_video",
                    message="search videos in haokan video failed",
                    data={
                        "title": seed_title,
                        "error": str(e),
                        "traceback": traceback.format_exc(),
                    },
                )


class GzhAccountCrawler(CrawlerAccounts):

    def get_task_list(self):
        fetch_query = f"""
            select id, article_url
            from publish_single_video_source
            where source_account = 1 and platform = 'gzh' limit 10;
        """
        task_list = self.db_client.fetch(query=fetch_query, cursor_type=DictCursor)
        return task_list

    def process_official_account(self, account_name, account_id):
        """
        process_official_account
        """
        account_item = Item()
        account_item.add("account_name", account_name)
        account_item.add("account_id", account_id)
        account_item.add("platform", "gzh")
        account_item.add("crawler_date", datetime.datetime.today().strftime("%Y-%m-%d"))

        # fetch account video first page video list
        fetch_response = get_article_list_from_account(account_id=account_id, index=None)
        msg_list = fetch_response["data"]["data"]
        title_list = []
        for msg in msg_list:
            sub_title_list = [i['Title'] for i in msg['AppMsg']['DetailInfo']]
            if len(title_list) > 10:
                continue
            else:
                title_list += sub_title_list

        title_list_str = json.dumps(title_list, ensure_ascii=False)
        account_item.add("title_list", title_list_str)

        # check item
        account_item.check(source="candidate_account")

        # insert into database
        self.insert_video_into_recommend_table(account_item.item)

    def extract_account_from_article_link(self, article_link):
        """
        try to get account info from article link
        """
        # is article link original
        article_detail = get_article_detail(article_link)
        is_original = article_detail["data"]["data"]["is_original"]

        if is_original:
            return
        # extract source account
        source_account = get_source_account_from_article(article_link)
        if not source_account:
            return
        else:
            account_name = source_account['name']
            gh_id = source_account['gh_id']
            self.process_official_account(account_name, gh_id)

    def update_crawler_article_status(self, article_id_tuple: tuple):
        """
        update crawler article status
        """
        update_query = f"""
            update publish_single_video_source
            set source_account = %s
            where id in %s;
        """
        affected_rows = self.db_client.save(
            query=update_query, params=(0, article_id_tuple)
        )
        return affected_rows

    def deal(self):
        task_list = self.get_task_list()
        task_id_list = []
        for crawler_article_obj in tqdm(task_list, desc="crawler article list"):
            article_url = crawler_article_obj['article_url']
            article_id = crawler_article_obj['id']
            task_id_list.append(int(article_id))
            try:
                self.extract_account_from_article_link(article_url)

            except Exception as e:
                log(
                    task="gzh_account_crawler",
                    function="extract_account_from_article_link",
                    message="extract account from article link failed",
                    data={
                        "article_url": article_url,
                        "error": str(e),
                        "traceback": traceback.format_exc(),
                    },
                )

        if task_id_list:
            article_id_tuple = tuple(task_id_list)
            affected_rows = self.update_crawler_article_status(article_id_tuple)
            print(affected_rows)