Underground-Digital/Workflow-Engine

1"""Document loader helpers."""2 3import concurrent.futures4from pathlib import Path5from typing import NamedTuple, Optional, cast6 7 8class FileEncoding(NamedTuple):9    """A file encoding as the NamedTuple."""10 11    encoding: Optional[str]12    """The encoding of the file."""13    confidence: float14    """The confidence of the encoding."""15    language: Optional[str]16    """The language of the file."""17 18 19def detect_file_encodings(file_path: str, timeout: int = 5) -> list[FileEncoding]:20    """Try to detect the file encoding.21 22    Returns a list of `FileEncoding` tuples with the detected encodings ordered23    by confidence.24 25    Args:26        file_path: The path to the file to detect the encoding for.27        timeout: The timeout in seconds for the encoding detection.28    """29    import chardet30 31    def read_and_detect(file_path: str) -> list[dict]:32        rawdata = Path(file_path).read_bytes()33        return cast(list[dict], chardet.detect_all(rawdata))34 35    with concurrent.futures.ThreadPoolExecutor() as executor:36        future = executor.submit(read_and_detect, file_path)37        try:38            encodings = future.result(timeout=timeout)39        except concurrent.futures.TimeoutError:40            raise TimeoutError(f"Timeout reached while detecting encoding for {file_path}")41 42    if all(encoding["encoding"] is None for encoding in encodings):43        raise RuntimeError(f"Could not detect encoding for {file_path}")44    return [FileEncoding(**enc) for enc in encodings if enc["encoding"] is not None]45