okr_kd/app.py

from functions.odps_function import OdpsFunction
from functions.feishu import Feishu
from config import column_map, alg_map, yesterday_columns
from functions.get_yesterday_data import find_yesterday_data


def read_odps_data():
    """
    从 odps 读取数据
    :return:
    """
    o = OdpsFunction()
    sql = """select * from loghubods.okr_kd where dt = '20240225';"""
    data = o.select(sql)
    return data


def process_data(data_obj):
    """
    把对象数据转化为 list
    :param data_obj:
    :return:
    """
    keys = ["c{}".format(i) for i in range(1, 350)]
    temp = {}
    for key in keys:
        try:
            temp[key] = int(data_obj[key])
        except:
            temp[key] = data_obj[key]
    # 获取昨天的数据并且更新到 temp 中
    for column in yesterday_columns:
        if temp[column_map[column]] != "-" or temp[column_map[column]] != "--":
            yes_d = find_yesterday_data(column)
            if type(yes_d) != str:
                temp[column_map[column]] = yes_d
    result = []
    for key in temp:
        if alg_map.get(key):
            obj = {
                "type": "formula",
                "text": alg_map[key],
                "number_setting": {"format": "percentage", "decimal_count": 2},
            }
            result.append(obj)
        else:
            result.append(temp[key])
    return result


def insert_into_feishu(data_list):
    """
    插入飞书表
    :param data_list:
    :return:
    """
    F = Feishu("C1Qrsa4HWh6bzEtv7aocrFlAnad")
    F.insert_value(sheet_id="Zi7oYW", values=[data_list[:100]], ranges="A5:CV5")
    F.insert_value(sheet_id="Zi7oYW", values=[data_list[100:200]], ranges="CW5:GR5")
    F.insert_value(sheet_id="Zi7oYW", values=[data_list[200:300]], ranges="GS5:KN5")
    F.insert_value(sheet_id="Zi7oYW", values=[data_list[300:]], ranges="KO5:MK5")


if __name__ == "__main__":
    data = read_odps_data()
    result_list = process_data(data)
    insert_into_feishu(result_list)