LongArticlesJob-fork/updatePublishedMsgDaily.py

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

import time
import json
import threading

import requests
import schedule
from tqdm import tqdm
from datetime import datetime

from applications import PQMySQL
from applications.decoratorApi import retryOnTimeout
from applications import WeixinSpider
from applications import Functions


@retryOnTimeout()
def bot(account_list):
    """
    机器人
    """
    url = "https://open.feishu.cn/open-apis/bot/v2/hook/b44333f2-16c0-4cb1-af01-d135f8704410"
    headers = {"Content-Type": "application/json"}
    payload = {
        "msg_type": "interactive",
        "card": {
            "elements": [
                {
                    "tag": "div",
                    "text": {
                        "content": "存在文章更新失败<at id=all></at>\n",
                        "tag": "lark_md",
                    },
                },
                {
                    "tag": "div",
                    "text": {
                        "content": json.dumps(
                            account_list, ensure_ascii=False, indent=4
                        ),
                        "tag": "lark_md",
                    },
                },
            ],
            "header": {"title": {"content": "【重点关注】", "tag": "plain_text"}},
        },
    }
    requests.request("POST", url=url, headers=headers, data=json.dumps(payload), timeout=10)


def getAccounts():
    """
    获取账号信息
    :return: [{}, {},...], [{}, {}, {}...]
    """
    with open("config/accountInfoV2.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)
        # if int(time.time()) - int(updateTime) <= 20 * 60 * 60:
        #     continue
        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)
                )
                # print(info_tuple)
                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)
                        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);
                        """
                    db_client.update(sql=insert_sql, params=info_tuple)
                    print("插入成功")
                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))
                        print("更新成功")
                    except Exception as e:
                        print("失败-{}".format(e))
                        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_name = resdata['data'].get('data', {}).get('account_name')
        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=check_name,
                    latest_update_time=latest_update_time,
                    cursor=next_cursor
                )
    else:
        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()

    for sub_item in tqdm(sub_accounts):
        # updateSingleJob(db_client, sub_item['ghId'])
        try:
            updateSingleJob(db_client, sub_item['ghId'])
            time.sleep(2)
        except Exception as e:
            print(e)
    print("订阅号更新完成")

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

    print("服务号更新完成")


def checkJob():
    """
    校验任务
    :return:
    """
    db_client = PQMySQL()
    sub_accounts, server_accounts = getAccounts()
    fail_list = []
    account_list = sub_accounts + server_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(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:30").do(job_with_thread, checkJob)

    while True:
        schedule.run_pending()
        time.sleep(1)