企业级RAG问答系统（二）数据处理模块

目录

一.Word解析模块

二.MD（markdown）解析模块

三.HTML解析模块

四.解析器工厂类

五.创建Web应用

1.解析上传文件API接口

2.解析已存在文件API接口

3.优化代码，改bug

---

一.Word解析模块

uv add python-docx

总体上和pdf解析差不多，个别地方改动

from pathlib import Path
from typing import Dict, List, Optional

import docx  # python-docx

from src.parsers.base import BaseParser, ParseResult
from src.parsers.chunker import TextChunker
from src.parsers.cleaner import TextCleaner
from src.parsers.table_parser import TableProcessor


class WordParser(BaseParser):
    def __init__(self,
                 chunker:Optional[TextChunker] =None,
                 cleaner:Optional[TextCleaner] =None,
                 table_processor:Optional[TableProcessor] =None
                 ):

        # 初始化组件
        self.chunker = chunker if chunker else TextChunker()
        self.cleaner = cleaner if cleaner else TextCleaner()
        self.table_processor = table_processor if table_processor else TableProcessor()
    
    def parse(self, file_path:str) -> ParseResult:
        '''
        通过文件路径解析文件
        参数：
            file_path: 文件路径
        '''
        # 创建Docx对象
        doc  = docx.Document(file_path)

        # 提取数据
        paragraphs = [p.text for p in doc.paragraphs if p.text.strip()]
        raw_text = '\n\n'.join(paragraphs)

        # 提取表格
        tables = self._extract_tables(doc)

        # 处理数据
        cleaned_text = self.cleaner.clean(raw_text)
        merged_tables = self.table_processor.merge_tables(tables)
        chunks = self.chunker.split(cleaned_text, merged_tables)

        return ParseResult(
            filename=Path(file_path).name,
            total_pages=1,  # 不分页
            total_chunks=len(chunks),
            table_count=len(merged_tables),
            chunks=chunks,
            raw_text=raw_text,
            tables=tables,
            metadata={'parser':'python-docx','filename':Path(file_path).name}
        )

    def parse_from_bytes(self, file_bytes, filename):
        '''
        通过字节流解析文件
        参数：
            file_bytes: 文件字节流
            filename: 文件名
        '''
        from io import BytesIO
        stream = BytesIO(file_bytes)
        doc = docx.Document(stream)

        # 提取数据
        paragraphs = [p.text for p in doc.paragraphs if p.text.strip()]
        raw_text = '\n\n'.join(paragraphs)

        # 提取表格
        tables = self._extract_tables(doc)

        # 处理数据
        cleaned_text = self.cleaner.clean(raw_text)
        merged_tables = self.table_processor.merge_tables(tables)
        chunks = self.chunker.split(cleaned_text, merged_tables)

        return ParseResult(
            filename=filename,
            total_pages=1,  # 不分页
            total_chunks=len(chunks),
            table_count=len(merged_tables),
            chunks=chunks,
            raw_text=raw_text,
            tables=tables,
            metadata={'parser':'python-docx','filename':filename}
        )
    
    def supported_extensions(self) -> list[str]:
        '''
        支持的文件扩展名
        '''
        return ['.docx','.DOCX']
    
    def _extract_tables(self,doc) -> List[Dict]:
        '''
        提取表格
        '''
        tables = []
        for table_index,table in  enumerate(doc.tables):
            # 转换markdown
            md_rows = []
            for row_index,row in enumerate(table.rows):
                # 获取列
                cells = [cell.text.strip() for cell in row.cells]
                # | col1 | col2 | col3 |
                md_rows.append('| ' + ' | '.join(cells) + ' |')
                # 增加表头
                if row_index == 0:
                    md_rows.append( '| ' + ' | '.join(['---']*len(cells)) + ' |')
            tables.append({
                'page':1,
                'rows':len(table.rows),
                'cols':len(table.columns),
                'markdown':'\n'.join(md_rows)
            })
        return tables

在代码中，会发现

为什么 Word 文档的总页数设置为 1？

这是因为 Word 文档（.docx）与 PDF 有本质区别：

核心原因

1. Word 是流式文档，没有固定页面 📄

