we-analysis/manager.py

"""
往飞书内写数据
"""
import numpy as np

from feishu.feishu import Feishu
from applications.mysql import Mysql
from applications.config import platform_map


class FeishuManager(object):
    """
    写入飞书管理器
    """

    def __init__(self):
        self.F = Feishu()
        self.M = Mysql()
        self.map = {
            "产品功能实验": "ca8499",
            "产品实验": "gwzBOM",
            "广告实验": "Pt6JUj",
            "产品/广告实验": "avHzKO",
            " 产品/广告实验": "avHzKO",
            "算法实验": "MFwgBr",
            "广告类型策略实验": "XCCAOP",
            "广告类型策略实验层": "XCCAOP"
        }
        self.subject_map = {
            "小程序分享人数": "微信分享人数",
            "小程序分享次数": "微信分享次数",
            "小程序访问人数": "微信访问人数",
            "广告 eCPM": "腾讯CPM",
            "广告收入": "腾讯收入",
            "广告曝光人数": "腾讯曝光人数",
            "广告曝光次数": "腾讯曝光次数"
        }
        self.order_map = {
            "腾讯曝光次数": 0,
            "腾讯曝光人数": 1,
            "腾讯CPM": 2,
            "腾讯收入": 11,
            "微信访问人数": 14,
            "微信分享人数": 15,
            "微信分享次数": 16
        }

    def select(self, minigram, date, task_name):
        """
        从mysql中把数据读出来
        :param task_name: 实验名称
        :param minigram: 小程序名称
        :param date: 日期
        :return: data_dict
        """
        select_sql = f"""SELECT * FROM we_analysis_results where mini_program = '{minigram}' and date = '{date}' and task_name = '{task_name}';"""
        we_tuple = self.M.select(select_sql)
        data_dict = {}
        for line in we_tuple:
            result = list(line[5:])
            key = line[3]
            data_dict[key] = result
        return data_dict

    def insert(self, we_list, date, mini_program, task_name):
        """
        处理矩阵，并且插入飞书表
        :param task_name: 实验名称
        :param date: 日期
        :param mini_program: 小程序
        :param we_list: we分析矩阵
        :return: None
        """
        rows = len(we_list)
        datas = [
            [date, 0, mini_program, "ab{}".format(i), "00:00", "", "实验名称"] + we_list[i]
            for i in range(rows)
        ]
        self.F.insert_value(
            sheet_id=self.map[task_name],
            values=datas,
            ranges="A4:CI{}".format(4 + len(datas))
        )

    def process_data_dict(self, data_dict):
        """
        把数据处理成格式
        :param data_dict: we-analysis-information
        :return:
        """
        init_array = np.zeros((25, 80))
        for key in data_dict.keys():
            sub_data = np.array(data_dict[key])
            index = self.order_map[self.subject_map[key]]
            init_array[:, index] = sub_data
        non_zero_rows = np.any(init_array != 0, axis=1)
        init_array = init_array[non_zero_rows]
        # 处理其他数据指标
        wechat_uv = np.where(init_array[:, 14] == 0, 0, init_array[:, 1] / init_array[:, 14])
        personal_expose = np.where(init_array[:, 1] == 0, 0, init_array[:, 0] / init_array[:, 1])
        total_arpu = np.where(init_array[:, 14] == 0, 0, (init_array[:, 11] + init_array[:, 12]) / init_array[:, 14])
        init_array[:, 10] = wechat_uv
        init_array[:, 3] = personal_expose
        init_array[:, 13] = total_arpu
        we_list = init_array.tolist()
        return we_list


if __name__ == '__main__':
    Fm = FeishuManager()
    date_str = "2024-01-17"
    for account_name in platform_map.keys():
        for task_n in platform_map[account_name].keys():
            temp = Fm.select(account_name, date_str, task_n)
            temp_list = Fm.process_data_dict(temp)
            Fm.insert(temp_list, date_str, account_name, task_n)
            print(account_name, task_n)