zwt13703
111 lines
3.7 KiB
Python
111 lines
3.7 KiB
Python
import csv
|
|
import io
|
|
import json
|
|
import subprocess
|
|
import tempfile
|
|
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)
|
|
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)
|
|
parts.append(f"[工作表] {sheet_name}")
|
|
parts.append(preview.to_csv(index=False).strip())
|
|
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 正文。"
|
|
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", ".xlsm"}:
|
|
return _summarize_excel(content, suffix)
|
|
if suffix == ".docx":
|
|
return _summarize_docx(content)
|
|
if suffix == ".doc":
|
|
return _summarize_doc(content)
|
|
if suffix == ".pdf":
|
|
return _summarize_pdf(content)
|
|
return f"暂不支持解析该文件内容：{file_name}"
|
|
|
|
|
|
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 = (file_name_mapping or {}).get(file_path) or Path(object_name).name
|
|
summaries.append(
|
|
{
|
|
"file_name": file_name,
|
|
"file_path": file_path,
|
|
"summary": summarize_file_bytes(file_name, content),
|
|
}
|
|
)
|
|
return summaries
|