LongArticlesJob/applications/utils/fetch_info_from_aigc.py

"""
fetch info from aigc database system
"""
from collections import defaultdict
from typing import List, Dict

from pymysql.cursors import DictCursor


def fetch_publishing_account_list(db_client) -> List[Dict]:
    """
    fetch account_list from aigc database
    """
    fetch_sql = f"""
        SELECT DISTINCT
            t3.`name` as account_name,
            t3.gh_id as gh_id,
            t3.follower_count as fans,
            t6.account_source_name as account_source,
            t6.mode_type as mode_type,
            t6.account_type as account_type,
            t6.`status` as status
        FROM
            publish_plan t1
            JOIN publish_plan_account t2 ON t1.id = t2.plan_id
            JOIN publish_account t3 ON t2.account_id = t3.id
            LEFT JOIN publish_account_wx_type t4 on t3.id = t4.account_id
            LEFT JOIN wx_statistics_group_source_account t5 on t3.id = t5.account_id
            LEFT JOIN wx_statistics_group_source t6 on t5.group_source_name = t6.account_source_name
        WHERE
            t1.plan_status = 1
            AND t3.channel = 5
            GROUP BY t3.id;
    """
    account_list = db_client.fetch(
        query=fetch_sql,
        cursor_type=DictCursor
    )
    return account_list

def fetch_account_fans(db_client, start_date: str) -> Dict:
    """
    fetch account fans from aigc database
    """
    sql = f"""
        SELECT t1.date_str, t1.fans_count, t2.gh_id
        FROM datastat_wx t1 JOIN publish_account t2 ON t1.account_id = t2.id
        WHERE t2.channel = 5
        AND t1.fans_count > 0
        AND t1.date_str >= '{start_date}' 
        ORDER BY t1.date_str;
        """
    result = db_client.fetch(sql)
    fans_dict = defaultdict(dict)
    for dt, fans, gh_id in result:
        fans_dict.setdefault(gh_id, {})[dt] = fans
    return fans_dict