refactor: 通用技能按类别拆分为独立目录
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>
This commit is contained in:
239
skills-integration/feishu-plugin/debug_image.py
Normal file
239
skills-integration/feishu-plugin/debug_image.py
Normal file
@@ -0,0 +1,239 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user