Word 文档是流式布局（Flow Layout）：

• 内容像水流一样连续排列
• 没有固定的"页"概念
• 同一个文档在不同设备/设置下，分页位置可能完全不同

举例：

• 在你的电脑上：某段文字在第 2 页
• 在别人的电脑上（字体不同、屏幕分辨率不同）：可能在第 3 页
• 打印成 A4 纸：可能在第 2 页
• 打印成 A3 纸：可能在第 1 页

2. PDF 是固定版式文档 📋

PDF 是固定布局（Fixed Layout）：

• 创建时就确定了每一页的内容
• 无论在什么设备上打开，第 5 页的内容永远是第 5 页
• 可以精确计算总页数

二.MD（markdown）解析模块

直接用python原生的open函数读出来就是字符串格式。

from pathlib import Path
from typing import Optional

from src.parsers.base import BaseParser, ParseResult
from src.parsers.chunker import ChunkStrategy, TextChunker
from src.parsers.cleaner import TextCleaner


class MarkdownParser(BaseParser):
    '''
    Markdown 解析器
    '''
    def __init__(
            self,
            chunker: Optional[TextChunker] = None,
            cleaner: Optional[TextCleaner] = None
    ):
        self.cleaner = cleaner if cleaner else TextCleaner()
        self.chunker = chunker if chunker else TextChunker(
            strategy=ChunkStrategy.PARAGRAPH,
            max_chunk_size=1000,
            chunk_overlap=200
        )
    
    def parse(self, file_path:str) ->ParseResult:
        '''
        解析Markdown文件
        '''
        # 直接读取即可
        with open(file_path,'r',encoding='utf-8') as f:
            raw_text = f.read()

        cleaned_text = self.cleaner.clean(raw_text)
        chunks = self.chunker.split(cleaned_text)

        return ParseResult(
            filename=Path(file_path).name,
            total_pages=1,
            total_chunks=len(chunks),
            table_count=0,
            chunks=chunks,
            raw_text=raw_text,
            tables=[],
            metadata={'parser':'markdown','filename':Path(file_path).name}
        )
    
    def parse_from_bytes(self, file_bytes:bytes, filename:str):

        raw_text = file_bytes.decode('utf-8')

        cleaned_text = self.cleaner.clean(raw_text)
        chunks = self.chunker.split(cleaned_text)

        return ParseResult(
            filename=filename,
            total_pages=1,
            total_chunks=len(chunks),
            table_count=0,
            chunks=chunks,
            raw_text=raw_text,
            tables=[],
            metadata={'parser':'markdown','filename':filename}
        )
    
    
    def supported_extensions(self) -> list[str]:
        return ['.md','.MD','.markdown','.MARKDOWN']

三.HTML解析模块

安装一个可以将html解析为markdown的库

uv add markdownify

supported_extensions 存在的意义是让未来的调度模块可以验证程序员设置的可解析的文件格式限制

from pathlib import Path
from typing import Optional

from markdownify import markdownify

from src.parsers.base import BaseParser, ParseResult
from src.parsers.chunker import ChunkStrategy, TextChunker
from src.parsers.cleaner import TextCleaner


class HTMLParser(BaseParser):
    '''
    HTML 解析器
    '''
    def __init__(
            self,
            chunker: Optional[TextChunker] = None,
            cleaner: Optional[TextCleaner] = None
    ):
        self.cleaner = cleaner if cleaner else TextCleaner()
        self.chunker = chunker if chunker else TextChunker(
            strategy=ChunkStrategy.PARAGRAPH,
            max_chunk_size=1000,
            chunk_overlap=200
        )
    
    def parse(self, file_path:str) ->ParseResult:
        '''
        解析HTML文件
        '''
        with open(file_path,'r',encoding='utf-8') as f:
            raw_text = f.read()

        # 转换为markdown
        raw_text = markdownify(raw_text)

        cleaned_text = self.cleaner.clean(raw_text)
        chunks = self.chunker.split(cleaned_text)

        return ParseResult(
            filename=Path(file_path).name,
            total_pages=1,
            total_chunks=len(chunks),
            table_count=0,
            chunks=chunks,
            raw_text=raw_text,
            tables=[],
            metadata={'parser':'markdownify','filename':Path(file_path).name}
        )
    
    def parse_from_bytes(self, file_bytes:bytes, filename:str):

        raw_text = file_bytes.decode('utf-8')

        # 转换为markdown
        raw_text = markdownify(raw_text)

        cleaned_text = self.cleaner.clean(raw_text)
        chunks = self.chunker.split(cleaned_text)

        return ParseResult(
            filename=filename,
            total_pages=1,
            total_chunks=len(chunks),
            table_count=0,
            chunks=chunks,
            raw_text=raw_text,
            tables=[],
            metadata={'parser':'markdown','filename':filename}
        )
    
    # supported_extensions 存在的意义是让未来的调度模块可以验证程序员设置的可解析的文件格式限制
    def supported_extensions(self) -> list[str]:
        return ['.html','.htm','.HTML','.HTM']

