Spaces:
Paused
Paused
File size: 3,973 Bytes
9792ea7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 | # -*- coding: utf-8 -*-
"""Plain-text file parser."""
import os
from ...message import TextBlock
from .._document import Section
from ._base import ParserBase
class TextParser(ParserBase):
"""Parser for plain-text file formats.
Reads the entire file as UTF-8 text and returns a single
:class:`Section`. No internal boundaries are inferred — the file
is treated as one unstructured blob, leaving all splitting to a
downstream :class:`~agentscope.rag.ChunkerBase`.
Supports a fixed set of standard text-based IANA media types
(``text/plain``, ``text/markdown``, ``text/csv``, …). Use
``TextParser.supported_media_types`` to enumerate them.
"""
supported_media_types: list[str] = [
"text/plain",
"text/markdown",
"text/csv",
"text/html",
"text/x-rst",
"application/json",
"application/xml",
"application/x-yaml",
]
"""Standard IANA media types this parser handles."""
@classmethod
def supported_extensions(cls) -> list[str]:
"""Return the human-friendly text extensions.
Override the base reverse-lookup because
:func:`mimetypes.guess_all_extensions` returns a long tail of
developer-tool extensions for ``text/plain`` (``.bat`` /
``.c`` / ``.pl`` / ``.ksh`` / …) that have no place in a KB
file picker, and returns the empty list for
``application/x-yaml``.
"""
return [
".csv",
".htm",
".html",
".json",
".markdown",
".md",
".rst",
".txt",
".xml",
".yaml",
".yml",
]
def __init__(self, encoding: str = "utf-8") -> None:
"""Initialize the text parser.
Args:
encoding (`str`, defaults to ``"utf-8"``):
The text encoding used to decode the file bytes.
"""
self.encoding = encoding
async def parse(
self,
file: bytes | str,
filename: str,
) -> list[Section]:
"""Read the file as text and return a single :class:`Section`.
Args:
file (`bytes | str`):
The file content. ``bytes`` is decoded with the
configured encoding. ``str`` is disambiguated at
runtime: if it names an existing file on disk the
file is read and decoded; otherwise it is used
verbatim as pre-decoded text — letting local-mode
callers skip the encode → decode round trip.
filename (`str`):
The source filename, copied verbatim into
:attr:`Section.source`.
Returns:
`list[Section]`:
Always a one-element list containing the entire file
contents.
Raises:
`ValueError`: If the bytes cannot be decoded with the
configured encoding.
"""
if isinstance(file, str):
if os.path.isfile(file):
with open(file, "rb") as fp:
raw = fp.read()
try:
text = raw.decode(self.encoding)
except UnicodeDecodeError as e:
raise ValueError(
f"Failed to decode {filename!r} as "
f"{self.encoding!r}: {e}",
) from e
else:
text = file
else:
try:
text = file.decode(self.encoding)
except UnicodeDecodeError as e:
raise ValueError(
f"Failed to decode {filename!r} as "
f"{self.encoding!r}: {e}",
) from e
return [
Section(
content=TextBlock(text=text),
source=filename,
metadata={},
),
]
|