we-analysis/feishu/feishu.py

"""
feishu python方法
"""
import json
import random

import requests
import datetime


def get_app_token():
    """
    获取飞书api token
    :return:
    """
    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal/"
    post_data = {
        "app_id": "cli_a51114cf8bf8d00c",  # 这里账号密码是发布应用的后台账号及密码
        "app_secret": "cNoTAqMpsAm7mPBcpCAXFfvOzCNL27fe",
    }
    response = requests.request("POST", url=url, data=post_data)
    tenant_access_token = response.json()["tenant_access_token"]
    return tenant_access_token


class Feishu(object):
    """
    feishu Python Object
    """

    def __init__(self):
        self.document_url = "https://w42nne6hzg.feishu.cn/sheets/C1Qrsa4HWh6bzEtv7aocrFlAnad"
        self.robot_url = "https://open.feishu.cn/open-apis/bot/v2/hook/34e2fdbc-0649-44d3-b5ce-b28b38cca1db"
        self.headers = {"Content-Type": "application/json"}
        self.document_token = "C1Qrsa4HWh6bzEtv7aocrFlAnad"

    def insert_columns_rows(self, sheet_id, major_dimension, start_index, end_index):
        """
        写入飞书表格
        :param sheet_id: 表的id
        :param major_dimension:行或者列， ROWS， COLUMNS
        :param start_index:开始位置
        :param end_index:结束位置
        """
        insert_columns_url = "https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/" \
                             + self.document_token + "/insert_dimension_range"
        headers = {
            "Authorization": "Bearer " + get_app_token(),
            "Content-Type": "application/json; charset=utf-8",
        }
        body = {
            "dimension": {
                "sheetId": sheet_id,
                "majorDimension": major_dimension,  # 默认 ROWS ，可选 ROWS、COLUMNS
                "startIndex": start_index,  # 开始的位置
                "endIndex": end_index  # 结束的位置
            },
            "inheritStyle": "AFTER"  # BEFORE 或 AFTER，不填为不继承 style
        }
        response = requests.post(url=insert_columns_url, headers=headers, json=body, verify=False)
        print(json.dumps(response.json(), ensure_ascii=False, indent=4))

    def insert_value(self, sheet_id, ranges, values):
        """
        在表的某一个sheet的ranges中插入数据，若该地方存在数据，会自动把已有的数据往下移动，再写如数据
        :param sheet_id: 飞书表的唯一ID
        :param ranges: 单元格位置的range, 从左上角到右下角， 两边都是闭区间
        :param values: 二维数组， 用于填充ranges的空格数组
        """
        insert_value_url = "https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/{}/values_prepend".format(
            self.document_token)
        headers = {
            "Authorization": "Bearer " + get_app_token(),
            'contentType': 'application/json'
        }
        body = {
            "valueRange": {
                "range": "{}!{}".format(sheet_id, ranges),
                "values": values
            }
        }
        response = requests.request("POST", url=insert_value_url, headers=headers, json=body)
        print(response.json())

    def bot(self, platform_name, flag=1):
        """
        飞书机器人，在群里报警用
        :param platform_name: 小程序的名称
        :param flag: 通知 or 报警， 默认为1， 通知
        :return:
        """
        if flag == 1:
            payload = {
                "msg_type": "interactive",
                "card": {
                    "elements": [
                        {
                            "tag": "div",
                            "text": {
                                "content": "**{}**\n数据自动导出完成;\n完成时间是：{}。".format(
                                    platform_name, datetime.datetime.now().__str__()
                                ),
                                "tag": "lark_md",
                            },
                        },
                    ],
                    "header": {"title": {"content": "We分析: 通知 ✅", "tag": "plain_text"}},
                },
            }
        else:
            payload = {
                "msg_type": "interactive",
                "card": {
                    "elements": [
                        {
                            "tag": "div",
                            "text": {
                                "content": "**{}**\n自动导出失败 ！！！".format(platform_name),
                                "tag": "lark_md",
                            },
                        },
                    ],
                    "header": {
                        "title": {"content": "We分析: 报警  ❌", "tag": "plain_text"}
                    },
                },
            }
        w = requests.request(
            "POST", url=self.robot_url, headers=self.headers, data=json.dumps(payload)
        )
        print(w.json())

    def remind(self):
        """
        飞书提醒机器人
        :return: None
        """
        payload = {
            "msg_type": "interactive",
            "card": {
                "elements": [
                    {
                        "tag": "div",
                        "text": {
                            "content": "扫码通知：记得要扫码导出we分析, ",
                            "tag": "lark_md",
                        },
                    },
                ],
                "header": {"title": {"content": "We分析: 提醒 ⏰", "tag": "plain_text"}},
            },
        }
        w = requests.request(
            "POST", url=self.robot_url, headers=self.headers, data=json.dumps(payload)
        )
        print(w.json())


if __name__ == "__main__":
    F = Feishu()
    for i in range(100):
        value = [[i + 1] + [random.randint(0, 100) for item in range(7)]]
        F.insert_value(sheet_id="gwzBOM", values=value, ranges="A2:G2")