vent_deep_agent/tools/file_reader_tools.py

# -*- coding: utf-8 -*-
"""
文件内容读取工具模块

提供统一的文件内容提取能力，自动识别文件格式，支持：
- PDF（复用 tools/pdf_tools 的 PyPDF2 提取）
- DOCX（Word 文档）
- XLSX（Excel 表格）
- PPTX（PowerPoint 演示文稿）
- TXT / MD / CSV / JSON 等纯文本文件

所有工具函数供 DeepAgent 调用，返回结构化 JSON 文本。
"""

import json
import os
import re

# ── 最大提取字符数 ──
_MAX_CONTENT_CHARS = 15000


async def read_file_content(file_path: str) -> str:
    """读取用户上传的文件内容。当用户上传了附件，你需要查看文件中的文字内容时调用此工具。
    
    支持的文件格式：
    - PDF 文档（.pdf）
    - Word 文档（.docx）
    - Excel 表格（.xlsx）
    - PowerPoint 演示文稿（.pptx）
    - 纯文本（.txt / .md / .csv / .json）
    
    典型调用场景：
    1. 用户上传了一个 Word 报告，需要你阅读和分析内容
    2. 用户上传了 Excel 数据表，需要你提取和理解数据
    3. 用户上传了 PDF 文档，需要你从中获取信息
    
    Args:
        file_path: 文件的完整路径（从用户消息中的「文件临时路径」获取）
    
    Returns:
        JSON 格式，包含 file_path、format（文件格式）、content（提取的文本）、
        content_length（字符数）。文本最多保留 15000 字符，超出部分截断。
    """
    if not file_path or not file_path.strip():
        return json.dumps({"error": "文件路径不能为空"}, ensure_ascii=False)

    fp = file_path.strip()

    if not os.path.exists(fp):
        return json.dumps(
            {"error": f"文件不存在: {fp}", "file_path": fp},
            ensure_ascii=False,
        )

    ext = os.path.splitext(fp)[1].lower()

    try:
        if ext == ".pdf":
            text = _read_pdf(fp)
            fmt = "pdf"
        elif ext == ".docx":
            text = _read_docx(fp)
            fmt = "docx"
        elif ext in (".xlsx", ".xlsm"):
            text = _read_xlsx(fp)
            fmt = "xlsx"
        elif ext == ".pptx":
            text = _read_pptx(fp)
            fmt = "pptx"
        elif ext in (".txt", ".md", ".csv", ".json", ".xml", ".html", ".htm", ".log", ".yaml", ".yml", ".py", ".js", ".ts", ".sql", ".cfg", ".ini"):
            text = _read_text(fp)
            fmt = ext.lstrip(".")
        else:
            return json.dumps(
                {
                    "error": f"不支持的文件格式「{ext}」。支持：pdf / docx / xlsx / pptx / txt / md / csv / json",
                    "file_path": fp,
                    "format": ext,
                },
                ensure_ascii=False,
            )
    except Exception as e:
        return json.dumps(
            {"error": f"文件读取失败: {str(e)}", "file_path": fp, "format": ext},
            ensure_ascii=False,
        )

    # ── 截断 ──
    content_length = len(text)
    if content_length > _MAX_CONTENT_CHARS:
        text = text[:_MAX_CONTENT_CHARS] + f"\n…（内容已截断，原始 {content_length} 字符）"

    return json.dumps(
        {
            "file_path": fp,
            "format": fmt,
            "content": text,
            "content_length": min(content_length, _MAX_CONTENT_CHARS),
        },
        ensure_ascii=False,
    )


# ═══════════════════════════════════════════════════════════════
# 各格式解析器（内部函数，不直接暴露给 Agent）
# ═══════════════════════════════════════════════════════════════


def _read_pdf(file_path: str) -> str:
    """PDF 文本提取（复用 tools/pdf_tools 的 PyPDF2 提取器）。"""
    from tools.pdf_tools import _extract_with_pypdf

    text = _extract_with_pypdf(file_path)
    if text:
        return text.strip()
    return "（PDF 文件无可提取的文字内容，可能是扫描件或图片型 PDF）"


def _read_docx(file_path: str) -> str:
    """Word 文档 (.docx) 文本提取。

    提取内容：
    - 正文段落文本
    - 表格中的文字（标为 [表格] 前缀）
    """
    from docx import Document

    doc = Document(file_path)
    parts = []

    for element in doc.element.body:
        tag = element.tag.split("}")[-1] if "}" in element.tag else element.tag

        if tag == "p":
            # 段落
            ns = {"w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main"}
            texts = element.findall(".//w:t", ns)
            line = "".join(t.text or "" for t in texts).strip()
            if line:
                parts.append(line)

        elif tag == "tbl":
            # 表格
            rows = []
            for row in element.findall(".//{http://schemas.openxmlformats.org/wordprocessingml/2006/main}tr"):
                cells = []
                for cell in row.findall(".//{http://schemas.openxmlformats.org/wordprocessingml/2006/main}tc"):
                    texts = cell.findall(".//{http://schemas.openxmlformats.org/wordprocessingml/2006/main}t")
                    cell_text = " ".join(t.text or "" for t in texts).strip()
                    cells.append(cell_text)
                rows.append(" | ".join(cells))
            if rows:
                parts.append("[表格] " + "\n[表格] ".join(rows))

    result = "\n".join(parts).strip()
    return result if result else "（Word 文档中未找到文字内容）"


def _read_xlsx(file_path: str) -> str:
    """Excel 表格 (.xlsx) 文本提取。

    每个 Sheet 标记为 [Sheet: xxx]，每行用 | 分隔单元格。
    """
    from openpyxl import load_workbook

    wb = load_workbook(file_path, read_only=True, data_only=True)
    parts = []

    for sheet_name in wb.sheetnames:
        ws = wb[sheet_name]
        parts.append(f"[Sheet: {sheet_name}]")

        row_count = 0
        for row in ws.iter_rows(max_row=200, values_only=True):
            row_vals = [str(v) if v is not None else "" for v in row]
            # 跳过完全空行
            if any(c.strip() for c in row_vals):
                parts.append(" | ".join(row_vals))
                row_count += 1

        if row_count == 0:
            parts.append("（空表）")

    wb.close()
    result = "\n".join(parts).strip()
    return result if result else "（Excel 文件中未找到数据）"


def _read_pptx(file_path: str) -> str:
    """PowerPoint 演示文稿 (.pptx) 文本提取。

    每张幻灯片标记为 [幻灯片 N]，提取所有文本框和占位符中的文字。
    """
    from pptx import Presentation

    prs = Presentation(file_path)
    parts = []

    for i, slide in enumerate(prs.slides, 1):
        parts.append(f"[幻灯片 {i}]")
        for shape in slide.shapes:
            if shape.has_text_frame:
                for paragraph in shape.text_frame.paragraphs:
                    line = paragraph.text.strip()
                    if line:
                        parts.append(line)
            if shape.has_table:
                table = shape.table
                for row in table.rows:
                    cells = [cell.text.strip() for cell in row.cells]
                    parts.append(" | ".join(cells))

    result = "\n".join(parts).strip()
    return result if result else "（PPT 中未找到文字内容）"


def _read_text(file_path: str) -> str:
    """纯文本文件读取（UTF-8 / GBK 自动检测）。"""
    for encoding in ("utf-8", "gbk", "gb2312", "latin-1"):
        try:
            with open(file_path, "r", encoding=encoding) as f:
                return f.read().strip()
        except (UnicodeDecodeError, UnicodeError):
            continue
    return "（无法解码文件内容，可能是二进制文件）"