四.解析器工厂类

用户在上传文件之后肯定不想再自主选择文件的格式，由系统根据文件的后缀自主选择解析方式更为友好，所以本节创建一个解析器工厂类来整合之前写的那些解析器。

方案一：

继续在psrser文件夹下创建一个工厂类来整合其他解析类

方案二：

将工厂类写到parser包的__init__.py文件中，这样的好处（其他好处可以豆包搜索一下）是当导入parser包时，会自动执行__init__.py文件的代码

采取方案二

构建一个工厂类，使用通用的创建解析器的代码，改变的只是不同的解析器类名，就可以得到不同的解析器。除了构建创造不同解析器实例的工厂类以外，还编写了一个独立于工厂类之外的文件解析接口，传入文件完整路径即可根据文件后缀自动创建对应的解析器来进行解析。

———————————————————————————————————————————

cls vs self 核心区别

先看这个函数：def get_parser(cls, file_path:str) -> BaseParser: 第一个参数是 cls，说明这是类方法（@classmethod）。

1. self

• 属于实例方法，第一个参数约定名为 self
• self 代表类的实例对象（new 出来的对象）
• 只能实例调用：obj.xxx()
• 作用：访问实例属性、实例方法

def func(self):
    print(self.name) # 读取实例上的成员

2. cls

• 属于类方法，必须加装饰器 @classmethod，第一个参数约定名为 cls
• cls 代表类本身（不是实例！）
• 可以类直接调用，也可以实例调用：BaseParser.get_parser() / obj.get_parser()
• 作用：访问类属性、静态资源、工厂方法（像你截图里这个，根据文件后缀创建解析器实例，典型工厂方法）

@classmethod
def get_parser(cls, file_path:str):
    ext = Path(file_path).suffix.lower()
    # cls 就是 BaseParser 这个类，可以用 cls 去创建子类实例
    if ext == ".pdf":
        return cls.PdfParser(file_path)

参数	方法类型	指代对象	使用场景
self	实例方法	实例对象	操作实例的数据
cls	类方法@classmethod	类本身	工厂方法、类级别的逻辑

---

__init__.py

这时候也能体现出通过base抽象类进行约束的好处，若干解析器的解析函数名都一样，统一调用一个函数即可。

from pathlib import Path

from .base import BaseParser, ParseResult
from .html_parser import HTMLParser
from .markdown_parser import MarkdownParser
from .pdf_parser import PDFParser
from .word_parser import WordParser


class ParserFactory:
    '''
    解析器工厂类
    '''
    _parsers = {
        '.pdf':PDFParser,
        '.html':HTMLParser,
        '.md':MarkdownParser,
        '.docx':WordParser
    }

    @classmethod
    def get_parser(cls, file_path:str) -> BaseParser:
        '''
        根据文件路径获取解析器
        Args:
            file_path:文件路径
            Returns:
                BaseParser:解析器实例
                '''
        # 从文件路径中提取文件扩展名（例如：.pdf、.html、.md、.docx等）并转为小写
        file_ext = Path(file_path).suffix.lower()
        if file_ext not in cls._parsers:
            raise ValueError(f'不支持的文件格式: {file_ext}')
        # 获取对应的解析器
        return cls._parsers[file_ext]()

    @classmethod
    def register_parser(cls, file_ext:str, parser:BaseParser) -> None:
        '''
        注册解析器
        Args:
            file_ext:文件扩展名
            parser:解析器实例
            '''
        cls._parsers[file_ext] = parser

    @classmethod
    def supported_extensions(cls) -> list[str]:
        '''
        获取支持的扩展名
        Returns:
            支持的扩展名列表
            '''
        return list(cls._parsers.keys())

