crawler_xiaoniangao/main/feishu_lib.py

# -*- coding: utf-8 -*-
# @Author: wangkun
# @Time: 2022/5/11
import datetime
import json
import time

import requests
import urllib3

from main.common import Common

proxies = {"http": None, "https": None}


class Feishu:
    """
    编辑飞书云文档
    """
    feishu_url = "https://w42nne6hzg.feishu.cn/sheets/shtcngRPoDYAi24x52j2nDuHMih?"
    spreadsheetToken = "shtcngRPoDYAi24x52j2nDuHMih"

    # 获取飞书api token
    @classmethod
    def get_token(cls):
        """
        获取飞书api token
        :return:
        """
        url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal/"
        post_data = {"app_id": "cli_a13ad2afa438d00b",  # 这里账号密码是发布应用的后台账号及密码
                     "app_secret": "4tK9LY9VbiQlY5umhE42dclBFo6t4p5O"}

        try:
            urllib3.disable_warnings()
            response = requests.post(url=url, data=post_data, proxies=proxies, verify=False)
            tenant_access_token = response.json()["tenant_access_token"]
            return tenant_access_token
        except Exception as e:
            Common.logger().error("获取飞书 api token 异常:{}", e)

    # 获取表格元数据
    @classmethod
    def get_metainfo(cls):
        """
        获取表格元数据
        :return:
        """
        url = "https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/" + cls.spreadsheetToken + "/metainfo"
        headers = {
            "Authorization": "Bearer " + cls.get_token(),
            "Content-Type": "application/json; charset=utf-8"
        }
        params = {
            "extFields": "protectedRange",  # 额外返回的字段，extFields=protectedRange时返回保护行列信息
            "user_id_type": "open_id"  # 返回的用户id类型，可选open_id,union_id
        }
        try:
            urllib3.disable_warnings()
            r = requests.get(url=url, headers=headers, params=params, proxies=proxies, verify=False)
            response = json.loads(r.content.decode("utf8"))
            return response
        except Exception as e:
            Common.logger().error("获取表格元数据异常:{}", e)

    # 读取工作表中所有数据
    @classmethod
    def get_values_batch(cls, sheetid):
        """
        读取工作表中所有数据
        :param sheetid: 哪张表
        :return: 所有数据
        """

        url = "https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/" + cls.spreadsheetToken + "/values_batch_get"
        headers = {
            "Authorization": "Bearer " + cls.get_token(),
            "Content-Type": "application/json; charset=utf-8"
        }
        params = {
            # 多个查询范围 如 url?ranges=range1,range2 ，其中 range 包含 sheetId 与单元格范围两部分
            "ranges": sheetid,

            # valueRenderOption=ToString 可返回纯文本的值(数值类型除外)；
            # valueRenderOption=FormattedValue 计算并格式化单元格；
            # valueRenderOption=Formula单元格中含有公式时返回公式本身；
            # valueRenderOption=UnformattedValue计算但不对单元格进行格式化
            "valueRenderOption": "ToString",

            # dateTimeRenderOption=FormattedString 计算并将时间日期按照其格式进行格式化，但不会对数字进行格式化，返回格式化后的字符串。
            "dateTimeRenderOption": "",

            # 返回的用户id类型，可选open_id,union_id
            "user_id_type": "open_id"
        }
        try:
            urllib3.disable_warnings()
            r = requests.get(url=url, headers=headers, params=params, proxies=proxies, verify=False)
            response = json.loads(r.content.decode("utf8"))
            values = response["data"]["valueRanges"][0]["values"]
            return values
        except Exception as e:
            Common.logger().error("读取工作表所有数据异常:{}", e)

    # 工作表，插入行
    @classmethod
    def insert_columns(cls, sheetid, majordimension, startindex, endindex):
        """
        工作表插入数据
        sheetid:哪张工作表
        majordimension:行或者列
        startindex:开始位置
        endindex:结束位置
        """
        url = "https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/"\
              + cls.spreadsheetToken + "/insert_dimension_range"
        headers = {
            "Authorization": "Bearer " + cls.get_token(),
            "Content-Type": "application/json; charset=utf-8"
        }
        body = {
            "dimension": {
                "sheetId": sheetid,
                "majorDimension": majordimension,  # 默认 ROWS ，可选 ROWS、COLUMNS
                "startIndex": startindex,  # 开始的位置
                "endIndex": endindex  # 结束的位置
            },
            "inheritStyle": "AFTER"  # BEFORE 或 AFTER，不填为不继承 style
        }
        try:
            urllib3.disable_warnings()
            r = requests.post(url=url, headers=headers, json=body, proxies=proxies, verify=False)
            Common.logger().info("插入行或列:{}", r.json()["msg"])
        except Exception as e:
            Common.logger().error("插入行或列异常:{}", e)

    # 工作表，首行写入数据
    @classmethod
    def update_values(cls, sheetid, a1, b1, c1, d1, e1, f1, g1, h1, i1, j1, k1, l1, m1, n1, o1):
        """
        写入数据
        :param sheetid: 哪张工作表
        :param a1: 单元格
        :param b1: 单元格
        :param c1: 单元格
        :param d1: 单元格
        :param e1: 单元格
        :param f1: 单元格
        :param g1: 单元格
        :param h1: 单元格
        :param i1: 单元格
        :param j1: 单元格
        :param k1: 单元格
        :param l1: 单元格
        :param m1: 单元格
        :param n1: 单元格
        :param o1: 单元格
        :return:
        """

        url = "https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/" + cls.spreadsheetToken + "/values_batch_update"
        headers = {
            "Authorization": "Bearer " + cls.get_token(),
            "Content-Type": "application/json; charset=utf-8"
        }
        body = {
            "valueRanges": [
                {
                    "range": sheetid + "!A2:O2",
                    "values": [
                        [a1, b1, c1, d1, e1, f1, g1, h1, i1, j1, k1, l1, m1, n1, o1]
                    ]
                },
            ],
        }
        try:
            urllib3.disable_warnings()
            r = requests.post(url=url, headers=headers, json=body, proxies=proxies, verify=False)
            Common.logger().info("空行写入视频数据:{}", r.json()["msg"])
        except Exception as e:
            Common.logger().error("空行写入视频数据异常:{}", e)

    # 写入数据
    @classmethod
    def update_hour_list_values(cls, sheetid, ranges, values):
        """
        小时榜写入数据
        sheetid:工作表 ID
        ranges:单元格范围
        values:写入的具体数据，list
        """
        # 表格 ID
        url = "https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/" + cls.spreadsheetToken + "/values_batch_update"
        headers = {
            "Authorization": "Bearer " + cls.get_token(),
            "Content-Type": "application/json; charset=utf-8"
        }
        body = {
            "valueRanges": [
                {
                    "range": sheetid + "!" + ranges,
                    "values": values
                },
            ],
        }

        try:
            urllib3.disable_warnings()
            r = requests.post(url=url, headers=headers, json=body, proxies=proxies, verify=False)
            Common.logger().info("小时榜写入数据:{}", r.json()["msg"])
        except Exception as e:
            Common.logger().error("小时榜写入数据异常:{}", e)

    # 合并单元格
    @classmethod
    def merge_cells(cls, sheetid, ranges):
        """
        合并刚插入的单元格 G1:I1
        sheetid:哪张工作表
        ranges:需要合并的单元格范围
        """
        url = "https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/" + cls.spreadsheetToken + "/merge_cells"
        headers = {
            "Authorization": "Bearer " + cls.get_token(),
            "Content-Type": "application/json; charset=utf-8"
        }

        body = {
            "range": sheetid + "!" + ranges,
            "mergeType": "MERGE_ROWS"
        }

        try:
            urllib3.disable_warnings()
            r = requests.post(url=url, headers=headers, json=body, proxies=proxies, verify=False)
            Common.logger().info("合并单元格:{}", r.json()["msg"])
        except Exception as e:
            Common.logger().error("合并单元格异常:{}", e)

    # 读取单元格数据
    @classmethod
    def get_range_value(cls, sheetid, cell):
        """
        读取单元格内容
        :param sheetid: 哪张工作表
        :param cell: 哪个单元格
        :return: 单元格内容
        """
        url = "https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/" \
              + cls.spreadsheetToken + "/values/" + sheetid + "!" + cell
        headers = {
            "Authorization": "Bearer " + cls.get_token(),
            "Content-Type": "application/json; charset=utf-8"
        }
        params = {
            # valueRenderOption=ToString 可返回纯文本的值(数值类型除外)；
            # valueRenderOption=FormattedValue 计算并格式化单元格；
            # valueRenderOption=Formula 单元格中含有公式时返回公式本身；
            # valueRenderOption=UnformattedValue 计算但不对单元格进行格式化。
            "valueRenderOption": "FormattedValue",

            # dateTimeRenderOption=FormattedString 计算并对时间日期按照其格式进行格式化，但不会对数字进行格式化，返回格式化后的字符串。
            "dateTimeRenderOption": "",

            # 返回的用户id类型，可选open_id,union_id
            "user_id_type": "open_id"
        }
        try:
            urllib3.disable_warnings()
            r = requests.get(url=url, headers=headers, params=params, proxies=proxies, verify=False)
            # print(url)
            return r.json()["data"]["valueRange"]["values"][0]
        except Exception as e:
            Common.logger().error("读取单元格数据异常:{}", e)

    # 删除行或列，可选 ROWS、COLUMNS
    @classmethod
    def dimension_range(cls, sheetid, major_dimension, startindex, endindex):
        """
        删除行或列
        :param sheetid:工作表
        :param major_dimension:默认 ROWS ，可选 ROWS、COLUMNS
        :param startindex:开始的位置
        :param endindex:结束的位置
        :return:
        """
        url = "https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/" + cls.spreadsheetToken + "/dimension_range"
        headers = {
            "Authorization": "Bearer " + cls.get_token(),
            "Content-Type": "application/json; charset=utf-8"
        }
        body = {
            "dimension": {
                "sheetId": sheetid,
                "majorDimension": major_dimension,
                "startIndex": startindex,
                "endIndex": endindex
                }
            }
        try:
            urllib3.disable_warnings()
            r = requests.delete(url=url, headers=headers, json=body, proxies=proxies, verify=False)
            Common.logger().info("删除视频数据:{}", r.json()["msg"])
        except Exception as e:
            Common.logger().error("删除视频数据异常:{}", e)


