完善附件管理与生成任务跟踪
This commit is contained in:
@@ -1,6 +1,8 @@
|
||||
import csv
|
||||
import io
|
||||
import json
|
||||
import subprocess
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import pandas as pd
|
||||
@@ -38,7 +40,10 @@ def _summarize_csv(content: bytes) -> str:
|
||||
|
||||
def _summarize_excel(content: bytes, suffix: str) -> str:
|
||||
excel_buffer = io.BytesIO(content)
|
||||
sheet_map = pd.read_excel(excel_buffer, sheet_name=None) if suffix == ".xlsx" else pd.read_excel(excel_buffer, sheet_name=None, engine="xlrd")
|
||||
if suffix in {".xlsx", ".xlsm"}:
|
||||
sheet_map = pd.read_excel(excel_buffer, sheet_name=None)
|
||||
else:
|
||||
sheet_map = pd.read_excel(excel_buffer, sheet_name=None, engine="xlrd")
|
||||
parts: list[str] = []
|
||||
for sheet_name, dataframe in list(sheet_map.items())[:5]:
|
||||
preview = dataframe.head(10).fillna("").astype(str)
|
||||
@@ -47,6 +52,21 @@ def _summarize_excel(content: bytes, suffix: str) -> str:
|
||||
return "\n".join(parts)[:5000]
|
||||
|
||||
|
||||
def _summarize_doc(content: bytes) -> str:
|
||||
with tempfile.NamedTemporaryFile(suffix=".doc") as temp_file:
|
||||
temp_file.write(content)
|
||||
temp_file.flush()
|
||||
result = subprocess.run(
|
||||
["textutil", "-convert", "txt", "-stdout", temp_file.name],
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
stderr = result.stderr.decode("utf-8", errors="ignore").strip()
|
||||
return f"旧版 Word 文件解析失败:{stderr or 'textutil 无法提取正文'}"
|
||||
return _decode_text(result.stdout)[:4000]
|
||||
|
||||
|
||||
def _summarize_pdf(content: bytes) -> str:
|
||||
if PdfReader is None:
|
||||
return "当前环境未安装 PDF 文本解析依赖,无法提取 PDF 正文。"
|
||||
@@ -68,18 +88,18 @@ def summarize_file_bytes(file_name: str, content: bytes) -> str:
|
||||
if suffix == ".docx":
|
||||
return _summarize_docx(content)
|
||||
if suffix == ".doc":
|
||||
return "当前暂不支持直接解析 .doc 旧版 Word 文件正文,建议先另存为 .docx 后再上传。"
|
||||
return _summarize_doc(content)
|
||||
if suffix == ".pdf":
|
||||
return _summarize_pdf(content)
|
||||
return f"暂不支持解析该文件内容:{file_name}"
|
||||
|
||||
|
||||
def summarize_minio_files(file_paths: list[str]) -> list[dict]:
|
||||
def summarize_minio_files(file_paths: list[str], file_name_mapping: dict[str, str] | None = None) -> list[dict]:
|
||||
summaries: list[dict] = []
|
||||
for file_path in file_paths:
|
||||
bucket, object_name = split_bucket_path(file_path)
|
||||
content = download_object_bytes(bucket, object_name)
|
||||
file_name = Path(object_name).name
|
||||
file_name = (file_name_mapping or {}).get(file_path) or Path(object_name).name
|
||||
summaries.append(
|
||||
{
|
||||
"file_name": file_name,
|
||||
|
||||
Reference in New Issue
Block a user