# 调用函数(暴露给用户的简单api接口)
def parse_file(file_path:str) -> ParseResult:
    '''
    自动识别文件类型并解析
    Args:
        file_path (str): 文件路径
        Returns:
            ParseResult: 解析结果
            '''
    # 通过工厂类获取解析器，将便捷 API 与工厂本身分离，保持了类的职责单一性
    parser = ParserFactory.get_parser(file_path)
    # 解析文件
    return parser.parse(file_path)

测试：

"""测试 ParserFactory 解析器工厂"""
import os
import sys

sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from src.parsers import ParserFactory, parse_file


def test_supported_extensions():
    """测试支持的文件类型"""
    print("\n--- 测试支持的文件类型 ---")

    extensions = ParserFactory.supported_extensions()
    print(f"支持的文件扩展名: {extensions}")

    expected = [".pdf", ".docx", ".md", ".html"]
    for ext in expected:
        assert ext in extensions, f"缺少支持的扩展名: {ext}"

    print("[OK] 支持的文件类型正确")


def test_get_parser():
    """测试获取解析器"""
    print("\n--- 测试获取解析器 ---")

    test_cases = [
        ("test.pdf", "PDFParser"),
        ("test.docx", "WordParser"),
        ("test.md", "MarkdownParser"),
        ("test.html", "HTMLParser"),
    ]

    for file_path, expected_parser in test_cases:
        parser = ParserFactory.get_parser(file_path)
        parser_type = type(parser).__name__
        print(f"  {file_path} -> {parser_type}")
        assert parser_type == expected_parser, f"应为 {expected_parser}，实际为 {parser_type}"

    print("[OK] 获取解析器正确")


def test_unsupported_file():
    """测试不支持的文件类型"""
    print("\n--- 测试不支持的文件类型 ---")

    try:
        ParserFactory.get_parser("test.xyz")
        assert False, "应该抛出 ValueError"
    except ValueError as e:
        print(f"  捕获预期错误: {e}")
        # 因为get_parser里 throw ValueError 所以这里需要捕获，而且错误信息包含“不支持的文件格式”
        assert "不支持的文件格式" in str(e)

    print("[OK] 不支持文件类型处理正确")


def test_case_insensitive():
    """测试扩展名大小写不敏感"""
    print("\n--- 测试扩展名大小写不敏感 ---")

    test_cases = ["test.PDF", "test.Pdf", "test.DOCX", "test.Md"]

    for file_path in test_cases:
        parser = ParserFactory.get_parser(file_path)
        print(f"  {file_path} -> {type(parser).__name__}")

    print("[OK] 大小写不敏感处理正确")


def test_register_parser():
    """测试注册自定义解析器"""
    print("\n--- 测试注册自定义解析器 ---")

    from src.parsers.base import BaseParser

    class CustomParser(BaseParser):
        def supported_extensions(self):
            return [".custom"]

        def parse(self, file_path):
            pass

        def parse_from_bytes(self, file_bytes, filename):
            pass

    # 注册前
    assert ".custom" not in ParserFactory.supported_extensions()

    # 注册
    ParserFactory.register_parser(".custom", CustomParser)
    print(f"  注册后支持的类型: {ParserFactory.supported_extensions()}")

    # 验证
    assert ".custom" in ParserFactory.supported_extensions()
    parser = ParserFactory.get_parser("test.custom")
    assert isinstance(parser, CustomParser)
    print("  自定义解析器注册成功")

    # 清理
    del ParserFactory._parsers[".custom"]

    print("[OK] 自定义解析器注册正确")


