Python实现Excel数据同步到飞书文档

2025-03-27 10:45 开发作者：失败尽常态523

一、整体目标

核心功能：自动将Excel数据同步到飞书文档的末尾，并添加时间戳。

应用场景：

每日销售数据报表自动更新到团队文档
周报自动化生成
实时数据看板同步

二、代码结构拆解

# 基础设施层
import pandas as pd        # 数据处理
import requests           # 网络请求
import logging            # 错误追踪
from datetime import datetime # 时间处理
 
# 配置层
logging配置 → API密钥配置
 
# 功能层
get_access_token() → 获取系统通行证
insert_to_feishu() → 内容插入逻辑
 
# 执行层
main() → 流程控制器

三、核心逻辑讲解（重点）

用「快递送货」做类比：

1. 建立安全连接（获取access_token）

好比：快递员需要先获取小区门禁卡

流程：

准备身份证（APP_ID）和密码（APP_SECRET）
到门卫处（飞书认证接口）验证身份
获得临时通行证（tenant_access_token）

response = requests.post(url, json=data)  # 发送验证请求
return response.json()["tenant_access_token"]  # 提取通行证

2. 定位文档位置

好比：找到要送货的楼层和房间

两步作：

1.获取整栋楼结构（获取文档所有内容块）

blocks = requests.get(blocks_url).json()["data"]["items"]

2.找到最后一间房（定位最后一个内容块）

last_block_id = blocks[-1]["block_id"]  # 列表末尾元素

3. 数据包装与投递

好比：打包货物并送货

包装阶段：

# 添加醒目时间标签
时间戳 = datetime.now().strftime(...)
# 将Excel转为Markdown表格
markdown_table = df.to_markdown(index=False)

投递作：

requests.post(insert_url, json={
    "content": "**时间标签** + 表格数据",
    "position": "after"  # 放在最后
})

四、异常处理机制

用「行车记录仪」做类比：

日志配置（相当于安装记录仪）

logging.basicConfig(
    filename='feishu_sync.log',  # 记录文件
    level=logging.INFO,          # 记录级别
    javascriptformat='时间-级别-信息'       # 记录格式
)

关键节点监控（示例）：

try:
    发送请求...
except Exception as e:
    logging.error(f"插入失败: {e}")  # 记录错误快照

五、函数讲解

get_access_token()

def get_access_token():
    """获取飞书API的access_token"""
    # 定义API请求地址
    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    # 设置HTTP请求头（告诉服务器我们发送的是JSON数据）
    headers = {"Content-Type": "application/json"}
