""" @author: luojunhui """ import json import time from tqdm import tqdm from datetime import datetime, timedelta from argparse import ArgumentParser from applications import PQMySQL, DeNetMysql, longArticlesMySQL from applications.const import updateAccountReadAvgTaskConst def get_account_fans_by_dt(db_client) -> dict: """ 获取每个账号发粉丝,通过日期来区分 :return: """ 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 t2.status = 1 AND t1.date_str >= '2024-09-01' ORDER BY t1.date_str; """ result = db_client.select(sql) D = {} for line in result: dt = line[0] fans = line[1] gh_id = line[2] if D.get(gh_id): D[gh_id][dt] = fans else: D[gh_id] = {dt: fans} return D class UpdateAccountInfoVersion3(object): """ 更新账号信息 v3 """ def __init__(self): self.const = updateAccountReadAvgTaskConst() self.pq = PQMySQL() self.de = DeNetMysql() self.lam = longArticlesMySQL() def get_account_position_read_rate(self, dt): """ 从长文数据库获取账号阅读均值 :return: """ dt = int(dt.replace("-", "")) sql = f""" SELECT gh_id, position, read_rate_avg FROM long_articles_read_rate WHERE dt_version = {dt}; """ result = self.lam.select(sql) account_read_rate_dict = {} for item in result: gh_id = item[0] position = item[1] rate = item[2] key = "{}_{}".format(gh_id, position) account_read_rate_dict[key] = rate return account_read_rate_dict def get_publishing_accounts(self): """ 获取每日正在发布的账号 :return: """ sql = f""" SELECT DISTINCT t3.`name`, t3.gh_id, t3.follower_count, t6.account_source_name, t6.mode_type, t6.account_type, t6.`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 = self.de.select(sql) result_list = [ { "account_name": i[0], "gh_id": i[1], "fans": i[2], "account_source_name": i[3], "mode_type": i[4], "account_type": i[5], "status": i[6] } for i in account_list ] return result_list def do_task_list(self, dt): """ do it """ fans_dict = get_account_fans_by_dt(db_client=self.de) account_list = self.get_publishing_accounts() rate_dict = self.get_account_position_read_rate(dt) for account in tqdm(account_list): business_type = self.const.TOULIU if account[ 'gh_id'] in self.const.TOULIU_ACCOUNTS else self.const.ARTICLES_DAILY fans = fans_dict.get(account['gh_id'], {}).get(dt, 0) if fans: for index in range(1, 9): gh_id_position = "{}_{}".format(account['gh_id'], index) if rate_dict.get(gh_id_position): rate = rate_dict[gh_id_position] read_avg = fans * rate print(rate, read_avg) insert_sql = f""" INSERT INTO account_avg_info_v3 (gh_id, position, update_time, account_name, fans, read_avg, like_avg, status, account_type, account_mode, account_source, account_status, business_type, read_rate_avg) values (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s); """ try: self.pq.update( sql=insert_sql, params=( account['gh_id'], index, dt, account['account_name'], fans, read_avg, 0, 1, account['account_type'], account['mode_type'], account['account_source_name'], account['status'], business_type, rate ) ) except Exception as e: updateSQL = f""" UPDATE account_avg_info_v3 set fans = %s, read_avg = %s, read_rate_avg = %s where gh_id = %s and position = %s and update_time = %s """ try: affected_rows = self.pq.update( sql=updateSQL, params=( fans, read_avg, rate, account['gh_id'], index, dt ) ) except Exception as e: print(e) # 修改前一天的状态为 0 update_status_sql = f""" UPDATE account_avg_info_v3 SET status = %s where update_time != %s and gh_id = %s and position = %s; """ rows_affected = self.pq.update( sql=update_status_sql, params=( 0, dt, account['gh_id'], index ) ) print("修改成功") def main(): """ main job :return: """ parser = ArgumentParser() parser.add_argument("--run-date", help="Run only once for date in format of %Y-%m-%d. \ If no specified, run as daily jobs.") args = parser.parse_args() Up = UpdateAccountInfoVersion3() if args.run_date: Up.do_task_list(dt=args.run_date) else: dt_object = datetime.fromtimestamp(int(time.time())) one_day = timedelta(days=1) yesterday = dt_object - one_day yesterday_str = yesterday.strftime('%Y-%m-%d') Up.do_task_list(dt=yesterday_str) if __name__ == '__main__': main()