def test_parse_pdf():
    """测试解析 PDF 文件"""
    print("\n--- 测试解析 PDF 文件 ---")

    file_path = "data/test.pdf"
    if not os.path.exists(file_path):
        print(f"  [SKIP] 文件不存在: {file_path}")
        return

    result = parse_file(file_path)
    print(f"  文件: {result.filename}")
    print(f"  总页数: {result.total_pages}")
    print(f"  总块数: {result.total_chunks}")
    print(f"  表格数: {result.table_count}")
    print(f"  文本长度: {len(result.raw_text)}")

    assert result.filename == "test.pdf"
    assert len(result.raw_text) > 0

    print("[OK] PDF 解析正确")


def test_parse_markdown():
    """测试解析 Markdown 文件"""
    print("\n--- 测试解析 Markdown 文件 ---")

    file_path = "data/test.md"
    if not os.path.exists(file_path):
        print(f"  [SKIP] 文件不存在: {file_path}")
        return

    result = parse_file(file_path)
    print(f"  文件: {result.filename}")
    print(f"  文本长度: {len(result.raw_text)}")
    print(f"  内容预览: {result.raw_text[:100]}...")

    assert result.filename == "test.md"
    assert len(result.raw_text) > 0

    print("[OK] Markdown 解析正确")


def test_parse_html():
    """测试解析 HTML 文件"""
    print("\n--- 测试解析 HTML 文件 ---")

    file_path = "data/test.html"
    if not os.path.exists(file_path):
        print(f"  [SKIP] 文件不存在: {file_path}")
        return

    result = parse_file(file_path)
    print(f"  文件: {result.filename}")
    print(f"  文本长度: {len(result.raw_text)}")
    print(f"  内容预览: {result.raw_text[:100]}...")

    assert result.filename == "test.html"
    assert len(result.raw_text) > 0

    print("[OK] HTML 解析正确")


def test_parse_word():
    """测试解析 Word 文件"""
    print("\n--- 测试解析 Word 文件 ---")

    file_path = "data/test.docx"
    if not os.path.exists(file_path):
        print(f"  [SKIP] 文件不存在: {file_path}")
        return

    result = parse_file(file_path)
    print(f"  文件: {result.filename}")
    print(f"  文本长度: {len(result.raw_text)}")
    print(f"  内容预览: {result.raw_text[:100]}...")

    assert result.filename == "test.docx"
    assert len(result.raw_text) > 0

    print("[OK] Word 解析正确")


def run_all_tests():
    """运行所有测试"""
    print("=" * 50)
    print("ParserFactory 测试")
    print("=" * 50)

    # 工厂类测试
    test_supported_extensions()
    test_get_parser()
    test_unsupported_file()
    test_case_insensitive()
    test_register_parser()

    # 实际文件解析测试
    test_parse_pdf()
    test_parse_markdown()
    test_parse_html()
    test_parse_word()

    print("\n" + "=" * 50)
    print("[PASS] 测试完成")
    print("=" * 50)


if __name__ == "__main__":
    run_all_tests()

五.创建Web应用

三个库：

1.解析上传文件API接口

config.py:

在配置类里配置 可解析的文件格式和可解析的文件最大体积

class Settings:
    '''
    API配置类
    '''
    #文件上传配置
    # 计算机存储、文件大小、FastAPI 文件上传限制，**全部默认按字节 Byte 计算**  
    MAX_FILE_SIZE = 1024 * 1024 * 50 #最大文件大小50MB
    ALLOWED_EXTENSIONS = ['.pdf','.md','.html','.docx']

# 全局配置对象
settings = Settings()

然后编写document.py:

编写将文件解析的路由节点功能代码，注意上传文件是通过IO流来上传的，所以解析函数要用通过字节刘解析的那个解析函数

from pathlib import Path

from fastapi import APIRouter, File, HTTPException, UploadFile

from src.api.config import settings
from src.parsers import ParserFactory

router = APIRouter(prefix="/documents",tags=["文件解析"])

