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

目录

一、整体目标

二、代码结构拆解

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

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

2. 定位文档位置

3. 数据包装与投递

四、异常处理机制

五、函数讲解

get_access_token()

关键概念解释

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

2. HTTP请求

3. APP_ID 和 APP_SECRET

4. 异常处理 (try-except)

完整逻辑流程图

insert_to_feishu()

1.获取访问令牌

2.获取文档块列表

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

4.定位最后一个块的ID

5.插入新内容

6.异常处理

main()

​编辑

相关库函数

1. import pandas as pd

2. import requests

3. import logging

4. from datetime import datetime

完整代码

---

一、整体目标

核心功能：自动将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,          # 记录级别format='时间-级别-信息'       # 记录格式
)

关键节点监控（示例）：

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": APP_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}")# 失败时返回Nonereturn 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），目的是向飞书API服务端证明当前请求的合法性。

1. 认证机制

• Bearer Token：这是现代API认证的常见方式，属于OAuth 2.0协议的一部分。
• 功能：通过将 access_token（访问令牌）放在请求头中，告知飞书服务器“持有此令牌的用户/应用有权访问该API”。
• 类比：类似于现实中的“门禁卡”，只有持有有效卡（access_token）的人才能通过门禁（API权限校验）。

2. 格式说明

• 键值对结构：
  • 键（Key）: Authorization
  • 值（Value）: Bearer {access_token}
    （将 access_token 的值替换到 {access_token} 的位置）

示例：

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

3. 技术细节

• 为什么必须用 Bearer 前缀？
  这是OAuth 2.0的标准规范，用于明确令牌类型。服务端会根据前缀判断认证方式。
• access_token 从何而来？
  通常通过飞书开放平台的身份认证流程生成，例如：
  • 用户登录后授权获取
  • 应用自身体验的令牌（如机器人或服务端API调用

4. 实际应用场景

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

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

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

blocks_response = requests.get(blocks_url, headers=headers)
blocks_response.raise_for_status()
blocks = blocks_response.json()["data"]["items"]

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

1. 发送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响应。

2. 检查HTTP响应状态

blocks_response.raise_for_status()

• 功能：
  自动检测HTTP响应状态码。如果状态码是 4xx（客户端错误） 或 5xx（服务端错误），直接抛出异常（如 HTTPError），终止后续代码执行。
• 为什么需要它？
  • 避免在请求失败时继续处理无效数据（例如令牌过期、URL错误、权限不足等）。
  • 典型错误场景：
    • 401 Unauthorized：access_token 失效或未传。
    • 404 Not Found：blocks_url 地址错误或文档不存在。

3. 解析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 的字典列表时，结果如下：

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

此时 items 的值是：

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

类比解释

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

2. data["data"]["items"]
   类似于 Excel 表格中所有行的数据（每一行是一个字典），对应代码中的 items 列表。

3. 最终结果 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 pddata = {"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    |

相关库函数

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"  # 目标云文档的IDdef 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 access_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 Trueexcept Exception as e:logging.error(f"插入数据失败: {e}")return Falsedef 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()