64 lines
2.0 KiB
Plaintext
64 lines
2.0 KiB
Plaintext
# office_save.dspy - univer-office 编辑后保存：IDocumentData -> docx 写回工作空间
|
||
import os
|
||
import json as _json
|
||
import aiohttp
|
||
|
||
file_id = (params_kw or {}).get('file', '').strip()
|
||
docdata_str = (params_kw or {}).get('documentData', '') or ''
|
||
|
||
uid = await get_user()
|
||
if not uid:
|
||
uid = 'user-01'
|
||
|
||
dbname = get_module_dbname('pipeline-sdlc')
|
||
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
ws_dir, _ = await get_workspace_dir(sor, uid)
|
||
|
||
if not file_id or not ws_dir:
|
||
return {"success": False, "error": "未指定文件或无工作空间"}
|
||
|
||
if not docdata_str:
|
||
return {"success": False, "error": "缺少文档数据"}
|
||
|
||
try:
|
||
documentData = _json.loads(docdata_str)
|
||
except Exception as e:
|
||
return {"success": False, "error": f"文档数据解析失败: {e}"}
|
||
|
||
full_path = ws_dir + '/' + file_id
|
||
|
||
# 路径穿越校验
|
||
real_ws = os.path.realpath(ws_dir)
|
||
real_full = os.path.realpath(full_path)
|
||
if not real_full.startswith(real_ws + os.sep):
|
||
return {"success": False, "error": "非法路径"}
|
||
|
||
# 只允许覆盖已存在的 office 文件（防任意写入）
|
||
if not os.path.isfile(full_path):
|
||
return {"success": False, "error": "目标文件不存在"}
|
||
|
||
title = os.path.basename(full_path)
|
||
|
||
# 调 univer-office 生成 docx（IDocumentData -> docx，保留格式）
|
||
try:
|
||
timeout = aiohttp.ClientTimeout(total=120)
|
||
async with aiohttp.ClientSession(timeout=timeout) as session:
|
||
async with session.post(
|
||
'http://127.0.0.1:19091/convert/doc-data-to-docx',
|
||
json={'documentData': documentData, 'title': title},
|
||
) as resp:
|
||
if resp.status != 200:
|
||
return {"success": False, "error": f"导出失败 HTTP {resp.status}"}
|
||
docx_bytes = await resp.read()
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
# 写回工作空间
|
||
try:
|
||
with open(full_path, 'wb') as f:
|
||
f.write(docx_bytes)
|
||
return {"success": True, "bytes": len(docx_bytes)}
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|