# 上传文件进行解析的路由
@router.post("/upload")
async def upload_document(file: UploadFile = File(...)): # 参数：参数类型 = 默认值
    '''
    上传并解析文档
    支持的格式：pdf,html,md,docx
    '''
    # # str
    # print(type(file.filename))
    
    # 获取转小写的文件格式
    ext = Path(file.filename).suffix.lower()
    file_bytes = await file.read()

    #判断文件的大小
    if len(file_bytes) > settings.MAX_FILE_SIZE:
        size_mb = settings.MAX_FILE_SIZE / 1024 / 1024
        raise HTTPException(status_code=400, detail=f"文件大小不能超过{size_mb:.2f}MB")
    # 判断文件格式
    if ext not in settings.ALLOWED_EXTENSIONS:
        raise HTTPException(status_code=400, detail=f"不支持的文件格式: {ext}")

    # 文件格式和大小都符合要求，调用解析器进行解析
    try:
        # 获得工厂类实例
        parser = ParserFactory.get_parser(file.filename)
        # 解析文件
        result = parser.parse_from_bytes(file_bytes=file_bytes, filename=file.filename)

        # 返回解析结果
        return {
            'message': '文件解析成功',
            'file_name': result.filename,
            'total_pages': result.total_pages,
            'chunks_count': result.total_chunks,
            'table_count': result.table_count,
            'chunks': result.chunks
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"文件解析失败: {str(e)}")

编写main.py文件:

将document的路由注册到Fast API的app当中

import os
import sys

sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))

from fastapi import FastAPI

from src.api.routers import document

app = FastAPI(
    title = "FastAPI RAG系统",
    description = "基于FastAPI的RAG系统",
    version = "0.1.0"
)

app.include_router(document.router)

@app.get("/")
async def index():
    return "Hello World"

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

启动fastapi，上传文件进行测试。

2.解析已存在文件API接口

先区分两个完全不一样的接口，别混在一起：

1. ✅ 之前的 /uoload：前端上传文件（UploadFile），文件从客户端传到服务端
2. ✅ 现在这个 /parse-path：文件早就躺在服务器磁盘上了，客户端只发一个字符串：本地路径 D:/xxx/a.pdf，告诉后端：你去本机这个路径读文件解析

面对文件已经存在于服务器的情况，是不需要再上传的，所以本节来写一下已存在文件的解析API接口。

接着在document文件下面追加路由即可。

# 请求体
class ParseRequest(BaseModel):
    file_path: str

# 解析已经存在的文件
@router.post('parse-path')
async def parse_file_by_path(body: ParseRequest):
    '''按服务器本地路径解析文档（用于已有文件或者批处理）'''
    path = body.file_path
    # 判断文件是否存在
    # `os.path.isfile(path)`：**判断给定路径，是不是一个真实存在的文件（不是文件夹）**
    if not path or not os.path.isfile(path):
        raise HTTPException(status_code=400, detail=f"文件不存在:{path}")
    try:
        # parser/__init__.py里的函数
        result = parse_file(path)
        # 返回解析结果
        return {
            'message': '文件解析成功',
            'file_name': result.filename,
            'total_pages': result.total_pages,
            'chunks_count': result.total_chunks,
            'table_count': result.table_count,
            'chunks': result.chunks
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"解析文件失败:{str(e)}")

还可以额外添加点别的功能：

@router.get('supported_formats')
async def get_supported_formats():
    '''获取支持的文件格式'''
    return {
        "formats": ParserFactory.supported_extensions()
    }

自启动修改（main.py）：

if __name__ == "__main__":
    import uvicorn
    # 没有代码修改之后自动重启的写法
    # uvicorn.run(app, host="127.0.0.1", port=8000)
    # 代码修改之后自动重启的写法
    uvicorn.run('src.api.main:app', host="127.0.0.1", port=8000,reload=True)

3.优化代码，改bug

(1)字节流解析bug

docling已经没有convert_from_bytes()这个函数了，所以还得用convert，但是convert处理的是文件,所以字节流解析时要改成如下的写法：

# 1. bytes 包装成 BytesIO
buf = BytesIO(file_bytes)
# 2. 构造 DocumentStream，必须带上文件名（docling靠后缀判断文档类型！非常关键）
source = DocumentStream(name=filename, stream=buf)
# 使用 DocumentConverter 从字节流解析 PDF 文件
doc = self.converter.convert(source)

