John Qiu
712063071c
skills/ → skills-dev(9), skills-req(10), skills-ops(4), skills-integration(8), skills-biz(4), skills-workflow(7) generate-marketplace.py 改为自动扫描所有 skills-* 目录。 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
240 lines
7.2 KiB
Python
240 lines
7.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
深入调试飞书云文档图片上传
|
|
"""
|
|
|
|
import requests
|
|
import os
|
|
from datetime import datetime, timedelta
|
|
|
|
ZHIYUN_APP_ID = "cli_a9f29dca82b9dbef"
|
|
ZHIYUN_APP_SECRET = "sDfhjG7QT1S4gfHiMVYSygmPQPN1R2Ho"
|
|
BASE_URL = "https://open.feishu.cn/open-apis"
|
|
|
|
_token = None
|
|
_token_expires = None
|
|
|
|
|
|
def get_token():
|
|
global _token, _token_expires
|
|
if _token and _token_expires and datetime.now() < _token_expires:
|
|
return _token
|
|
|
|
url = f"{BASE_URL}/auth/v3/tenant_access_token/internal"
|
|
response = requests.post(url, json={
|
|
"app_id": ZHIYUN_APP_ID,
|
|
"app_secret": ZHIYUN_APP_SECRET
|
|
})
|
|
data = response.json()
|
|
if data.get("code") != 0:
|
|
raise Exception(f"获取 token 失败: {data}")
|
|
|
|
_token = data["tenant_access_token"]
|
|
_token_expires = datetime.now() + timedelta(seconds=data.get("expire", 7200) - 60)
|
|
return _token
|
|
|
|
|
|
def headers():
|
|
return {
|
|
"Authorization": f"Bearer {get_token()}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
|
|
def set_document_permission(document_id: str, editable: bool = True):
|
|
"""设置文档权限"""
|
|
url = f"{BASE_URL}/drive/v1/permissions/{document_id}/public"
|
|
payload = {
|
|
"external_access_entity": "open",
|
|
"security_entity": "anyone_can_view",
|
|
"comment_entity": "anyone_can_view",
|
|
"share_entity": "anyone",
|
|
"link_share_entity": "tenant_editable" if editable else "tenant_readable",
|
|
"invite_external": False
|
|
}
|
|
response = requests.patch(url, headers=headers(), params={"type": "docx"}, json=payload)
|
|
return response.json().get("code") == 0
|
|
|
|
|
|
def create_document(title: str, editable: bool = True):
|
|
"""创建文档（自动设置为组织内可编辑）"""
|
|
url = f"{BASE_URL}/docx/v1/documents"
|
|
response = requests.post(url, headers=headers(), json={"title": title})
|
|
data = response.json()
|
|
if data.get("code") != 0:
|
|
raise Exception(f"创建文档失败: {data}")
|
|
doc = data["data"]["document"]
|
|
document_id = doc["document_id"]
|
|
print(f"[OK] 文档创建成功: {document_id}")
|
|
|
|
# 自动设置权限
|
|
if editable and set_document_permission(document_id, True):
|
|
print(f"[OK] 权限设置成功: 组织内可编辑")
|
|
|
|
return document_id
|
|
|
|
|
|
def get_raw_content(document_id: str):
|
|
"""获取文档原始内容"""
|
|
url = f"{BASE_URL}/docx/v1/documents/{document_id}/raw_content"
|
|
response = requests.get(url, headers=headers())
|
|
data = response.json()
|
|
print(f"[DEBUG] raw_content 响应: {data}")
|
|
return data
|
|
|
|
|
|
def get_blocks(document_id: str):
|
|
"""获取文档块"""
|
|
url = f"{BASE_URL}/docx/v1/documents/{document_id}/blocks"
|
|
|
|
# 尝试添加 document_revision_id 参数
|
|
params = {"document_revision_id": -1} # -1 表示最新版本
|
|
|
|
response = requests.get(url, headers=headers(), params=params)
|
|
data = response.json()
|
|
return data
|
|
|
|
|
|
def create_image_block(document_id: str):
|
|
"""创建图片块"""
|
|
url = f"{BASE_URL}/docx/v1/documents/{document_id}/blocks/{document_id}/children"
|
|
|
|
# 添加 document_revision_id 参数
|
|
params = {"document_revision_id": -1}
|
|
|
|
payload = {
|
|
"children": [{
|
|
"block_type": 27,
|
|
"image": {}
|
|
}]
|
|
}
|
|
|
|
response = requests.post(url, headers=headers(), params=params, json=payload)
|
|
data = response.json()
|
|
print(f"[DEBUG] 创建图片块响应: {data}")
|
|
|
|
if data.get("code") != 0:
|
|
raise Exception(f"创建图片块失败: {data}")
|
|
|
|
block_id = data["data"]["children"][0]["block_id"]
|
|
print(f"[OK] 图片块创建成功: {block_id}")
|
|
return block_id
|
|
|
|
|
|
def upload_image(file_path: str, block_id: str):
|
|
"""上传图片"""
|
|
url = f"{BASE_URL}/drive/v1/medias/upload_all"
|
|
|
|
file_name = os.path.basename(file_path)
|
|
file_size = os.path.getsize(file_path)
|
|
|
|
with open(file_path, 'rb') as f:
|
|
files = {'file': (file_name, f, 'image/png')}
|
|
data = {
|
|
'file_name': file_name,
|
|
'parent_type': 'docx_image',
|
|
'parent_node': block_id,
|
|
'size': str(file_size)
|
|
}
|
|
|
|
response = requests.post(
|
|
url,
|
|
headers={"Authorization": f"Bearer {get_token()}"},
|
|
files=files,
|
|
data=data
|
|
)
|
|
|
|
result = response.json()
|
|
print(f"[DEBUG] 上传响应: {result}")
|
|
|
|
if result.get("code") != 0:
|
|
raise Exception(f"上传失败: {result}")
|
|
|
|
return result["data"]["file_token"]
|
|
|
|
|
|
def bind_image_to_block(document_id: str, block_id: str, file_token: str):
|
|
"""
|
|
绑定图片到图片块 (关键步骤!)
|
|
|
|
使用 replace_image 字段通过 PATCH 请求绑定
|
|
"""
|
|
url = f"{BASE_URL}/docx/v1/documents/{document_id}/blocks/{block_id}"
|
|
|
|
params = {"document_revision_id": -1}
|
|
|
|
# 正确的 payload: 使用 replace_image
|
|
payload = {
|
|
"replace_image": {
|
|
"token": file_token
|
|
}
|
|
}
|
|
|
|
print(f"[INFO] 绑定图片: PATCH {url}")
|
|
print(f"[INFO] payload: {payload}")
|
|
|
|
response = requests.patch(url, headers=headers(), params=params, json=payload)
|
|
result = response.json()
|
|
print(f"[DEBUG] 绑定响应: {result}")
|
|
|
|
if result.get("code") == 0:
|
|
print(f"[OK] 图片绑定成功!")
|
|
return True
|
|
else:
|
|
print(f"[ERROR] 图片绑定失败: code={result.get('code')}, msg={result.get('msg')}")
|
|
return False
|
|
|
|
|
|
def main():
|
|
print("\n" + "#" * 60)
|
|
print("# 深入调试飞书云文档图片上传")
|
|
print("#" * 60)
|
|
|
|
# 生成测试图片
|
|
from PIL import Image, ImageDraw
|
|
test_image = "/tmp/debug_test_image.png"
|
|
img = Image.new('RGB', (200, 150), color='#4a90d9')
|
|
draw = ImageDraw.Draw(img)
|
|
draw.text((100, 75), "Test", fill='white', anchor='mm')
|
|
img.save(test_image)
|
|
print(f"[OK] 测试图片: {test_image}")
|
|
|
|
# Step 1: 创建文档
|
|
print("\n--- Step 1: 创建文档 ---")
|
|
doc_id = create_document(f"调试图片上传 - {datetime.now().strftime('%H:%M:%S')}")
|
|
|
|
# Step 2: 创建图片块
|
|
print("\n--- Step 2: 创建图片块 ---")
|
|
block_id = create_image_block(doc_id)
|
|
|
|
# Step 3: 上传图片
|
|
print("\n--- Step 3: 上传图片 ---")
|
|
file_token = upload_image(test_image, block_id)
|
|
print(f"[OK] file_token: {file_token}")
|
|
|
|
# Step 4: 检查块状态
|
|
print("\n--- Step 4: 检查块状态 ---")
|
|
blocks = get_blocks(doc_id)
|
|
for item in blocks.get("data", {}).get("items", []):
|
|
if item.get("block_type") == 27:
|
|
print(f"[INFO] 图片块: {item}")
|
|
|
|
# Step 5: 绑定图片到图片块 (关键步骤!)
|
|
print("\n--- Step 5: 绑定图片到图片块 ---")
|
|
bind_image_to_block(doc_id, block_id, file_token)
|
|
|
|
# Step 6: 再次检查
|
|
print("\n--- Step 6: 再次检查块状态 ---")
|
|
import time
|
|
time.sleep(2) # 等待2秒
|
|
blocks = get_blocks(doc_id)
|
|
for item in blocks.get("data", {}).get("items", []):
|
|
if item.get("block_type") == 27:
|
|
print(f"[INFO] 图片块: {item}")
|
|
|
|
print(f"\n文档地址: https://feishu.cn/docx/{doc_id}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|