| 1 | # -*- coding: utf-8 -*- |
| 2 | |
| 3 | import os |
| 4 | import sys |
| 5 | |
| 6 | models_dir = os.path.dirname(os.path.abspath(__file__)) |
| 7 | backend_dir = os.path.dirname(models_dir) |
| 8 | if backend_dir not in sys.path: |
| 9 | sys.path.insert(0, backend_dir) |
| 10 | |
| 11 | import logging |
| 12 | from typing import Optional |
| 13 | from docx import Document |
| 14 | |
| 15 | logger = logging.getLogger(__name__) |
| 16 | |
| 17 | class FileReader: |
| 18 | """ |
| 19 | 文档内容提取工具类 |
| 20 | """ |
| 21 | |
| 22 | @staticmethod |
| 23 | def extract_text(file_path: str) -> str: |
| 24 | """ |
| 25 | 根据文件后缀名，从文件中提取文本内容 |
| 26 | 支持 .docx, .txt, .md |
| 27 | :param file_path: 文件路径 |
| 28 | :return: 提取出的文本字符串 |
| 29 | """ |
| 30 | if not os.path.exists(file_path): |
| 31 | logger.error(f"文件不存在: {file_path}") |
| 32 | return "" |
| 33 | |
| 34 | ext = os.path.splitext(file_path)[1].lower() |
| 35 | |
| 36 | if ext in [".docx", ".doc"]: |
| 37 | return FileReader._extract_docx(file_path) |
| 38 | elif ext in [".txt", ".md"]: |
| 39 | return FileReader._extract_plain_text(file_path) |
| 40 | elif ext == ".pdf": |
| 41 | return FileReader._extract_pdf(file_path) |
| 42 | else: |
| 43 | logger.error(f"不支持的文件格式: {ext}") |
| 44 | return "" |
| 45 | |
| 46 | @staticmethod |
| 47 | def _extract_docx(file_path: str) -> str: |
| 48 | try: |
| 49 | doc = Document(file_path) |
| 50 | full_text = [] |
| 51 | |
| 52 | # 1. 提取所有段落内容 |
| 53 | for para in doc.paragraphs: |
| 54 | cleaned_text = para.text.strip() |
| 55 | if cleaned_text: |
| 56 | full_text.append(cleaned_text) |
| 57 | |
| 58 | # 2. 提取表格中的内容 |
| 59 | for table in doc.tables: |
| 60 | for row in table.rows: |
| 61 | for cell in row.cells: |
| 62 | cleaned_cell = cell.text.strip() |
| 63 | if cleaned_cell: |
| 64 | full_text.append(cleaned_cell) |
| 65 | |
| 66 | return "\n".join(full_text) |
| 67 | except Exception as e: |
| 68 | logger.error(f"解析 Word 文档失败: {e}") |
| 69 | return "" |
| 70 | |
| 71 | @staticmethod |
| 72 | def _extract_plain_text(file_path: str) -> str: |
| 73 | try: |
| 74 | # 尝试使用 multiple encodings |
| 75 | for encoding in ['utf-8', 'gbk', 'utf-16']: |
| 76 | try: |
| 77 | with open(file_path, 'r', encoding=encoding) as f: |
| 78 | return f.read() |
| 79 | except UnicodeDecodeError: |
| 80 | continue |
| 81 | return "" |
| 82 | except Exception as e: |
| 83 | logger.error(f"解析文本文件失败: {e}") |
| 84 | return "" |
| 85 | |
| 86 | @staticmethod |
| 87 | def _extract_pdf(file_path: str) -> str: |
| 88 | try: |
| 89 | import PyPDF2 |
| 90 | text = "" |
| 91 | with open(file_path, 'rb') as f: |
| 92 | reader = PyPDF2.PdfReader(f) |
| 93 | for page in reader.pages: |
| 94 | text += page.extract_text() + "\n" |
| 95 | return text.strip() |
| 96 | except ImportError: |
| 97 | logger.error("未安装 PyPDF2，无法解析 PDF 文件。请运行: pip install PyPDF2") |
| 98 | return "[错误: 后端未安装 PDF 解析插件]" |
| 99 | except Exception as e: |
| 100 | logger.error(f"解析 PDF 失败: {e}") |
| 101 | return "" |
| 102 | |
| 103 | @staticmethod |
| 104 | def format_as_prompt(filename: str, content: str) -> str: |
| 105 | """ |
| 106 | 将提取的内容格式化为 LLM Prompt 友好的字符串 |
| 107 | """ |
| 108 | if not content: |
| 109 | return "" |
| 110 | |
| 111 | return ( |
| 112 | f"\n--- [上传的文件内容开始] ---\n" |
| 113 | f"文件名: {filename}\n" |
| 114 | f"内容如下:\n" |
| 115 | f"{content}\n" |
| 116 | f"--- [上传的文件内容结束] ---\n" |
| 117 | ) |
| 118 |