（2）已经存在的文件（服务器本身已经存在/批处理）解析接口路径解析bug

这是fastapi的测试文件，此时app是在本地运行的，所以此时本机电脑充当服务器，客户端要提交的服务器端的文件路径即本机电脑上的文件路径，下图为正确的文件路径（额外手动设置了双斜线，如果是单斜线的话会报json decode error）

为什么会这样？

当你访问 FastAPI 自动生成的 Swagger 文档页面（/docs），在 /documents/parse-path 接口中输入路径时，请求体是以 JSON 格式 发送的。

JSON 规范中，反斜杠 \ 是一个转义字符。

JSON 解析器看到 \A、\c、\E、\d 这些序列时，它们 不是合法的 JSON 转义序列，因此直接报 JSON decode error，请求还没到你的代码就已经失败了。

而当你输入双反斜杠 D:\\Agent Development\\... 时，\\ 是合法的 JSON 转义，被正确解析为一个反斜杠 \，所以能正常工作。

如何解决？

方案一（推荐）：前端设置校验与修正

例如：

Windows 系统本身也支持正斜杠作为路径分隔符，所以在 Swagger 页面中输入：

D:/Agent Development/code/Enterprise_RAG_QA_system/data/test_document.docx

这是最简单、不需要改任何代码的方案。

方案二：后端改用 Query 参数或 Form 表单接收路径

如果希望用户直接粘贴 Windows 路径而不需要手动转义，可以把 parse_file_by_path 接口改为 GET 请求 + Query 参数，或者 POST + Form 表单。这样路径就不会经过 JSON 解析器了。例如：

# 改为 GET + Query 参数
@router.get('/parse-path')
async def parse_file_by_path(file_path: str):
    ...

这样用户在浏览器地址栏或 Swagger 的 query 参数框中输入 D:\Agent Development\... 就不会有 JSON 转义问题了。

总结：这不是代码逻辑的问题，而是 JSON 协议本身的限制。

在企业级开发中，这个问题通常通过以下几个层次的方案来解决：

方案一：改为 GET 请求（最简单、最推荐）

如上

方案二：前端预处理（最常见）

用户在输入框粘贴路径后，前端 JS 在发送请求前自动做转义处理。

*****************************************************************************

本文给出另一种办法，虽然仍然存在漏洞，但是作为只是编写后端代码时的一些思路还是记录一下，本文的思路就是既然输入常规的路径会被json解析错误，所以索性把代码写成只需要输入文件名（包括后缀）即可，不需要文件名之前完整的文件路径，例如：test_document.docx

具体做法：

在data文件夹下建一个upload文件夹，默认以后客户端上传到服务器的文件都会放到这个文件夹下，上传到这里的文件自然就成为了服务器已有的文件，此时对服务器已有文件解析时就有固定的路径，即data/upload + /你的文件名.后缀名

代码如下：

fastapi的配置文件config.py:

添加了一个上传文件目录

class Settings:
    '''
    API配置类
    '''
    #文件上传配置
    # 计算机存储、文件大小、FastAPI 文件上传限制，**全部默认按字节 Byte 计算**  
    MAX_FILE_SIZE = 1024 * 1024 * 50 #最大文件大小50MB
    ALLOWED_EXTENSIONS = ['.pdf','.md','.html','.docx']

    #配置上传目录
    UPLOAD_DIR = 'data/uploads'

# 全局配置对象
settings = Settings()

路由文件document.py：

def ensure_upload_dir_exists():
    '''确保上传目录存在'''
    upload_dir = Path(settings.UPLOAD_DIR) # 获取上传目录
    upload_dir.mkdir(parents=True, exist_ok=True) # 创建目录,自动创建父目录
    return upload_dir

