完善模板测试弹窗与多文件解析链路

backend/services/file_summary.py

@@ -0,0 +1,88 @@
+import csv
+import io
+import json
+from pathlib import Path
+
+import pandas as pd
+from docx import Document
+
+from services.minio_client import download_object_bytes, split_bucket_path
+
+try:
+    from pypdf import PdfReader
+except Exception:  # pragma: no cover
+    PdfReader = None
+
+
+def _decode_text(content: bytes) -> str:
+    for encoding in ("utf-8", "utf-8-sig", "gbk", "gb18030"):
+        try:
+            return content.decode(encoding)
+        except Exception:
+            continue
+    return content.decode("utf-8", errors="ignore")
+
+
+def _summarize_docx(content: bytes) -> str:
+    doc = Document(io.BytesIO(content))
+    texts = [paragraph.text.strip() for paragraph in doc.paragraphs if paragraph.text.strip()]
+    return "\n".join(texts[:40])[:4000]
+
+
+def _summarize_csv(content: bytes) -> str:
+    text = _decode_text(content)
+    reader = csv.reader(io.StringIO(text))
+    rows = list(reader[:20])
+    return "\n".join([" | ".join(row) for row in rows])[:4000]
+
+
+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")
+    parts: list[str] = []
+    for sheet_name, dataframe in list(sheet_map.items())[:5]:
+        preview = dataframe.head(10).fillna("").astype(str)
+        parts.append(f"[工作表] {sheet_name}")
+        parts.append(preview.to_csv(index=False).strip())
+    return "\n".join(parts)[:5000]
+
+
+def _summarize_pdf(content: bytes) -> str:
+    if PdfReader is None:
+        return "当前环境未安装 PDF 文本解析依赖，无法提取 PDF 正文。"
+    reader = PdfReader(io.BytesIO(content))
+    texts: list[str] = []
+    for page in reader.pages[:10]:
+        texts.append((page.extract_text() or "").strip())
+    return "\n".join(filter(None, texts))[:4000]
+
+
+def summarize_file_bytes(file_name: str, content: bytes) -> str:
+    suffix = Path(file_name).suffix.lower()
+    if suffix in {".txt", ".md", ".json"}:
+        return _decode_text(content)[:4000]
+    if suffix == ".csv":
+        return _summarize_csv(content)
+    if suffix in {".xlsx", ".xls"}:
+        return _summarize_excel(content, suffix)
+    if suffix == ".docx":
+        return _summarize_docx(content)
+    if suffix == ".pdf":
+        return _summarize_pdf(content)
+    return f"暂不支持解析该文件内容：{file_name}"
+
+
+def summarize_minio_files(file_paths: list[str]) -> 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
+        summaries.append(
+            {
+                "file_name": file_name,
+                "file_path": file_path,
+                "summary": summarize_file_bytes(file_name, content),
+            }
+        )
+    return summaries