123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188 |
- import datetime
- import json
- import os
- import sys
- import traceback
- from itertools import product
- import pandas as pd
- import requests
- import yaml
- from loguru import logger
- curr_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_db, application_inner_db
- from utils.BaseClass import NpEncoder
- from utils.commonFunc import age_code_transform, gender_code_transform
- logger.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, project_id):
- self.project_id = project_id
- 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}' and project_id = {self.project_id} " \
- f"group by signature, {self.filed_name}"
- audience_df = pd.read_sql(get_audience_sql, application_inner_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个(开发环境或者生产环境,这部分都读取生产数据库)
- days = config['activeMaterialFilterRule']['days']
- activation_sum = config['activeMaterialFilterRule']['activation']
- pay_first_pct = config['activeMaterialFilterRule']['event_pay_first_day_pct']
- sql = f"select signature from kuaishou_material_video_report_daily_dw where project_id = {project_id} " \
- f"and stat_date >= date_format(date_sub(now(), interval {days} DAY),'%%Y%%m%%d') group by signature " \
- f"having sum(activation) >= {activation_sum} " \
- f"and sum(event_pay_first_day) / sum(activation) >= {pay_first_pct}"
- df = pd.read_sql(sql, application_inner_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 kuaishou_unit_audience_age_report_daily_d where signature in {tuple(active_signatures)} " \
- f"and project_id = {project_id} " \
- f"group by signature having sum(activation) >= {config['getBaysCombineMaterialFilterRule']['activation']}"
- df = pd.read_sql(sql, application_inner_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], project_id)
- # 计算滑窗组合
- 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)
|