we-analysis/manager.py

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

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


class FeishuManager(object):
    """
    写入飞书管理器
    """
    def __init__(self):
        self.F = Feishu()
        self.M = Mysql()
        self.map = {
            "产品功能实验": "ca8499",
            "产品实验": "gwzBOM",
            "广告实验": "Pt6JUj",
            "产品/广告实验": "avHzKO",
            "算法实验": "MFwgBr",
            "广告类型策略实验": "XCCAOP"
        }

    def select(self, minigram, date, task_name):
        """
        从mysql中把数据读出来
        :param task_name: 实验名称
        :param minigram: 小程序名称
        :param date: 日期
        :return: data_list
        """
        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)
        init_list = []
        for line in we_tuple:
            result = [i for i in ([line[3]] + list(line[5:]))]
            init_list.append(result)
        # We分析矩阵
        we_array = np.transpose(init_list)
        we_array_v2 = we_array[1:, :]
        # 去掉全部是0的行
        non_zero_rows = np.any(we_array_v2 != 0, axis=1)
        we_array_v3 = we_array_v2[non_zero_rows]
        we_list = we_array_v3.tolist()
        return we_list

    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:AW24"
        )


if __name__ == '__main__':
    Fm = FeishuManager()
    account_name = "票圈内容精选"
    date_str = "2024-01-17"
    temp = Fm.select(account_name, date_str)
    Fm.insert(temp, date_str, account_name)