rov-offline/user_group_update.py

import datetime
import logging
import multiprocessing
import time
import traceback
import gevent
import asyncio
from threading import Timer
from concurrent.futures import ThreadPoolExecutor

from utils import RedisHelper, data_check, get_feature_data, send_msg_to_feishu
from config import set_config
from log import Log
config_, _ = set_config()
log_ = Log()
redis_helper = RedisHelper()

# features = [
#     'apptype',
#     'return1mids',
#     'return2_3mids',
#     'return4_8mids',
#     'return9_24mids',
#     'return25_nmids',
#     'return0share1mids',
#     'return0share2_nmids'
# ]

mid_group_mapping_global = {}
mids_global = []


def to_redis(group, mid_list, class_key_list):
    log_.info(f"group = {group} update redis start ...")
    start_time = time.time()
    log_.info(f"mid count = {len(mid_list)}")
    for class_key in class_key_list:
        for i in range(len(mid_list) // 100 + 1):
            # log_.info(f"i = {i}")
            mid_temp_list = mid_list[i * 100:(i + 1) * 100]
            # print(mid_temp_list)
            task_list = [
                gevent.spawn(redis_helper.set_data_to_redis,
                             f"{config_.KEY_NAME_PREFIX_MID_GROUP}{class_key}:{mid}", group, 28 * 3600)
                for mid in mid_temp_list
            ]
            gevent.joinall(task_list)
    log_.info(f"group = {group}, mid count = {len(mid_list)}, update redis finished! "
              f"execute time = {(time.time() - start_time) / 60}min")


def to_redis2(process_mid_list, mid_group_mapping, ad_mid_group_key_params):
    log_.info(f"mid count = {len(process_mid_list)}")
    start_time = time.time()
    for i in range(len(process_mid_list) // 100 + 1):
        mid_temp_list = process_mid_list[i * 100:(i + 1) * 100]
        task_list = []
        for mid in mid_temp_list:
            group_list = mid_group_mapping.get(mid)
            mid_value = {}
            for group in group_list:
                for class_key in ad_mid_group_key_params.get(group, []):
                    mid_value[class_key] = group
            # print(f"mid={mid}, mid_value={mid_value}")
            if len(mid_value) > 0:
                task_list.append(
                    gevent.spawn(redis_helper.set_data_to_redis,
                                 f"{config_.KEY_NAME_PREFIX_MID_GROUP}{mid}", str(mid_value), 28 * 3600)
                )
        gevent.joinall(task_list)
    log_.info(f"mid count = {len(process_mid_list)}, update redis finished! "
              f"execute time = {(time.time() - start_time) / 60}min")


def mapping_process(group, mid_list):
    global mids_global, mid_group_mapping_global
    for mid in mid_list:
        if mid is None:
            continue
        if mid in mids_global:
            mid_group_mapping_global[mid].append(group)
        else:
            mid_group_mapping_global[mid] = [group]
            mids_global.append(mid)


def mapping_process2(mid_list, group_list, group_mid_list):
    global mid_group_mapping_global
    for mid in mid_list:
        if mid is None:
            continue
        mid_group = [group for group in group_list if mid in group_mid_list.get(group, [])]
        mid_group_mapping_global[mid] = mid_group


async def get_mid_group_mapping2(feature_df, group_list):
    """获取mid对应的分组列表"""
    start_time = time.time()
    group_mid_list = {}
    mid_list_all = []
    for group in group_list:
        start_time = time.time()
        mid_list = feature_df[group].tolist()
        mid_list = list(set(mid_list))
        group_mid_list[group] = mid_list
        mid_list_all.extend(mid_list)
        log_.info(f"group = {group}, mid_list_count = {len(mid_list)}")

    mid_list_all = list(set(mid_list_all))
    log_.info(f"mid_list_all count = {len(mid_list_all)}")
    global mids_global, mid_group_mapping_global
    mids_global = mid_list_all
    step = 10000
    loop = asyncio.get_running_loop()
    executor = ThreadPoolExecutor(max_workers=20)
    tasks = []
    for i in range(len(mid_list_all) // step + 1):
        log_.info(f"i = {i}")
        process_mid_list = mid_list_all[i * step:(i + 1) * step]
        tasks.append(loop.run_in_executor(executor, mapping_process2, process_mid_list, group_list, group_mid_list))
    await asyncio.wait(tasks)
    # mid_group_mapping = {}
    # for mid in mid_list_all:
    #     if mid is None:
    #         continue
    #     mid_group = [group for group in group_list if mid in group_mid_list.get(group, [])]
    #     mid_group_mapping[mid] = mid_group
    # return mid_group_mapping, mid_list_all
    log_.info(f"group mid mapping finished! "
              f"mid_count = {len(mids_global)}, mid_group_mapping_count = {len(mid_group_mapping_global)}, "
              f"execute time = {(time.time() - start_time) / 60}min")


async def get_mid_group_mapping(feature_df, group_list):
    """获取mid对应的分组列表"""
    for group in group_list:
        start_time = time.time()
        mid_list = feature_df[group].tolist()
        mid_list = list(set(mid_list))
        log_.info(f"group = {group}, mid_list_count = {len(mid_list)}")
        # pool = multiprocessing.Pool(processes=10)
        # step = len(mid_list) // (10 - 1)
        # for i in range(10 + 1):
        #     process_mid_list = mid_list[i * step:(i + 1) * step]
        #     pool.apply_async(func=mapping_process, args=(group, process_mid_list))
        # pool.close()
        # pool.join()

        # step = len(mid_list) // (100 - 1)
        step = 100
        loop = asyncio.get_running_loop()
        executor = ThreadPoolExecutor(max_workers=20)
        tasks = []
        for i in range(len(mid_list)//step + 1):
            log_.info(f"i = {i}")
            process_mid_list = mid_list[i * step:(i + 1) * step]
            tasks.append(loop.run_in_executor(executor, mapping_process, group, process_mid_list))
        await asyncio.wait(tasks)
        global mids_global, mid_group_mapping_global
        log_.info(f"group = {group} mid mapping finished! "
                  f"mid_count = {len(mids_global)}, mid_group_mapping_count = {len(mid_group_mapping_global)}, "
                  f"execute time = {(time.time() - start_time) / 60}min")

        # for mid in mid_list:
        #     if mid is None:
        #         continue
        #     if mid in mids:
        #         mid_group_mapping[mid].append(group)
        #     else:
        #         mid_group_mapping[mid] = [group]
        #         mids.append(mid)
    # mid_group_mapping, mids = mid_group_mapping_global, mids_global
    # return mid_group_mapping, mids


def update_user_group_to_redis(project, table, dt, app_type_list, features, ad_mid_group_key_params):
    """更新mid对应分组到redis中"""
    # 获取用户分组数据
    feature_df = get_feature_data(project=project, table=table, features=features, dt=dt)
    feature_df['apptype'] = feature_df['apptype'].astype(int)
    feature_df = feature_df[feature_df['apptype'].isin(app_type_list)]
    # print(len(feature_df))

    group_list = [group for group in ad_mid_group_key_params]
    # mid_group_mapping, mids = get_mid_group_mapping(feature_df=feature_df, group_list=group_list)
    asyncio.run(get_mid_group_mapping2(feature_df=feature_df, group_list=group_list))
    global mid_group_mapping_global, mids_global
    mid_group_mapping, mids = mid_group_mapping_global, mids_global

    pool = multiprocessing.Pool(processes=len(ad_mid_group_key_params))
    step = len(mids) // (len(ad_mid_group_key_params) - 1)
    for i in range(len(ad_mid_group_key_params) + 1):
        process_mid_list = mids[i*step:(i+1)*step]
        pool.apply_async(func=to_redis2, args=(process_mid_list, mid_group_mapping, ad_mid_group_key_params))

    # for group, class_key_list in ad_mid_group_key_params.items():
    #     mid_list = feature_df[group].tolist()
    #     mid_list = list(set(mid_list))
    #     mid_list = [mid for mid in mid_list if mid is not None]
    #     # class_key_list = ad_mid_group_key_params.get(group)
    #     pool.apply_async(func=to_redis, args=(group, mid_list, class_key_list))
    pool.close()
    pool.join()


def get_group_keys_mapping(ad_mid_group):
    ad_mid_group_key_params = {}
    features = ['apptype']
    for class_key, group_list in ad_mid_group.items():
        for group in group_list:
            if group not in features:
                features.append(group)
                ad_mid_group_key_params[group] = [class_key]
            else:
                ad_mid_group_key_params[group].append(class_key)
    return features, ad_mid_group_key_params


def timer_check():
    try:
        app_type_list = config_.AD_APP_TYPE_LIST
        ad_mid_group = config_.AD_MID_GROUP
        project = config_.ad_model_data['user_group'].get('project')
        table = config_.ad_model_data['user_group'].get('table')
        now_date = datetime.datetime.today()
        dt = datetime.datetime.strftime(now_date, '%Y%m%d')
        log_.info(f"now_date: {dt}")
        now_min = datetime.datetime.now().minute
        # 查看当前更新的数据是否已准备好
        data_count = data_check(project=project, table=table, dt=dt)
        if data_count > 0:
            log_.info(f"user group data count = {data_count}")
            # 获取features & 用户分组对应key
            features, ad_mid_group_key_params = get_group_keys_mapping(ad_mid_group=ad_mid_group)
            log_.info(f"features = {features}, \nad_mid_group_key_params = {ad_mid_group_key_params}")
            # 数据准备好，进行更新
            update_user_group_to_redis(project=project, table=table, dt=dt, app_type_list=app_type_list,
                                       features=features, ad_mid_group_key_params=ad_mid_group_key_params)
            log_.info(f"user group data update end!")
        # elif now_min > 45:
        #     log_.info('user group data is None!')
        #     send_msg_to_feishu(
        #         webhook=config_.FEISHU_ROBOT['server_robot'].get('webhook'),
        #         key_word=config_.FEISHU_ROBOT['server_robot'].get('key_word'),
        #         msg_text=f"rov-offline{config_.ENV_TEXT} - 用户分组数据未准备好！\n"
        #                  f"traceback: {traceback.format_exc()}"
        #     )
        else:
            # 数据没准备好，1分钟后重新检查
            Timer(60, timer_check).start()

    except Exception as e:
        log_.error(f"用户分组数据更新失败, exception: {e}, traceback: {traceback.format_exc()}")
        send_msg_to_feishu(
            webhook=config_.FEISHU_ROBOT['server_robot'].get('webhook'),
            key_word=config_.FEISHU_ROBOT['server_robot'].get('key_word'),
            msg_text=f"rov-offline{config_.ENV_TEXT} - 用户分组数据更新失败\n"
                     f"exception: {e}\n"
                     f"traceback: {traceback.format_exc()}"
        )


if __name__ == '__main__':
    timer_check()