"""
服务器会根据 Content-Type 的值，调用对应的解析方式。例如：
如果是 application/json，服务器会用 JSON 解析器处理数据。
如果是 application/x-www-form-urlencoded，服务器会按表单格式解析数据。
"""
    # 准备请求数据（飞书应用的ID和密钥）
    data = {"app_id": Ahttp://www.devze.comPP_ID, "app_secret": APP_SECRET}
    
    try:
        # 发送POST请求到飞书服务器
        response = requests.post(url, headers=headers, json=data)
        # 检查响应状态码（如果不是200，会抛出异常）
        response.raise_for_status()
        # 解析返回的JSON数据，提取访问令牌
        return response.json()["tenant_access_token"]
    
    except Exception as e:
        # 记录错误日志
        logging.error(f"获取access_token失败: {e}")
        # 失败时返回None
        return None

关键概念解释

1. 飞书API访问令牌 (tenant_access_token)

这是飞书API的“通行证”，调用其他API（如发送消息、读取数据）时必须携带它。
令牌有效期通常为2小时，需要定期重新获取。

2. HTTP请求

POST：一种HTTP方法，表示向服务器提交数据（这里是提交和）。app_idapp_secret
headers：请求头，表示发送的数据是JSON格式。Content-Type: application/json
data：实际发送的数据，包含飞书应用的凭证。

3. APP_ID 和 APP_SECRET

这两个值需要从飞书开放平台注册应用后获取。
非常重要！相当于你的应用账号密码，需严格保密

4. 异常处理 (try-except)

    try:
        response = requests.post(url, headers=headers, json=data)
        response.raise_for_status()
        return response.json()["tenant_access_token"]
 
    except Exception as e:
        logging.error(f"获取access_token失败: {e}")
        return None

1. response = requests.post(url, headers=headers, json=data)

作用：向指定的 url 发送一个 HTTP POST 请求。

参数说明：

url: 目标 API 接口地址（例如飞书的 tenant_access_token 获取接口）。

headers: 请求头，通常包含 Content-Type 等信息（如之前设置的 application/json）。

json=data: 自动将 data 对象（字典）序列化为 JSON 字符串，并设置 Content-Type: application/json。

关键点：

使用 json=data 比手动转换数据更简洁（等价于 data=json.dumps(data) + 设置请求头）。

2. response.raise_for_status()

作用：检查 HTTP 响应状态码，如果状态码表示错误（如 4xx 客户端错误、5xx 服务器错误），直接抛出异常。

常见状态码：

200 OK: 请求成功。
400 Bad Request: 请求参数错误。
401 Unauthorized: 身份验证失败。
500 Internal Server Error: 服务器内部错误。

意义：

如果服务器返回错误状态码（如飞书接口返回 {"code": 9999, "msg": "invalid params"}），此方法会抛出 HTTPError 异常，阻止后续代码执行，避免处理错误数据。

3. return response.json()["tenant_access_token"]

作用：

response.json(): 将 HTTP 响应内容解析为 python 字典（前提是响应内容是 JSON 格式）。

["tenant_access_token"]: 从解析后的字典中提取 tenant_access_token 字段的值并返回。

关键点：

假设飞书接口返回的 JSON 结构为 {"code":0, "tenant_access_token":"xxx", "expire":7200}。

如果响应不是合法的 JSON，response.json() 会抛出 JSONDecodeError。

4.except Exception as e:

作用：捕获所有类型的异常（包括 HTTPError、JSONDecodeError、网络超时等）。

细节：

logging.error(...): 记录错误日志（比 print 更规范，可输出到文件或监控系统）。

return None: 返回 None 表示获取 access_token 失败，调用方可通过判断返回值是否为 None 处理错误。

完整逻辑流程图

开始
│
├─ 发送 POST 请求 → 成功?
│ ├─ 是 → 检查状态码 → 正常?
│ │ ├─ 是 → 解析 JSON → 返回 tenant_access_token
│ │ └─ 否 → 抛出 HTTPError
│ └─ 否 → 抛出异常（如网络错误）
│
└─ 捕获异常 → 记录日志 → 返回 None

insert_to_feishu()

1.获取访问令牌

access_token = get_access_token()

调用 get_access_token() 函数获取飞书API的访问令牌（需提前实现）。

若获取失败（如认证错误），直接返回 False。

2.获取文档块列表

blocks_url = f"https://open.feishu.cn/open-apis/docx/v1/documents/{DOCUMENT_ID}/blocks"
headers = {"Authorization": f"Bearer {access_token}"}
blocks_response = requests.get(blocks_url, headers=headers)

通过飞书API获取文档的块（Block）结构。块是文档内容的组成单元（如段落、标题、表格等）。

DOCUMENT_ID 是目标文档的唯一标识，需提前从飞书后台获取。

 headers = {"Authorization": f"Bearer {access_token}"}

这行代码用于构造一个 HTTP 请求头（Request Headers），目TudthZ的是向飞书API服务端证明当前请求的合法性。

认证机制

Bearer Token：这是现代API认证的常见方式，属于OAuth 2.0协议的一部分。

功能：通过将 access_token（访问令牌）放在请求头中，告知飞书服务器“持有此令牌的用户/应用有权访问该API”。

类比：类似于现实中的“门禁卡”，只有持有有效卡（access_token）的人才能通过门禁（API权限校验）。

格式说明

键值对结构：

键（Key）: Authorization

值（Value）: Bearer {access_token}

（将 access_token 的值替换到 {access_token} 的位置）

示例：

# 假设 access_token = "abc123xyz"
headers = {"Authorization": "Bearer abc123xyz"}

技术细节

为什么必须用 Bearer 前缀？

这是OAuth 2.0的标准规范，用于明确令牌类型。服务端会根据前缀判断认证方式。

access_token 从何而来？

通常通过飞书开放平台的身份认证流程生成，例如：

用户登录后授权获取
应用自身体验的令牌（如机器人或服务端API调用

实际应用场景

在飞书API请求中，所有需要权限的操作（如读写文档、获取用户信息等）都必须携带此头。例如：

# 发送请求时附加 headers
response = requests.get(url, headers=headers)

3.飞书文档中获取所有内容块

这三行代码用于从飞书文档中获取所有内容块（Blocks），是操作飞书文档的核心步骤。

发送HTTP GET请求

blocks_response = requests.get(blocks_url, headers=headers)

功能：

向飞书API发送一个 GET请求，目标地址是 blocks_url（例如 https://open.feishu.cn/open-apis/docx/v1/documents/{document_id}/blocks）。

参数说明：

headers=headers：携带认证头（包含 Authorization: Bearer {access_token}），用于权限验证。

返回值：

blocks_response 是一个 Response 对象，包含服务端返回的原始HTTP响应。

检查HTTP响应状态

blocks_response.raise_for_status()

功能：

自动检测HTTP响应状态码。如果状态码是 4xx（客户端错误）或 5xx（服务端错误），直接抛出异常（如 HTTPError），终止后续代码执行。

为什么需要它？

避免在请求失败时继续处理无效数据（例如令牌过期、URL错误、权限不足等）。

典型错误场景：

401 Unauthorized：access_token 失效或未传。
404 Not Found：blocks_url 地址错误或文档不存在。

解析JSON数据并提取内容块

blocks = blocks_response.json()["data"]["items"]

将HTTP响应的原始内容（JSON格式）解析为Python字典，并提取其中的文档块列表。

数据结构（以飞书文档API为例）：

{
  "code": 0,
  "data": {
    "items": [
      {"block_id": "xxxx", "content": "文本内容1"},
      {"block_id": "yyyy", "content": "文本内容2"}
    ]
  }
}

它的操作可以拆解为以下步骤：

blocks_response.json()：将 HTTP 响应的原始内容（JSON 格式）转换为 Python 的字典（dict）或列表（list）。

["data"]：从 JSON 字典中提取键为 "data" 的值（通常包含核心数据）。

["items"]：从 "data" 对应的字典中，进一步提取键为 "items" 的值（通常是一个列表，包含多个数据项）。

结果：

blocks 变量存储所有文档块的列表，后续可用于遍历、修改或插入新内容。

示例：

Python实现Excel数据同步到飞书文档

当我们将这个表格转换为 Python 的字典列表时，结果如下：

data = {
    "code": 0,
    "message": "成功",
    "data": {
        "items": [
            {"姓名": "张三", "年龄": 25, "城市": "北京"},
            {"姓名": "李四", "年龄": 30, "城市": "上海"}
        ]
    }
}
 
# 提取表格中的每一行（即 items 列表）
items = data["data"]["items"]

此时 items 的值是：

[
{"姓名": "张三", "年龄": 25, "城市": "北京"},
{"姓名": "李四", "年龄": 30, "城市": "上海"}
]

类比解释

data["data"]

类似于 Excel 表格的整个数据区域（包含表头和所有行），对应代码中 data 字段的内容。

data["data"]["items"]

类似于 Excel 表格中所www.devze.com有行的数据（每一行是一个字典），对应代码中的 items 列表。

最终结果 blocks

就是一个列表，每个元素代表一个独立的“块”（对应 Excel 中的一行），例如：

blocks = [
    {"block_id": "段落1", "text": "这是第一段内容"},
    {"block_id": "图片1", "url": "https://example.com/image.jpg"}
]

4.定位最后一个块的ID

last_block_id = blocks[-1]["block_id"] if blocks else None

从返回的块列表中提取最后一个块的 block_id。

若文档为空（无任何块），last_block_id 为 None，可能导致后续插入失败。

5.插入新内容

insert_url = f"https://open.feishu.cn/open-apis/docx/v1/documents/{DOCUMENT_ID}/blocks/{last_block_id}/insert"
data = {
    "content": f"**数据更新时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}**\n{content}",
    "position": "after"
}
insert_response = requests.post(insert_url, headers=headers, json=data)

在最后一个块之后插入新内容。

content 字段包含带时间戳的加粗标题（Markdown语法）和传入的 content 参数。

飞书API支持富文本格式，此处使用 ** 表示加粗，实际可能需要根据API要求调整格式

"content": f"**数据更新时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}**\n{content}",
"position": "after" # 在最后一个块后插入

这是**字典（dict）**中的两个键值对，通常用于配置内容插入逻辑（例如生成报告、更新文档等场景）。

(1) "content": ...

功能：定义要插入的文本内容，包含动态时间和原始内容。

细节：

f"..."：使用 Python 的 f-string 动态生成字符串。
datetime.now()：获取当前时间。
strftime('%Y-%m-%d %H:%M:%S')：将时间格式化为年-月-日时:分:秒（例如 2023-10-05 14:30:00）。
**数据更新时间：...**：用 Markdown 的粗体语法包裹时间戳，使其突出显示。
\n{content}：换行后追加原始内容（假设 content 是已定义的变量）。

(2) "position": "after"

功能：指定插入位置为目标位置之后（例如在文档的最后一个段落/区块后追加内容）。

常见值：

"after"：在指定位置后插入。
"before"：在指定位置前插入。
"replace"：替换原有内容。

实际输出示例

假设 content 的值为 "这是原始内容"，则生成的 "content" 结果为：

效果：在原始内容前添加时间戳，并用粗体显示。

**数据更新时间：2023-10-05 14:30:00**
这是原始内容

 insert_response = requests.post(insert_url, headers=headers, json=data)
        insert_response.raise_for_status()
        logging.info("数据插入成功！")

这段代码用于向指定的URL发送HTTP POST请求，插入数据并处理响应。

6.异常处理

try:
    # ...请求代码...
except Exception as e:
    logging.error(f"插入数据失败: {e}")
    return False

捕获网络错误、API响应错误等异常，记录日志并返回 False。

六、main()

def main():
    """主函数：读取Excel并同步到飞书"""
    try:
        # 读取Excel数据
        df = pd.read_excel("data.xlsx")  # 替换为你的Excel路径
        markdown_table = df.to_markdown(index=False)
 
        # 插入到飞书文档
        if insert_to_feishu(markdown_table):
            print("同步成功！")
        else:
            print("同步失败，请查看日志文件。")
    except Exception as e:
        logging.error(f"主流程异常: {e}")
markdown_table = df.to_markdown(index=False)

代码功能

作用：将 pandas 的 DataFrame (df) 转换为 Markdown 格式的表格字符串，并保存到变量 markdown_table 中。
参数 index=False：表示生成的表格中不包含行索引列（默认会显示索引）。

示例与输出

假设有一个 DataFrame：

import pandas as pd
 
data = {
        "Name": ["Alice", "Bob",  "Charlie"],
    "Age":  [25,  30, 35],
        "City": ["New York",  "London",  "Tokyo"]
}
df  =  pd.DataFrame(data)

执行代码：

markdown_table  =  df.to_markdown(index=False)
print(markdown_table)

输出结果：

| Name | Age | City |
|---------|-------|----------|
| Alice | 25 | New York |
| Bob | 30 | London |
| Charlie | 35 | Tokyo |

Python实现Excel数据同步到飞书文档

七、相关库函数

1. import pandas as pd

用途：数据分析和处理（表格数据操作）。

常用函数/类：

pd.read_csv('file.csv')：读取 CSV 文件生成 DataFrame。
df.to_csv('output.csv')：将 DataFrame 保存为 CSV 文件。
df.head(n)：显示 DataFrame 前 n 行（默认 5 行）。

df.merge(df2, on='key')：按列合并两个 DataFrame。 - df.groupby('column').sum()：按列分组聚合数据。 - pd.DataFrame(data)：从字典/列表创建 DataFrame。

#  示例：读取  CSV  并处理数据
data = pd.read_csv("data.csv")
filtered_data  =  data[data["age"] >  18]
filtered_data.to_csv("adults.csv")

2. import requests

用途：发送 HTTP 请求（如 API 调用、网页抓取）。

常用函数/方法：

requests.get(url, params={})：发送 GET 请求。

requests.post(url, data={})：发送 POST 请求。 - response.json()：将响应内容解析为 JSON。 - response.status_code：获取 HTTP 状态码（如 200 表示成功）。 - response.text：获取原始文本响应内容。

#  示例：调用 API  获取数据
response  =  requests.get("https://api.example.com/data")
if  response.status_code  ==  200:
        data  = response.json()
     print(data["results"])

3. import logging

用途：记录程序运行日志（调试、警告、错误等）。

常用函数/配置：

logging.basicConfig(level=logging.INFO)：设置日志级别。
logging.info('message')：记录一般信息。
logging.warning('message')：记录警告。
logging.error('message')：记录错误。
logger = logging.getLogger(__name__)：创建自定义日志记录器。

#  示例：记录程序运行状态
logging.basicConfig(level=logging.INFO)
logging.info("程序启动")
try:
        #  某些操作
except Exception as  e:
        logging.error(f"发生错误:  {e}")

4. from datetime import datetime

用途：处理日期和时间。

常用函数/方法：

datetime.now()：获取当前时间。
datetime.strftime('%Y-%m-%d')：格式化时间为字符串。
datetime.strptime('2023-10-05', '%Y-%m-%d')：将字符串解析为时间对象。
datetime.timestamp()：获取时间戳（秒数）。

#  示例：生成带时间戳的日志
current_time =  datetime.now().strftime("%Y-%m-%d  %H:%M:%S")
print(f"操作时间:  {current_time}")

八、完整代码

import pandas as pd
import requests
import logging
from datetime import datetime
 
# 配置日志记录
logging.basicConfig(
    filename='feishu_sync.log',
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
"""
1.logging.basicConfig(...)
logging 是 Python 标准库中的日志模块，用于记录程序的运行状态，方便调试和排查问题。
basicConfig() 方法用于配置日志系统的基本设置。
2.filename='feishu_sync.log'
指定日志输出的文件名为 feishu_sync.log，所有日志信息都会被写入该文件。
这样可以在程序运行后，查看 feishu_sync.log 文件来了解程序的执行情况。
3.level=logging.INFO
设置日志的最低级别为 INFO，即只记录 INFO 及更高级别的日志（包括 WARNING、ERROR 和 CRITICAL）。
低于 INFO 级别的 DEBUG 级日志不会被记录。
4.format='%(asctime)s - %(levelname)s - %(message)s'
指定日志的输出格式，其中：
%(asctime)s：时间戳，表示日志记录的时间（格式如 2025-02-18 12:34:56）。
%(levelname)s：日志级别，如 INFO、ERROR 等。
%(message)s：具体的日志消息内容。
"""
 
 
# 飞书API配置
APP_ID = "your_app_id"
APP_SECRET = "your_app_secret"
DOCUMENT_ID = "Gu4LwF32Ci0OUakZo50cQ8p2njk"  # 目标云文档的ID
 
 
def get_access_token():
    """获取飞书API的access_token"""
    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    headers = {"Content-Type": "application/json"}
    data = {"app_id": APP_ID, "app_secret": APP_SECRET}
    try:
        response = requests.post(url, headers=headers, json=data)
        response.raise_for_status()
        return response.json()["tenant_access_token"]
 
    except Exception as e:
        logging.error(f"获取access_token失败: {e}")
        return None
"""
requests.post(url, headers=headers, json=data)：使用 requests 库发送一个 POST 请求到指定的 URL，并携带请求头和请求体数据。json=data 会自动将字典 data 转换为 JSON 格式的字符串。
response.raise_for_status()：检查响应的状态码，如果状态码不是 200（表示请求成功），则抛出 HTTPError 异常。
response.json()["tenant_access_token"]：将响应的内容解析为 JSON 格式，并返回其中的 tenant_access_token 字段，即获取到的租户访问令牌。
except Exception as e:：捕获所有异常，并将异常对象赋值给变量 e。
logging.error(f"获取access_token失败: {e}")：使用 logging 模块记录错误信息，方便后续排查问题。
return None：如果发生异常，返回 None 表示获取令牌失败。
"""
 
def insert_to_feishu(content):
    """向飞书文档插入内容"""
    access_token = get_access_token()
    if not accejsss_token:
        return False
 
    # 步骤1：获取文档块列表，找到最后一个块的位置
    blocks_url = f"https://open.feishu.cn/open-apis/docx/v1/documents/{DOCUMENT_ID}/blocks"
    headers = {"Authorization": f"Bearer {access_token}"}
 
    try:
        # 获取文档结构
        blocks_response = requests.get(blocks_url, headers=headers)
        blocks_response.raise_for_status()
        blocks = blocks_response.json()["data"]["items"]
 
        # 找到最后一个块的ID（用于插入位置）
        last_block_id = blocks[-1]["block_id"] if blocks else None
 
        # 步骤2：插入新内容到文档末尾
        insert_url = f"https://open.feishu.cn/open-apis/docx/v1/documents/{DOCUMENT_ID}/blocks/{last_block_id}/insert"
        data = {
            "content": f"**数据更新时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}**\n{content}",
            "position": "after"  # 在最后一个块后插入
        }
        insert_response = requests.post(insert_url, headers=headers, json=data)
        insert_response.raise_for_status()
        logging.info("数据插入成功！")
        return True
    except Exception as e:
        logging.error(f"插入数据失败: {e}")
        return False
 
 
def main():
    """主函数：读取Excel并同步到飞书"""
    try:
        # 读取Excel数据
        df = pd.read_excel("data.xlsx")  # 替换为你的Excel路径
        markdown_table = df.to_markdown(index=False)
 
        # 插入到飞书文档
        if insert_to_feishu(markdown_table):
            print("同步成功！")
        else:
            print("同步失败，请查看日志文件。")
    except Exception as e:
        logging.error(f"主流程异常: {e}")
 
 
if __name__ == "__main__":
    main()

以上就是Python实现Excel数据同步到飞书文档的详细内容，更多关于Python Excel数据同步到飞书的资料请关注编程客栈(www.devze.com)其它相关文章！

继续阅读：Python Excel数据 Python Excel数据同步 Python Excel数据同步到飞书

Python实现Excel数据同步到飞书文档

目录

一、整体目标

二、代码结构拆解

三、核心逻辑讲解（重点）

1. 建立安全连接（获取access_token）

2. 定位文档位置

四、异常处理机制

五、函数讲解

get_access_token()

insert_to_feishu()

六、main()

七、相关库函数

1. import pandas as pd

2. import requests

3. import logging

4. from datetime import datetime

八、完整代码

更多精彩内容

精彩评论

最新开发

Spring AOP实现断路器方式

Java 参数值传递机制的实现

Java常用的Arrays函数举例详解

使用Java实现Navicat密码的加密与解密的代码解析

go如何调用C动态库函数

开发排行榜

springboot后端存储富文本内容的思路与步骤(含图片内容)

PyCharm运行python测试,报错“没有发现测试”/“空套件”的解决

return base64.b64encode(b).decode(

基于C语言实现钻石棋游戏的示例代码

Sublime Text 3解决中文乱码问题（实测可用）