# 解析已经存在的文件
@router.post('parse-path')
async def parse_file_by_path(body: ParseRequest):
    print('***********************************')
    '''按服务器本地路径解析文档（用于已有文件或者批处理）'''
    path = body.file_path

    # 判断path是否是完整路径(其实不一定是真正的完整路径,因为path可能是一个相对路径)
    #os.path.dirname只是把路径中最后的文件名去掉了而已
    if os.path.dirname(path) :
        target_path = path
    else:
        # 只是一个单独的文件名，需要拼接上传目录
        upload_dir = ensure_upload_dir_exists()
        # Path对象拼接另一个Path对象或字符串的特有写法,但不能两边都是字符串
        target_path = str(upload_dir/path)
    
    # 判断文件是否存在
    # `os.path.isfile(path)`：**判断给定路径，是不是一个真实存在的文件（不是文件夹）**
    if not os.path.isfile(target_path):
        raise HTTPException(status_code=400, detail=f"文件不存在:{target_path}")
    try:
        # parser/__init__.py里的函数
        result = parse_file(target_path)
        # 返回解析结果
        return {
            'message': '文件解析成功',
            'file_name': result.filename,
            'total_pages': result.total_pages,
            'chunks_count': result.total_chunks,
            'table_count': result.table_count,
            'chunks': result.chunks
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"解析文件失败:{str(e)}")

测试第二个接口时，将file_name修改为uploads文件下你要解析的文件，如下

程序会自动拼接为data\uploads\test_document.py

然后解析成功。注意虽然是相对路径，但是docling支持相对路径。

再做另一个测试

输入data/uploads/test_document.docx依然可以解析成功

因为执行了if，这算是一个完整的相对路径（从项目目录的以一级目录开始的）

但是如果输入/uploads/test_document.docx，那就报文件不存在了

因为此时依然执行if，把/uploads/test_document.docx当成完整相对路径了，所以说这个代码依然存在漏洞。但是企业开发时这里的问题在前端就被解决了，所以先不纠结了。

（3）上传文件接口的逻辑断连

第二个接口是解析已有文件，问题是已有文件从哪来的呢？对于企业来说有很多方式，我们能想到的方式第一个就是客户端上传到服务器进行解析，所以文件就顺便保存到服务器上了，第二种是企业员工将文件放到服务器中的某个文件夹里。

所以现在要添加一个上传文件解释时，在解释之前将文件保存到服务器的功能，要修改第一个接口。代码如下：

def _generate_filename(filename:str) -> str:
    ''' 生成文件名 '''
    name,ext = os.path.splitext(filename)
    timpestamp = time.time()
    return f"{name}_{timpestamp}{ext}"

# 上传文件进行解析的路由
@router.post("/upload")
async def upload_document(file: UploadFile = File(...)): # 参数：参数类型 = 默认值
    '''
    上传并解析文档
    支持的格式：pdf,html,md,docx
    '''
    # # str
    # print(type(file.filename))

    # 获取转小写的文件格式
    ext = Path(file.filename).suffix.lower()
    file_bytes = await file.read()

    #判断文件的大小
    if len(file_bytes) > settings.MAX_FILE_SIZE:
        size_mb = settings.MAX_FILE_SIZE / 1024 / 1024
        raise HTTPException(status_code=400, detail=f"文件大小不能超过{size_mb:.2f}MB")
    # 判断文件格式
    if ext not in settings.ALLOWED_EXTENSIONS:
        raise HTTPException(status_code=400, detail=f"不支持的文件格式: {ext}")

    # 保存数据
    upload_dir = ensure_upload_dir_exists()
    # 生成新的文件名
    new_filename = _generate_filename(file.filename)
    # 完整路径
    target_path = str(upload_dir/new_filename)

    # 保存文件
    with open(target_path, 'wb') as f:
        f.write(file_bytes)

    # 文件格式和大小都符合要求，调用解析器进行解析
    try:
        # 获得工厂类实例
        parser = ParserFactory.get_parser(file.filename)
        # 解析文件
        result = parser.parse_from_bytes(file_bytes=file_bytes, filename=file.filename)

        # 返回解析结果
        return {
            'message': '文件解析成功',
            'file_name': result.filename,
            'total_pages': result.total_pages,
            'chunks_count': result.total_chunks,
            'table_count': result.table_count,
            'chunks': result.chunks
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"文件解析失败: {str(e)}")

此时执行第一个接口上传的文件也会保存到服务段的指定路径里。

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/weixin_44593822/article/details/168445385