if __name__ == "__main__":
    feishu = Feishu()

    # # 获取飞书api token
    # feishu.get_token()
    # 获取表格元数据
    # feishu.get_metainfo()

    # 读取工作表中所有数据
    # print(feishu.get_values_batch("k2rKkv"))
    # print(len(feishu.get_values_batch("k2rKkv")))
    # for i in range(3, len(feishu.get_values_batch("k2rKkv"))+1):
    #     print(feishu.get_range_value("k2rKkv", "A" + str(i) + ":" + "A" + str(i))[0])
    #     print(feishu.update_hour_list_values("k2rKkv", "G" + str(i) + ":" + "H" + str(i), [["333"]]))
    #     time.sleep(0.5)
    # feishu.get_range_value("k2rKkv", "F3:F3")[0]
    print(type(feishu.get_range_value("k2rKkv", "H19:H19")[0].split(" ")[-1].split(":")[0]))
    print(feishu.get_range_value("k2rKkv", "H19:H19")[0])
    print(int(feishu.get_range_value("k2rKkv", "H19:H19")[0].split(" ")[-1].split(":")[0]))

    feishu.update_hour_list_values("k2rKkv", "H12:H12", [["2022-05-17 15:14:27"]])

    # 看一看+工作表，插入首行
    # print(feishu.insert_columns("k2rKkv", "COLUMNS", 6, 9))
    # print(feishu.update_hour_list_values("k2rKkv"))
    # print(feishu.merge_cells("k2rKkv", "G1:I1"))
    #
    # # 看一看+工作表，首行写入数据
    # print(feishu.update_values("Y8N3Vl", "a1", "b1", "c1", "d1", "e1", "f1", "g1",
    #                            "h1", "i1", "j1", "k1", "l1", "m1", "n1", "o1"))

    # 查询单元格内容
    # print(type(feishu.get_range_value("k2rKkv", "G1:G1")[0]))
    #
    # 删除行或列，可选 ROWS、COLUMNS
    # feishu.dimension_range("k2rKkv", "ROWS", 3, 3)

    pass