LongArticlesJob/tasks/crawler_accounts_by_association.py

"""
@author: luojunhui
"""

from __future__ import annotations

import datetime
import traceback
import numpy as np

from pymysql.cursors import DictCursor
from tqdm import tqdm

from applications import log
from applications.api import similarity_between_title_list
from applications.db import DatabaseConnector
from applications.pipeline import scrape_account_entities_process
from applications.utils import Item
from applications.utils import insert_into_associated_recommendation_table
from coldStartTasks.crawler.baidu import haokan_search_videos
from coldStartTasks.crawler.toutiao import get_associated_recommendation
from coldStartTasks.crawler.channels import search_in_wechat_channel
from coldStartTasks.crawler.channels import get_channel_account_videos
from config import apolloConfig, long_articles_config

config = apolloConfig()
cookie = config.getConfigValue("toutiao_detail_recommend_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[dict]:
        """
        get search keys from database
        """
        fetch_query = "select title from article_pool_promotion_source where status = 1 and deleted = 0 order by level limit 100;"
        result = self.db_client.fetch(query=fetch_query, cursor_type=DictCursor)
        return result

    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_associated_recommendation_table(
                db_client=self.db_client, associated_recommendation_item=final_item
            )

    def save_similarity_score_to_table(self, association_list: list[dict]) -> int:
        """
        calculate similarity between seed_title_list and association_title_list
        """
        association_id_list = [i["id"] for i in association_list]
        association_title_list = [i["title"] for i in association_list]
        seed_title_list = [i["seed_title"] for i in association_list]
        similarity_score_list = similarity_between_title_list(
            seed_title_list, association_title_list
        )
        similarity_score_array = np.array(similarity_score_list)

        # get main diagonal score
        score_list = np.diag(similarity_score_array)

        batch_update_query = """
            update video_association
            set score = case id
                {}
            end
            where id in %s and score is null;
        """
        case_statement = []
        params = []
        for index, score in enumerate(score_list):
            association_id = association_id_list[index]
            case_statement.append(f"when %s then %s")
            params.extend([association_id, score])

        params.append(tuple(association_id_list))
        case_statements = "\n".join(case_statement)
        formatted_sql = batch_update_query.format(case_statements)
        affected_rows = self.db_client.save(formatted_sql, params)
        return affected_rows

    def get_video_list_without_score(self):
        fetch_query = f"""
            select id, title, seed_title
            from video_association
            where score is null;
        """
        fetch_response = self.db_client.fetch(query=fetch_query, cursor_type=DictCursor)
        return fetch_response

    def get_video_list_with_score(self, platform: str):
        """
        find video from video association
        """
        fetch_query = f"""
            select id, account_name, recommend_video_id, title, read_cnt, duration, seed_account, seed_title
            from video_association 
            where score > %s and platform = %s and status = %s
            order by account_name;
        """
        fetch_response = self.db_client.fetch(query=fetch_query, params=(0.5, platform, 0))
        return fetch_response

    def update_video_status(self, video_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, video_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_video(self, video: dict, seed_title: str, account_name: str, account_id: str):
        """
        process video item and save to database
        """

        video_item = Item()
        video_item.add("account_name", account_name)
        video_item.add("account_id", account_id)
        video_item.add("recommend_video_id", video["id"])
        video_item.add("title", video["objectDesc"]["description"])
        video_item.add("duration", video["objectDesc"]["media"][0]["VideoPlayLen"])
        video_item.add("seed_account", "SearchWithOutAccount")
        video_item.add("seed_title", seed_title)
        video_item.add(
            "recommend_date", datetime.datetime.today().strftime("%Y-%m-%d")
        )
        video_item.add("platform", "sph")
        # check item
        video_item.check(source="association")

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

    def process_search_response(self, video: dict, seed_title: str):
        """
        通过搜索视频的账号名称去搜索账号，并且抓取该账号下的第一页视频
        """
        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
        search_video_response = get_channel_account_videos(account_id)
        video_list = search_video_response["data"]["object"]

        # process and insert each video
        for video in video_list:
            try:
                self.process_channels_video(video, seed_title, account_name, account_id)
            except Exception as e:
                log(
                    task="crawler_channels_account_videos",
                    function="process_channels_video",
                    message="process video failed",
                    data={
                        "video": video,
                        "error": str(e),
                        "traceback": traceback.format_exc()
                    }
                )

    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, seed_title=title)
            except Exception as e:
                log(
                    task="channels account crawler",
                    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 item in tqdm(seed_title_list, desc="crawler each title"):
            try:
                self.search_video_in_channels(title=item["title"])
            except Exception as e:
                log(
                    task="channels account crawler",
                    function="search_video_in_channels",
                    message="search video in channels failed",
                    data={
                        "title": item["title"],
                        "error": str(e),
                        "traceback": traceback.format_exc()
                    }
                )

        # cal similarity score
        video_list = self.get_video_list_without_score()
        affected_rows = self.save_similarity_score_to_table(video_list)
        print(affected_rows)


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 process_toutiao_video(self, video, seed_account_name, seed_title):

        # process video item and save to database
        video_item = Item()
        user_info = video["user_info"]
        video_item.add("account_name", user_info["name"])
        video_item.add("account_id", user_info["user_id"])
        video_item.add("platform", "toutiao")
        video_item.add("recommend_video_id", video["id"])
        video_item.add("title", video["title"])
        video_item.add("read_cnt", video.get("read_count"))
        video_item.add("duration", video["video_duration"])
        video_item.add("seed_account", seed_account_name)
        video_item.add("seed_title", seed_title)
        video_item.add("recommend_date", datetime.datetime.today().strftime("%Y-%m-%d"))
        # check item
        video_item.check(source="association")

        # insert into database
        self.insert_video_into_recommend_table(video_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"]
        seed_account_name = seed_video["out_account_name"]
        seed_title = seed_video["article_title"]
        recommend_response = get_associated_recommendation(seed_video_id, cookie)
        recommend_video_list = recommend_response["data"]
        for video in tqdm(recommend_video_list):
            try:
                self.process_toutiao_video(video, seed_account_name, seed_title)

            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()
        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,
                    },
                )

        # cal similarity score
        video_list = self.get_video_list_without_score()
        affected_rows = self.save_similarity_score_to_table(video_list)
        print(affected_rows)


class HaoKanAccountCrawler(CrawlerAccounts):

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

        video_item = Item()
        video_item.add("account_name", video['author'])
        video_item.add("account_id", video['author_id'])
        video_item.add("platform", "hksp")
        video_item.add("recommend_video_id", video['vid'])
        video_item.add("title", video['title'])
        read_num_string = video['read_num'].replace("次播放", "")
        if "万" in read_num_string:
            read_num_string = read_num_string.replace("万", "")
            read_num = int(float(read_num_string) * 10000)
        else:
            read_num = int(read_num_string)
        video_item.add("read_cnt", int(read_num))
        duration_string = video['duration']
        duration_list = duration_string.split(":")
        if len(duration_list) > 2:
            # video too long
            return
        duration = int(duration_list[0]) * 60 + int(duration_list[1])
        video_item.add("duration", duration)
        video_item.add("seed_account", "SearchWithOutAccount")
        video_item.add("seed_title", seed_title)
        video_item.add("recommend_date", datetime.datetime.today().strftime("%Y-%m-%d"))
        # check item
        video_item.check(source="association")

        # insert into database
        self.insert_video_into_recommend_table(video_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, seed_title=title)
            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["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["title"],
                        "error": str(e),
                        "traceback": traceback.format_exc()
                    }
                )

        video_list = self.get_video_list_without_score()
        affected_rows = self.save_similarity_score_to_table(video_list)
        print(affected_rows)