LongArticlesJob/updatePublishedMsgDaily.py

"""
@author: luojunhui
@description: update daily information into official articles v2
"""

import time
import json
import threading

import schedule
from tqdm import tqdm
from datetime import datetime

from applications import PQMySQL, WeixinSpider, Functions, log, bot


def getAccounts():
    """
    获取账号信息
    :return: [{}, {},...], [{}, {}, {}...]
    """
    with open("config/accountInfoV0914.json", encoding="utf-8") as f:
        account_list = json.loads(f.read())
    subscription_account = [i for i in account_list if i['type'] == '订阅号']
    server_account = [i for i in account_list if i['type'] == '服务号']
    return subscription_account, server_account


def insertEachMsg(db_client, gh_id, account_name, msg_list):
    """
    把消息数据更新到数据库中
    :param db_client:
    :param account_name:
    :param gh_id:
    :param msg_list:
    :return:
    """
    for info in msg_list:
        baseInfo = info.get("BaseInfo", {})
        appMsgId = info.get("AppMsg", {}).get("BaseInfo", {}).get("AppMsgId", None)
        createTime = info.get("AppMsg", {}).get("BaseInfo", {}).get("CreateTime", None)
        updateTime = info.get("AppMsg", {}).get("BaseInfo", {}).get("UpdateTime", None)
        Type = info.get("AppMsg", {}).get("BaseInfo", {}).get("Type", None)
        detail_article_list = info.get("AppMsg", {}).get("DetailInfo", [])
        if detail_article_list:
            for article in detail_article_list:
                title = article.get("Title", None)
                Digest = article.get("Digest", None)
                ItemIndex = article.get("ItemIndex", None)
                ContentUrl = article.get("ContentUrl", None)
                SourceUrl = article.get("SourceUrl", None)
                CoverImgUrl = article.get("CoverImgUrl", None)
                CoverImgUrl_1_1 = article.get("CoverImgUrl_1_1", None)
                CoverImgUrl_235_1 = article.get("CoverImgUrl_235_1", None)
                ItemShowType = article.get("ItemShowType", None)
                IsOriginal = article.get("IsOriginal", None)
                ShowDesc = article.get("ShowDesc", None)
                show_stat = Functions().show_desc_to_sta(ShowDesc)
                ori_content = article.get("ori_content", None)
                show_view_count = show_stat.get("show_view_count", 0)
                show_like_count = show_stat.get("show_like_count", 0)
                show_zs_count = show_stat.get("show_zs_count", 0)
                show_pay_count = show_stat.get("show_pay_count", 0)
                wx_sn = ContentUrl.split("&sn=")[1].split("&")[0] if ContentUrl else None
                info_tuple = (
                    gh_id,
                    account_name,
                    appMsgId,
                    title,
                    Type,
                    createTime,
                    updateTime,
                    Digest,
                    ItemIndex,
                    ContentUrl,
                    SourceUrl,
                    CoverImgUrl,
                    CoverImgUrl_1_1,
                    CoverImgUrl_235_1,
                    ItemShowType,
                    IsOriginal,
                    ShowDesc,
                    ori_content,
                    show_view_count,
                    show_like_count,
                    show_zs_count,
                    show_pay_count,
                    wx_sn,
                    json.dumps(baseInfo, ensure_ascii=False),
                    Functions().str_to_md5(title)
                )
                try:
                    insert_sql = f"""
                        INSERT INTO official_articles_v2
                        (ghId, accountName, appMsgId, title, Type, createTime, updateTime, Digest, ItemIndex, ContentUrl, SourceUrl, CoverImgUrl, CoverImgUrl_1_1, CoverImgUrl_255_1, ItemShowType, IsOriginal, ShowDesc, ori_content, show_view_count, show_like_count, show_zs_count, show_pay_count, wx_sn, baseInfo, title_md5)
                        values
                        (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s);
                        """
                    db_client.update(sql=insert_sql, params=info_tuple)
                    log(
                        task="updatePublishedMsgDaily",
                        function="insertEachMsg",
                        message="插入文章数据成功",
                        data={
                            "info": info_tuple
                        }

                    )
                except Exception as e:
                    try:
                        update_sql = f"""
                        UPDATE official_articles_v2
                        SET show_view_count = %s, show_like_count=%s
                        WHERE wx_sn = %s;
                        """
                        db_client.update(sql=update_sql,
                                         params=(show_view_count, show_like_count, wx_sn))
                        log(
                            task="updatePublishedMsgDaily",
                            function="insertEachMsg",
                            message="更新文章数据成功",
                            data={
                                "wxSn": wx_sn,
                                "likeCount": show_like_count,
                                "viewCount": show_view_count
                            }

                        )
                    except Exception as e:
                        log(
                            task="updatePublishedMsgDaily",
                            function="insertEachMsg",
                            message="更新文章失败, 报错原因是： {}".format(e),
                            status="fail"
                        )
                        continue


