feat: add parser registry

Author: liningping

parsers/__init__.py

@@ -1 +1,13 @@
 # Parsers package
+
+from .parser_registry import (
+    PARSER_REGISTRY,
+    DocumentParser,
+    can_parse,
+    get_parser,
+    get_supported_formats,
+    list_registered_parsers,
+    register_parser,
+)
+
+__all__ = ['PARSER_REGISTRY', 'register_parser', 'DocumentParser', 'get_parser', 'can_parse', 'get_supported_formats', 'list_registered_parsers']

parsers/base_models.py

@@ -1,5 +1,4 @@
 import logging
-from abc import ABC, abstractmethod
 from enum import Enum
 from typing import Any
 
@@ -40,19 +39,3 @@ class DocumentData(BaseModel):
     processing_time: float = 0
     success: bool
     error_message: str | None = None
-
-class DocumentParser(ABC):
-    """文档解析器基类"""
-
-    def __init__(self) -> None:
-        self.supported_formats: list[str] = Field(default_factory=list)
-
-    @abstractmethod
-    async def parse(self, file_path: str) -> DocumentData:
-        """解析文档"""
-        pass
-
-    @abstractmethod
-    def can_parse(self, file_path: str) -> bool:
-        """检查是否可以解析该文件"""
-        return any(file_path.lower().endswith(fmt) for fmt in self.supported_formats)

parsers/document_parser.py

@@ -29,13 +29,14 @@
     ChunkData,
     ChunkType,
     DocumentData,
-    DocumentParser,
     TableDataItem,
 )
+from parsers.parser_registry import DocumentParser, register_parser
 
 logger = logging.getLogger(__name__)
 
 
+@register_parser(['.docx'])
 class DocxDocumentParser(DocumentParser):
     """DOCX文档解析器
 
@@ -46,32 +47,20 @@ class DocxDocumentParser(DocumentParser):
     def __init__(self) -> None:
         """初始化解析器"""
         super().__init__()
-        self.supported_formats = [".docx"]
         self._converter = DocumentConverter(
             format_options={InputFormat.DOCX: WordFormatOption(pipeline_cls=SimplePipeline)},
             allowed_formats=[InputFormat.DOCX]
         )
         logger.debug("DocxDocumentParser initialized with SimplePipeline")
 
-    def can_parse(self, file_path: str) -> bool:
-        """检查是否可以解析该文件
-
-        Args:
-            file_path: 文件路径
-
-        Returns:
-            bool: 是否支持该文件格式
-        """
-        return any(file_path.lower().endswith(fmt) for fmt in self.supported_formats)
-
     async def parse(self, file_path: str) -> DocumentData:
         """异步解析DOCX文件
 
         Args:
             file_path: DOCX文件路径
 
         Returns:
-            ParseResult: 解析结果，包含标题、内容、处理时间和错误信息
+            DocumentData: 解析结果，包含标题、内容、处理时间和错误信息
         """
         start_time = time.time()
         try:

parsers/docx_parser.py

@@ -23,9 +23,9 @@
     ChunkData,
     ChunkType,
     DocumentData,
-    DocumentParser,
     TableDataItem,
 )
+from parsers.parser_registry import DocumentParser, register_parser
 
 # 忽略 openpyxl 的特定警告
 warnings.filterwarnings('ignore', category=UserWarning, module='openpyxl')
@@ -49,6 +49,7 @@ class ExcelParseError(Exception):
     pass
 
 
+@register_parser(['.xlsx', '.xls'])
 class ExcelParser(DocumentParser):
     """Excel文件解析器类"""
 
@@ -61,17 +62,6 @@ def __init__(self, config: ExcelParseConfig | None = None):
         super().__init__()
         self.config: ExcelParseConfig = config or ExcelParseConfig()
         self.image_index: int = 0
-        self.supported_formats: list[str] = ['.xlsx', '.xls']
-
-    def can_parse(self, file_path: str) -> bool:
-        """
-        验证输入文件
-        Args:
-            file_path: 文件路径
-        Returns:
-            bool: 是否支持解析
-        """
-        return any(file_path.lower().endswith(fmt) for fmt in self.supported_formats)
 
     async def parse(self, excel_path: str) -> DocumentData:
         """
@@ -183,7 +173,7 @@ def _process_image_object(self, img_obj: Image) -> ChunkData | None:
         Args:
             img_obj: 图片对象
         Returns:
