| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182 | import datetimeimport jsonimport osimport sysimport tracebackfrom itertools import productimport pandas as pdimport requestsimport yamlfrom loguru import loggercurr_path = os.path.abspath(os.path.dirname(__file__))project_root_path = curr_path[:curr_path.find("ai_target") + len("ai_target")]sys.path.append(project_root_path)from config.url_and_db import estimate_people_number_url, headers, jeecg_db, jeecg_product_db, application_product_dbfrom utils.BaseClass import NpEncoderfrom utils.commonFunc import age_code_transform, gender_code_transformlogger.remove()  # 删去 import logger之后自动产生的handler,不删除的话会出现重复输出的现象logger.add("/data/pythonProject/ai_target/logs/bayes_combine.{time:YYYY-MM-DD}.log",           rotation="00:00",           format="{time:YYYY-MM-DD HH:mm:ss,SSS} [{process}] [{thread}] {level} {file} {line} - {message}",           level="INFO")class GetSingleDimFeatures(object):    """    计算素材单维度滑窗组合转化占比情况    """    def __init__(self, signature, dim, dim_config):        self.signature = signature        self.dim = dim        self.dim_config = dim_config        self.table = self.dim_config['table']        self.filed_name = self.dim_config['fieldName']        self.dim_value_lst = self.dim_config['Lst']        self.ratio_threshold = self.dim_config['ratio_threshold']        self.ratio_diff_threshold = self.dim_config['ratio_diff_threshold']        self.window_size = self.dim_config['windowSize']        self.window_combine = []        self.features = {}    def get_window_combine(self):        """        根据size得到指定维度的滑窗组合。        如 age: ['18-23岁', '24-30岁','31-40岁',....] 和 size = 2        得到 ['18-23岁', '24-30岁'],['24-30岁','31-40岁'],......        """        start = 0        while (start < len(self.dim_value_lst)) and (start + self.window_size <= len(self.dim_value_lst)):            self.window_combine.append(self.dim_value_lst[start: start + self.window_size])            start += 1        # 添加一个全部的组合,等同于'不限'        self.window_combine.append(['不限'])    def get_features(self):        # 1、获取素材在指定维度下的人群数据        get_audience_sql = f"select signature, {self.filed_name}, sum(activation) activation  " \                           f"from {self.table} where signature = '{self.signature}' group by signature, {self.filed_name}"        audience_df = pd.read_sql(get_audience_sql, jeecg_product_db)        for combine in self.window_combine:            if combine != ['不限']:                self.features['|'.join(combine)] = round(audience_df[audience_df[self.filed_name].isin(combine)].activation.sum() / \                                                         audience_df.activation.sum(), 4)            else:                if self.dim == 'gender':                    # 如果男/女 激活占比差异小于0.1, 说明素材的受众没有明显的性别倾向,可以对性别通投                    male_ratio = audience_df[audience_df['gender'] == '男'].activation.sum() / audience_df.activation.sum()                    female_ratio = audience_df[audience_df['gender'] == '女'].activation.sum() / audience_df.activation.sum()                    if abs(male_ratio - female_ratio) <= self.ratio_diff_threshold:                        self.features['不限'] = 1                if self.dim == 'age':                    self.features['不限'] = 1        # 过滤得到占比达到 ratio_threshold 的组合值        self.features = {f"{self.dim}_{combine}": ratio for combine, ratio in self.features.items() if ratio > self.ratio_threshold}class GetCompositeDimFeatures(object):    """    """    def __init__(self, signature, feature_lst, project_id):        self.signature = signature        self.feature_lst = feature_lst        self.project_id = project_id    def get_composite_features(self):        feature_combine_lst = [list(item.keys()) for item in self.feature_lst]        fea_combine_df = pd.DataFrame()        for ele in product(*feature_combine_lst):            fea_combine = dict(zip([e.split('_')[0] for e in ele], [e.split('_')[-1] for e in ele]))            fea_combine['unlimited_cnt'] = list(fea_combine.values()).count('不限')            fea_combine['features'] = str([(e.split('_')[0], e.split('_')[1], item[e]) for e in ele                                           for item in self.feature_lst if e in item.keys()])            try:                get_open_account_sql = f"select account_id from ctop_user_allocation where project_id = {self.project_id} and " \                                       f"account_status = 0 limit 1"                account_df = pd.read_sql(get_open_account_sql, jeecg_product_db)                account_id = account_df.account_id.values[0]                # account_id = 9774238                request_data = {'ages_range': age_code_transform(fea_combine.get('age')),                                'gender': gender_code_transform(fea_combine.get('gender')),                                'advertiser_id': account_id}                request = requests.post(url=estimate_people_number_url,                                        headers=headers,                                        data=json.dumps(request_data, cls=NpEncoder))                response_data = json.loads(request.text)                if response_data['code'] == 0:                    fea_combine['crowd_coverage_cnt'] = response_data['data'].get('audience_prediction_num')            except:                logger.error(f"人群预估覆盖接口调用报错,请求数据: {request_data}, 返回数据: {request.text}, 异常信息: {traceback.format_exc()}")            fea_combine_df = fea_combine_df.append(pd.DataFrame([fea_combine]), ignore_index=True)        # 计算结果写入数据库中        fea_combine_df['signature'] = self.signature        fea_combine_df['stat_date'] = str(datetime.datetime.now().date())        fea_combine_df['project_id'] = self.project_id        fea_combine_df.to_sql(name="ctop_ai_kuaishou_signature_recommended_target_combine_v2",                              con=jeecg_db,                              if_exists='append',                              index=False)if __name__ == '__main__':    # 1、读取配置文件    with open('/data/pythonProject/ai_target/config/config.yaml', mode='r', encoding='utf-8') as f:        config = yaml.load(f.read(), Loader=yaml.FullLoader)    # 2、 参与定向组合的维度    target_dim = [key for key in config['bayesDim'].keys() if config['bayesDim'][key]['isOn']]    # 4、计算贝叶斯组合入库    for project_id in config['projectId']:        logger.info(f"开始获取项目{project_id}的当前活跃素材")        # 4.1 获取指定项目下当前活跃的素材信息        # 如近60天内累计激活个数达到100个(开发环境或者生产环境,这部分都读取生产数据库)        start_date = datetime.datetime.now().date() + datetime.timedelta(days=-config['activeMaterialFilterRule']['days'])        sql = f"select signature from kuaishou_material_video_report_daily_dw where project_id = {project_id} " \              f"and stat_date >= {start_date.year * 10000 + start_date.month * 100 + start_date.day} group by signature " \              f"having sum(activation) >= {config['activeMaterialFilterRule']['activation']} "  \              f"and sum(event_pay_first_day) / sum(activation) >= {config['activeMaterialFilterRule']['event_pay_first_day_pct']}"        df = pd.read_sql(sql, application_product_db)        active_signatures = list(df[~df.signature.isnull()].signature.values)        active_signatures = active_signatures * 2 if len(active_signatures) == 1 else active_signatures        # 4.2 在素材人群报表筛选里面累计100个激活        logger.info(f"开始获取项目{project_id}的活跃素材的人群分析报表")        sql = f"select signature from ctop_kuaishou_audience_report_daily_age_material where signature in {tuple(active_signatures)} " \              f"group by signature having sum(activation) >= {config['getBaysCombineMaterialFilterRule']['activation']}"        df = pd.read_sql(sql, jeecg_product_db)        signature_lst = df['signature'].values        # 这行代码用于测试        # signature_lst = ['f7adee9229e58176d132cf0bd7590a95']        # 4.2 计算指定素材定向        for sig in signature_lst:            try:                feature_lst = []                for dimension in target_dim:                    get_single_fea_ins = GetSingleDimFeatures(sig, dimension, config['bayesDim'][dimension])                    # 计算滑窗组合                    get_single_fea_ins.get_window_combine()                    get_single_fea_ins.get_features()                    feature_lst.append(get_single_fea_ins.features)                # 依据特征值,计算多维度的组合预估值                get_composite_fea_ins = GetCompositeDimFeatures(sig, feature_lst, project_id)                get_composite_fea_ins.get_composite_features()                logger.info('project_id=%s, signature= %s 完成贝叶斯组合计算!' % (project_id, sig))            except:                logger.error('project_id=%s, signature=%s 贝叶斯组合计算出错,异常信息为%s!' % (project_id, sig, traceback.format_exc()))    logger.info('project_id is %s 完成贝叶斯组合计算!' % project_id)
 |