131 lines
3.8 KiB
Python
131 lines
3.8 KiB
Python
"""文件对象存储服务客户端 — https://file.wg.zgitm.com
|
||
|
||
认证方式：一次性 Token，每次操作前获取
|
||
系统编码：mokeetext-edit
|
||
"""
|
||
|
||
import requests
|
||
|
||
BASE_URL = "https://file.wg.zgitm.com"
|
||
SYSTEM_CODE = "mokeetext-edit"
|
||
FOLDER_PATH = "/mokeetext-images/" # 图片统一存放目录
|
||
|
||
|
||
def get_token(system_code: str = SYSTEM_CODE) -> str:
|
||
"""获取一次性 Token"""
|
||
resp = requests.post(
|
||
f"{BASE_URL}/api/token",
|
||
json={"systemCode": system_code},
|
||
timeout=15,
|
||
)
|
||
data = resp.json()
|
||
if data.get("code") != 200:
|
||
raise RuntimeError(f"获取文件服务 Token 失败: {data.get('message')}")
|
||
return data["data"]["token"]
|
||
|
||
|
||
def upload_file(
|
||
file_data: bytes,
|
||
filename: str,
|
||
content_type: str,
|
||
upload_user: str = "",
|
||
) -> dict:
|
||
"""
|
||
上传文件到文件服务
|
||
|
||
返回: {"fileId": "...", "fileName": "...", "fileSize": 123}
|
||
"""
|
||
token = get_token()
|
||
|
||
resp = requests.post(
|
||
f"{BASE_URL}/api/upload",
|
||
headers={"Authorization": f"Bearer {token}"},
|
||
files={"file": (filename, file_data, content_type)},
|
||
data={
|
||
"folderPath": FOLDER_PATH,
|
||
"uploadUser": upload_user,
|
||
},
|
||
timeout=60,
|
||
)
|
||
data = resp.json()
|
||
if data.get("code") != 200:
|
||
raise RuntimeError(f"上传文件失败: {data.get('message')}")
|
||
return data["data"]
|
||
|
||
|
||
def download_file(file_id: str) -> tuple[bytes, str, str]:
|
||
"""
|
||
从文件服务下载文件
|
||
|
||
返回: (binary_data, content_type, filename)
|
||
"""
|
||
token = get_token()
|
||
|
||
resp = requests.get(
|
||
f"{BASE_URL}/api/download/{file_id}",
|
||
headers={"Authorization": f"Bearer {token}"},
|
||
timeout=30,
|
||
)
|
||
if resp.status_code == 404:
|
||
raise FileNotFoundError(f"文件不存在: {file_id}")
|
||
if resp.status_code != 200:
|
||
try:
|
||
err = resp.json()
|
||
raise RuntimeError(f"下载文件失败: {err.get('message', resp.text)}")
|
||
except ValueError:
|
||
raise RuntimeError(f"下载文件失败: HTTP {resp.status_code}")
|
||
|
||
content_type = resp.headers.get("Content-Type", "application/octet-stream")
|
||
# 从 Content-Disposition 提取文件名
|
||
filename = file_id
|
||
disposition = resp.headers.get("Content-Disposition", "")
|
||
if "filename*=" in disposition:
|
||
# filename*=UTF-8''原始文件名
|
||
parts = disposition.split("filename*=")
|
||
if len(parts) > 1:
|
||
fname = parts[1].strip()
|
||
if "''" in fname:
|
||
filename = fname.split("''", 1)[1]
|
||
elif 'filename="' in disposition:
|
||
filename = disposition.split('filename="')[1].rstrip('"')
|
||
|
||
return resp.content, content_type, filename
|
||
|
||
|
||
def delete_file(file_id: str) -> bool:
|
||
"""删除文件服务中的文件"""
|
||
token = get_token()
|
||
|
||
resp = requests.get(
|
||
f"{BASE_URL}/api/delete/{file_id}",
|
||
headers={"Authorization": f"Bearer {token}"},
|
||
timeout=15,
|
||
)
|
||
data = resp.json()
|
||
if data.get("code") == 404:
|
||
return False # 文件已不存在，也算删除成功
|
||
if data.get("code") != 200:
|
||
raise RuntimeError(f"删除文件失败: {data.get('message')}")
|
||
return True
|
||
|
||
|
||
def ensure_folder() -> bool:
|
||
"""确保图片存放目录存在"""
|
||
token = get_token()
|
||
|
||
resp = requests.post(
|
||
f"{BASE_URL}/api/folder/create",
|
||
headers={"Authorization": f"Bearer {token}"},
|
||
json={
|
||
"folderPath": FOLDER_PATH,
|
||
"createdBy": SYSTEM_CODE,
|
||
},
|
||
timeout=15,
|
||
)
|
||
data = resp.json()
|
||
if data.get("code") == 400 and "已存在" in data.get("message", ""):
|
||
return True # 文件夹已存在，没问题
|
||
if data.get("code") != 200:
|
||
raise RuntimeError(f"创建文件夹失败: {data.get('message')}")
|
||
return True
|