目录
一、整体目标
二、代码结构拆解
三、核心逻辑讲解（重点）
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}")
# 失败时返回None
return None关键概念解释
1. 飞书API访问令牌 (tenant_access_token)
- 这是飞书API的“通行证”，调用其他API（如发送消息、读取数据）时必须携带它。
- 令牌有效期通常为2小时，需要定期重新获取。
2. HTTP请求
POST：一种HTTP方法，表示向服务器提交数据（这里是提交和）。app_idapp_secretheaders：请求头， 表示发送的数据是JSON格式。Content-Type: application/jsondata：实际发送的数据，包含飞书应用的凭证。
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 None1. 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。
- 假设飞书接口返回的 JSON 结构为
4.except Exception as e:
- 作用：捕获所有类型的异常（包括
HTTPError、JSONDecodeError、网络超时等）。 - 细节：
logging.error(...): 记录错误日志（比print更规范，可输出到文件或监控系统）。return None: 返回None表示获取access_token失败，调用方可通过判断返回值是否为None处理错误。
完整逻辑流程图
开始
│
├─ 发送 POST 请求 → 成功?
│ ├─ 是 → 检查状态码 → 正常?
│ │ ├─ 是 → 解析 JSON → 返回 tenant_access_token
│ │ └─ 否 → 抛出 HTTPError
│ └─ 否 → 抛出异常（如网络错误）
│
└─ 捕获异常 → 记录日志 → 返回 Noneinsert_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}的位置）
- 键（Key）:
示例：
# 假设 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, "城市": "上海"}
]类比解释
-
data["data"]
类似于 Excel 表格的整个数据区域（包含表头和所有行），对应代码中data字段的内容。 -
data["data"]["items"]
类似于 Excel 表格中所有行的数据（每一行是一个字典），对应代码中的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 |
相关库函数
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 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 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()
