wehub-resource-sync
156 lines
5.1 KiB
Python
156 lines
5.1 KiB
Python
"""Unified parser contract for native + external parser engines.
|
|
|
|
Every engine (native docx, mineru, docling, legacy, plus the internal
|
|
``reuse``/``passthrough`` format handlers) implements :class:`BaseParser`.
|
|
The pipeline dispatches through the registry
|
|
(:mod:`lightrag.parser.registry`) instead of a growing ``if engine == …``
|
|
chain.
|
|
|
|
Design notes:
|
|
|
|
- ``BaseParser`` carries *behaviour only* (the ``parse`` coroutine and any
|
|
engine-private hooks). Capability metadata (supported suffixes, queue
|
|
group, endpoint requirements) lives in the registry's lightweight
|
|
``ParserSpec`` table so capability queries never import a parser
|
|
implementation.
|
|
- ``ParseResult.to_dict()`` emits only semantically-present fields so the
|
|
returned dict stays byte-for-byte compatible with the pre-refactor
|
|
``parse_native``/``parse_mineru``/``parse_docling`` return shapes (the
|
|
worker reads these by ``.get(...)``).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
if TYPE_CHECKING:
|
|
from lightrag.sidecar.ir import IRDoc # noqa: F401
|
|
|
|
|
|
@dataclass
|
|
class ResolvedSource:
|
|
"""The common parse preamble shared by every engine."""
|
|
|
|
source_path: Path
|
|
document_name: str
|
|
parsed_dir: Path
|
|
|
|
|
|
@dataclass
|
|
class ParseContext:
|
|
"""Inputs handed to :meth:`BaseParser.parse`.
|
|
|
|
Wraps the LightRAG instance plus the per-document handles so a parser
|
|
does not receive a bare ``self``. Convenience helpers lazily import
|
|
pipeline-layer functions at call time, keeping this module import-cheap
|
|
and free of import cycles.
|
|
"""
|
|
|
|
rag: Any
|
|
doc_id: str
|
|
file_path: str
|
|
content_data: dict[str, Any]
|
|
|
|
def source_path(self, parser_engine: str) -> Path:
|
|
"""Resolve the on-disk source file for this document."""
|
|
from lightrag.pipeline import (
|
|
_call_source_file_resolver,
|
|
_read_source_file,
|
|
)
|
|
|
|
return Path(
|
|
_call_source_file_resolver(
|
|
self.rag,
|
|
self.file_path,
|
|
source_file=_read_source_file(self.content_data),
|
|
parser_engine=parser_engine,
|
|
)
|
|
)
|
|
|
|
def resolve(self, parser_engine: str) -> ResolvedSource:
|
|
"""Resolve ``(source_path, document_name, parsed_dir)``.
|
|
|
|
Mirrors the preamble shared by ``parse_mineru``/``parse_docling``/
|
|
``parse_native``: canonicalize the document name defensively (so
|
|
direct callers may pass absolute or hint-bearing paths) and derive
|
|
the ``__parsed__/<base>.parsed/`` output directory.
|
|
"""
|
|
from lightrag.utils_pipeline import (
|
|
normalize_document_file_path,
|
|
parsed_artifact_dir_for,
|
|
)
|
|
|
|
source_path = self.source_path(parser_engine)
|
|
document_name = normalize_document_file_path(self.file_path)
|
|
if document_name == "unknown_source":
|
|
document_name = source_path.name or f"{self.doc_id}.bin"
|
|
parsed_dir = parsed_artifact_dir_for(
|
|
document_name, parent_hint=source_path.parent
|
|
)
|
|
return ResolvedSource(source_path, document_name, parsed_dir)
|
|
|
|
async def archive_source(self, source_path: str) -> str | None:
|
|
"""Archive the source after a successful parse + full_docs sync.
|
|
|
|
Resolved through the pipeline module's namespace (where the function
|
|
is imported) so existing tests that patch
|
|
``lightrag.pipeline.archive_docx_source_after_full_docs_sync`` keep
|
|
intercepting it now that the call site lives in the parser layer.
|
|
"""
|
|
import lightrag.pipeline as _pipeline
|
|
|
|
return await _pipeline.archive_docx_source_after_full_docs_sync(source_path)
|
|
|
|
|
|
@dataclass
|
|
class ParseResult:
|
|
"""Structured parser output.
|
|
|
|
``to_dict`` only emits fields that carry meaning so the dict matches the
|
|
pre-refactor return shapes exactly (no spurious ``None``/``False`` keys).
|
|
"""
|
|
|
|
doc_id: str
|
|
file_path: str
|
|
parse_format: str
|
|
content: str
|
|
blocks_path: str = ""
|
|
parse_engine: str | None = None
|
|
parse_stage_skipped: bool = False
|
|
parse_warnings: dict[str, Any] | None = None
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
out: dict[str, Any] = {
|
|
"doc_id": self.doc_id,
|
|
"file_path": self.file_path,
|
|
"parse_format": self.parse_format,
|
|
"content": self.content,
|
|
"blocks_path": self.blocks_path,
|
|
}
|
|
if self.parse_engine is not None:
|
|
out["parse_engine"] = self.parse_engine
|
|
if self.parse_stage_skipped:
|
|
out["parse_stage_skipped"] = True
|
|
if self.parse_warnings:
|
|
out["parse_warnings"] = self.parse_warnings
|
|
return out
|
|
|
|
|
|
class BaseParser(ABC):
|
|
"""Abstract base for every parser engine.
|
|
|
|
Subclasses set ``engine_name`` and implement :meth:`parse`. Capability
|
|
metadata (suffixes/queue group/endpoint) is declared in the registry
|
|
``ParserSpec``, not here.
|
|
"""
|
|
|
|
engine_name: str
|
|
|
|
@abstractmethod
|
|
async def parse(self, ctx: ParseContext) -> ParseResult:
|
|
"""Parse one document and return its :class:`ParseResult`."""
|
|
...
|