mirror of
https://github.com/hwchase17/langchain
synced 2024-11-06 03:20:49 +00:00
f7a1fd91b8
So this arose from the https://github.com/langchain-ai/langchain/pull/18397 problem of document loaders not supporting `pathlib.Path`. This pull request provides more uniform support for Path as an argument. The core ideas for this upgrade: - if there is a local file path used as an argument, it should be supported as `pathlib.Path` - if there are some external calls that may or may not support Pathlib, the argument is immidiately converted to `str` - if there `self.file_path` is used in a way that it allows for it to stay pathlib without conversion, is is only converted for the metadata. Twitter handle: https://twitter.com/mwmajewsk
52 lines
1.6 KiB
Python
52 lines
1.6 KiB
Python
"""Document loader helpers."""
|
|
|
|
import concurrent.futures
|
|
from pathlib import Path
|
|
from typing import List, NamedTuple, Optional, Union, cast
|
|
|
|
|
|
class FileEncoding(NamedTuple):
|
|
"""File encoding as the NamedTuple."""
|
|
|
|
encoding: Optional[str]
|
|
"""The encoding of the file."""
|
|
confidence: float
|
|
"""The confidence of the encoding."""
|
|
language: Optional[str]
|
|
"""The language of the file."""
|
|
|
|
|
|
def detect_file_encodings(
|
|
file_path: Union[str, Path], timeout: int = 5
|
|
) -> List[FileEncoding]:
|
|
"""Try to detect the file encoding.
|
|
|
|
Returns a list of `FileEncoding` tuples with the detected encodings ordered
|
|
by confidence.
|
|
|
|
Args:
|
|
file_path: The path to the file to detect the encoding for.
|
|
timeout: The timeout in seconds for the encoding detection.
|
|
"""
|
|
import chardet
|
|
|
|
file_path = str(file_path)
|
|
|
|
def read_and_detect(file_path: str) -> List[dict]:
|
|
with open(file_path, "rb") as f:
|
|
rawdata = f.read()
|
|
return cast(List[dict], chardet.detect_all(rawdata))
|
|
|
|
with concurrent.futures.ThreadPoolExecutor() as executor:
|
|
future = executor.submit(read_and_detect, file_path)
|
|
try:
|
|
encodings = future.result(timeout=timeout)
|
|
except concurrent.futures.TimeoutError:
|
|
raise TimeoutError(
|
|
f"Timeout reached while detecting encoding for {file_path}"
|
|
)
|
|
|
|
if all(encoding["encoding"] is None for encoding in encodings):
|
|
raise RuntimeError(f"Could not detect encoding for {file_path}")
|
|
return [FileEncoding(**enc) for enc in encodings if enc["encoding"] is not None]
|