def updateEachAccount(db_client, gh_id, account_name, latest_update_time, cursor=None):
    """
    更新每一个账号信息
    :param account_name:
    :param cursor:
    :param latest_update_time: 最新更新时间
    :param db_client: 数据库连接信息
    :param gh_id: 公众号 gh_id
    :return: None
    """
    response = WeixinSpider().update_msg_list(ghId=gh_id, index=cursor)
    msg_list = response.get("data", {}).get("data", {})
    if msg_list:
        # do
        last_article_in_this_msg = msg_list[-1]
        last_time_stamp_in_this_msg = last_article_in_this_msg['AppMsg']['BaseInfo']['UpdateTime']
        last_url = last_article_in_this_msg['AppMsg']['DetailInfo'][0]['ContentUrl']
        resdata = WeixinSpider().get_account_by_url(last_url)
        check_id = resdata['data'].get('data', {}).get('wx_gh')
        if check_id == gh_id:
            insertEachMsg(
                db_client=db_client,
                gh_id=gh_id,
                account_name=account_name,
                msg_list=msg_list
            )
            if last_time_stamp_in_this_msg > latest_update_time:
                next_cursor = response['data']['next_cursor']
                return updateEachAccount(
                    db_client=db_client,
                    gh_id=gh_id,
                    account_name=account_name,
                    latest_update_time=latest_update_time,
                    cursor=next_cursor
                )
            log(
                task="updatePublishedMsgDaily",
                function="updateEachAccount",
                message="账号文章更新成功",
                data=response
            )
    else:
        log(
            task="updatePublishedMsgDaily",
            function="updateEachAccount",
            message="账号文章更新失败",
            status="fail",
            data=response
        )
        return


def checkAccountInfo(db_client, gh_id):
    """
    通过 gh_id查询视频信息
    :param db_client:
    :param gh_id:
    :return:
    """
    sql = f"""
        select accountName, updateTime 
        from official_articles_v2 
        where ghId = '{gh_id}' 
        order by updateTime DESC;
        """
    result = db_client.select(sql)
    if result:
        account_name, update_time = result[0]
        return {
            "account_name": account_name,
            "update_time": update_time,
            "account_type": "history"
        }
    else:
        return {
            "account_name": "",
            "update_time": int(time.time()) - 30 * 24 * 60 * 60,
            "account_type": "new"
        }


def updateSingleJob(db_client, gh_id):
    """

    :param db_client:
    :param gh_id:
    :return:
    """
    account_info = checkAccountInfo(db_client, gh_id)
    account_name = account_info['account_name']
    update_time = account_info['update_time']
    updateEachAccount(
        db_client=db_client,
        gh_id=gh_id,
        account_name=account_name,
        latest_update_time=update_time
    )


def checkSingleAccount(db_client, account_item):
    """
    校验每个账号是否更新
    :param db_client:
    :param account_item:
    :return: True / False
    """
    gh_id = account_item['ghId']
    account_type = account_item['type']
    today_str = datetime.today().strftime("%Y-%m-%d")
    today_date_time = datetime.strptime(today_str, "%Y-%m-%d")
    today_timestamp = today_date_time.timestamp()
    sql = f"""
                select updateTime
                from official_articles_v2
                where ghId = '{gh_id}'
                order by updateTime
                desc;
            """
    try:
        latest_update_time = db_client.select(sql)[0][0]
        # 判断该账号当天发布的文章是否被收集
        if account_type == "订阅号":
            if int(latest_update_time) > int(today_timestamp):
                return True
            else:
                return False
        else:
            if int(latest_update_time) > int(today_timestamp) - 7 * 24 * 3600:
                return True
            else:
                return False
    except Exception as e:
        print("updateTime Error -- {}".format(e))
        return False


def updateJob():
    """
    更新任务
    :return:
    """
    db_client = PQMySQL()
    sub_accounts, server_accounts = getAccounts()
    s_count = 0
    f_count = 0
    for sub_item in tqdm(sub_accounts):
        try:
            updateSingleJob(db_client, sub_item['ghId'])
            s_count += 1
            time.sleep(5)
        except Exception as e:
            f_count += 1
            log(
                task="updatePublishedMsgDaily",
                function="updateJob",
                message="单个账号文章更新失败, 报错信息是: {}".format(e),
                status="fail",
            )
    log(
        task="updatePublishedMsgDaily",
        function="updateJob",
        message="订阅号更新完成",
        data={
            "success": s_count,
            "fail": f_count
        }
    )

    if f_count / (s_count + f_count) > 0.3:
        bot(
            title="订阅号超过 30% 的账号更新失败",
            detail={
                "success": s_count,
                "fail": f_count,
                "failRate": f_count / (s_count + f_count)
            }
        )

    for sub_item in tqdm(server_accounts):
        try:
            updateSingleJob(db_client, sub_item['ghId'])
            time.sleep(5)
        except Exception as e:
            print(e)


def checkJob():
    """
    校验任务
    :return:
    """
    db_client = PQMySQL()
    sub_accounts, server_accounts = getAccounts()
    fail_list = []
    # account_list = sub_accounts + server_accounts
    account_list = sub_accounts
    # check and rework if fail
    for sub_item in tqdm(account_list):
        res = checkSingleAccount(db_client, sub_item)
        if not res:
            updateSingleJob(db_client, sub_item['ghId'])
    # check whether success and bot if fails
    for sub_item in tqdm(account_list):
        res = checkSingleAccount(db_client, sub_item)
        if not res:
            fail_list.append(sub_item)
    if fail_list:
        try:
            bot(
                title="日常报警, 存在账号更新失败",
                detail=fail_list
            )
        except Exception as e:
            print("Timeout Error: {}".format(e))


def job_with_thread(job_func):
    """
    每个任务放到单个线程中
    :param job_func:
    :return:
    """
    job_thread = threading.Thread(target=job_func)
    job_thread.start()


if __name__ == '__main__':
    schedule.every().day.at("20:50").do(job_with_thread, updateJob)

    schedule.every().day.at("21:45").do(job_with_thread, checkJob)

    while True:
        schedule.run_pending()
        time.sleep(1)
        # log(
        #     task="updatePublishedMsgDaily",
        #     function="main",
        #     message="更新公众号文章信息任务正常执行"
        # )