-            Optional[DocumentData]: 图片信息，处理失败时返回None
+            ChunkData|None: 图片信息，处理失败时返回None
         """
         try:
             # 获取图片数据

parsers/excel_parser.py

@@ -0,0 +1,132 @@
+"""
+解析器注册器模块
+
+提供基于装饰器的解析器自动注册机制，支持多种文件格式的解析器注册和查找。
+"""
+
+import logging
+from abc import ABC, abstractmethod
+from collections.abc import Callable
+from pathlib import Path
+
+from .base_models import DocumentData
+
+logger = logging.getLogger(__name__)
+
+# 全局解析器注册表
+PARSER_REGISTRY: dict[str, type['DocumentParser']] = {}
+
+
+class DocumentParser(ABC):
+    """文档解析器基类"""
+
+    @abstractmethod
+    async def parse(self, file_path: str) -> DocumentData:
+        """解析文档"""
+        pass
+
+
+def register_parser(suffixes: list[str]) -> Callable[[type['DocumentParser']], type['DocumentParser']]:
+    """
+    解析器注册装饰器
+
+    Args:
+        suffixes: 支持的文件扩展名列表，如 ['.docx', '.doc']
+
+    Returns:
+        装饰器函数
+
+    Example:
+        @register_parser(['.docx'])
+        class DocxDocumentParser(DocumentParser):
+            ...
+    """
+    def decorator(cls: type['DocumentParser']) -> type['DocumentParser']:
+        # 验证类是否继承自 DocumentParser
+        if not issubclass(cls, DocumentParser):
+            raise TypeError(f"解析器类 {cls.__name__} 必须继承自 DocumentParser")
+
+        # 注册到全局注册表
+        for suffix in suffixes:
+            suffix = suffix.lower()  # 统一转换为小写
+            if suffix in PARSER_REGISTRY:
+                logger.warning(f"覆盖已存在的解析器: {suffix} -> {cls.__name__}")
+            PARSER_REGISTRY[suffix] = cls
+            logger.info(f"注册解析器: {suffix} -> {cls.__name__}")
+
+        return cls
+
+    return decorator
+
+
+def get_parser(file_path: str) -> 'DocumentParser' | None:
+    """
+    根据文件路径获取合适的解析器实例
+
+    Args:
+        file_path: 文件路径
+
+    Returns:
+        解析器实例，如果没有找到则返回 None
+    """
+    file = Path(file_path)
+    suffix = file.suffix.lower()
+
+    if suffix not in PARSER_REGISTRY:
+        logger.warning(f"未找到支持 {suffix} 格式的解析器")
+        return None
+
+    parser_class = PARSER_REGISTRY[suffix]
+    try:
+        return parser_class()
+    except Exception as e:
+        logger.error(f"创建解析器实例失败: {parser_class.__name__}, 错误: {e}")
+        return None
+
+
+def can_parse(file_path: str) -> bool:
+    """
+    检查文件是否可以被解析
+
+    Args:
+        file_path: 文件路径
+
+    Returns:
+        bool: 是否支持该文件格式
+    """
+    file = Path(file_path)
+    suffix = file.suffix.lower()
+    return suffix in PARSER_REGISTRY
+
+
+def get_supported_formats() -> list[str]:
+    """
+    获取所有支持的文件格式
+
+    Returns:
+        支持的文件扩展名列表
+    """
+    return list(PARSER_REGISTRY.keys())
+
+
+def get_parser_class(suffix: str) -> type['DocumentParser'] | None:
+    """
+    根据文件扩展名获取解析器类
+
+    Args:
+        suffix: 文件扩展名，如 '.docx'
+
+    Returns:
+        解析器类，如果没有找到则返回 None
+    """
+    return PARSER_REGISTRY.get(suffix.lower())
+
+
+def list_registered_parsers() -> dict[str, str]:
+    """
+    列出所有已注册的解析器
+
+    Returns:
+        扩展名到解析器类名的映射字典
+    """
+    return {suffix: cls.__name__ for suffix, cls in PARSER_REGISTRY.items()}

parsers/parser_registry.py

@@ -4,13 +4,13 @@
 from sanic import Sanic
 
 from enhancers.information_enhancer import InformationEnhancerFactory
+from parsers import get_parser
 from parsers.base_models import ChunkData
-from parsers.document_parser import DocumentParserFactory
 
 
 async def worker(app: Sanic) -> dict[str, Any]:
     # 使用工厂获取合适的解析器
-    parser_factory = DocumentParserFactory()
+
     enhancer_factory = InformationEnhancerFactory()
     redis = app.ctx.redis
     while True:
@@ -19,7 +19,10 @@ async def worker(app: Sanic) -> dict[str, Any]:
             await asyncio.sleep(1)
             continue
         file_path = task.get("file_path")
-        parse_result = await parser_factory.parse_document(file_path)
+        parser = get_parser(file_path)
+        if not parser:
+            continue
+        parse_result = await parser.parse(file_path)
         if not parse_result.success:
             continue
         chunk_list = parse_result.texts + parse_result.tables + parse_result.images + parse_result.formulas