成为开发者

自己创建一个企业， 或者让你所在的公司给予你的账号一个权限 钉钉开发者后台登录

创建一个应用

如图，
这里只需要

1. Client ID (原 AppKey 和 SuiteKey)
2. Client Secret (原 AppSecret 和 SuiteSecret)
   这两个东西 获取 access_token
   然后还需要一个userid 提供的operatorId：表示待办事项的操作者的ID，是一个唯一的标识符。
   这个可以在钉钉提供的API接口直接获取到， 目前来看是不会变化的

后面就是源码了

import requests
import json
import  pandas as pd
import datetime




class Ding():
    """
    钉钉表格推送数据
    文档信息
    https://open.dingtalk.com/document/orgapp/update-cell-properties

    """
    def __init__(self,appkey,appsecret,operatorid,sheetid):

        #获取access_token
        access_token = requests.get(f"https://oapi.dingtalk.com/gettoken?"\
                                    f"appkey={appkey}&appsecret={appsecret}",\
                                    timeout=5).json()["access_token"]
        #设置基础的请求头
        self.default_settings = {
            'headers':{'Host':'api.dingtalk.com',
                    'x-acs-dingtalk-access-token':access_token,
                    'Content-Type':'application/json'},
            'verify': False,
            'timeout': 20}
        self.primary_domain = "https://api.dingtalk.com/v1.0/doc/workbooks/"
        self.operatorid = operatorid
        self.sheetid= sheetid



    def get(self,sheetname,get_range):
        """ 单独获取单元格内容 """
        url =   f"{self.primary_domain}{self.sheetid}"\
                f"/sheets/{sheetname}/ranges/{get_range}"\
                f"?operatorId={self.operatorid}"

        res = requests.get(url,**self.default_settings)
        return res.json()["displayValues"]

    def put(self,sheetname,wt_range,content):
        """ 单独写入单元格内容 """
        url =   f"{self.primary_domain}{self.sheetid}"\
                f"/sheets/{sheetname}/ranges/{wt_range}"\
                f"?operatorId={self.operatorid}"
        json1 = {"values" : [content]}
        res = requests.put(url,data=json.dumps(json1),**self.default_settings)
        if res.status_code != 200:
            print("钉钉表格推送失败")
            print(res.text)

    def clear(self,sheetname,wt_range):
        """ 清空单元格全部 """
        url =   f"{self.primary_domain}{self.sheetid}"\
                f"/sheets/{sheetname}/ranges/{wt_range}/clear"\
                f"?operatorId={self.operatorid}"
        res = requests.post(url,**self.default_settings)
        if res.status_code != 200:
            print("钉钉表格推送失败")
            print(res.text)
    def cleardata(self,sheetname,wt_range):
        """ 清空单元格数据 """
        url =   f"{self.primary_domain}{self.sheetid}"\
                f"/sheets/{sheetname}/ranges/{wt_range}/clearData"\
                f"?operatorId={self.operatorid}"
        res = requests.post(url,**self.default_settings)
        if res.status_code != 200:
            print("钉钉表格推送失败")
            print(res.text)

    def put_df_to_ding(self,sheetname:str,df_data:pd.DataFrame,wt_range:str,is_mater_head=False):
        """ 写入dataframe数据到 钉钉表格中 """

        letters = list(string.ascii_uppercase)  +   [letters1 + letters2 \
                                                    for letters1 in string.ascii_uppercase\
                                                    for letters2 in string.ascii_uppercase]
        # 写入表格前面的字母
        wt_range_latter = ''.join([char for char  in wt_range if char.isalpha()])
        # 写入表格后面的数字
        wt_range_numberr = int(''.join([char for char  in wt_range if char.isdigit()]))
        # 写入表格的宽度
        df_width = df_data.shape[1]
        # 写入表格的起始和结束位置的字母
        left_letter = wt_range_latter
        right_letter = letters[letters.index(left_letter)+df_width-1]


        def meterhead(df_data:pd.DataFrame):
            """ 写入表头 """
            url =   f"{self.primary_domain}{self.sheetid}"\
                    f"/sheets/{sheetname}/ranges/"\
                    f"{left_letter}{wt_range_numberr}:{right_letter}{wt_range_numberr}"\
                    f"?operatorId={self.operatorid}"
            input_data = {"values" : [df_data.columns.tolist()]}
            res = requests.put(url,data=json.dumps(input_data),**self.default_settings)
            if res.status_code != 200:
                print("钉钉表格推送失败")
                print(res.text)
        def put_data(df_data:pd.DataFrame,start_index,end_index):
            """ 写入表格数据,  """
            #如果需要写入表头 就需要往下移动一格 就将offset设置为1
            offset = 1 if is_mater_head else 0

            url =   f"{self.primary_domain}{self.sheetid}"\
                    f"/sheets/{sheetname}/ranges/"\
                    f"{left_letter}{wt_range_numberr+start_index+offset}:"\
                    f"{right_letter}{wt_range_numberr+end_index+offset}"\
                    f"?operatorId={self.operatorid}"
            input_data = {"values" :\
                        df_data.astype("str").iloc[start_index:end_index+1].values.tolist()}
            res = requests.put(url,data=json.dumps(input_data),**self.default_settings)
            if res.status_code != 200:
                print("钉钉表格推送失败")
                print(res.text)
        def cycle(df:pd.DataFrame):
            df_height = df.shape[0]
            width_per_cycle = 1000
            num_segments = df_height // width_per_cycle
            for i in range(num_segments):
                start = i * width_per_cycle
                end = (i + 1) * width_per_cycle-1
                put_data(df,start,end)
            if df_height % width_per_cycle != 0:
                start = num_segments * width_per_cycle
                end = df_height-1
                put_data(df,start,end)


        if is_mater_head:
            meterhead(df_data)
        cycle(df_data)


if __name__ == "__main__":

    DING =   Ding(appkey = "appkey",
                  appsecret ="appsecret",
                  operatorid = "operatorid",
                  sheetid= "表格ID",)
    df_test = pd.DataFrame({"A": [x for x in range(1,1002)], "B": [x for x in range(1,1002)]})
    DING.put_df_to_ding(sheetname="Sheet1",df_data=df_test,wt_range="A15",is_mater_head=False)
    DING.cleardata(sheetname="Sheet1",wt_range="A15:C15")