""" @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": "存在文章更新失败\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)