Function parse

text_parse/parse.py:112–155

File parsing function, supports progress callback :param hparams: Hyperparameters :param progress_callback: Progress callback function :return: Path of the parsed file

(hparams: HyperParams, progress_callback=None)

def parse(hparams: HyperParams, progress_callback=None):
    """
    File parsing function, supports progress callback
    :param hparams: Hyperparameters
    :param progress_callback: Progress callback function
    :return: Path of the parsed file
    """
    assert os.path.exists(hparams.file_path), "File does not exist."
    file_type = hparams.file_path.split('.')[-1].lower()
    save_path = os.path.join(hparams.save_path, 'parsed_file')
    os.makedirs(save_path, exist_ok=True)

    if file_type in {'tex', 'txt', 'json'}:
        # Use file size instead of line count to estimate progress
        CHUNK_SIZE = 1024 * 1024  # 1MB
        file_size = os.path.getsize(hparams.file_path)
        read_bytes = 0
        content = []

        with open(hparams.file_path, 'r', encoding='utf-8') as f:
            while chunk := f.read(CHUNK_SIZE):
                content.append(chunk)
                read_bytes += len(chunk.encode('utf-8'))
                if progress_callback:
                    progress = int((read_bytes/file_size) * 70)
                    progress_callback(min(progress, 70))  # Ensure it doesn't exceed 70%

        # Save processed content
        save_file_path = os.path.join(
            save_path,
            f"{os.path.splitext(os.path.basename(hparams.file_path))[0]}.txt"
        )

        with open(save_file_path, 'w', encoding='utf-8') as f:
            f.writelines(content)

        if progress_callback:
            progress_callback(100)
        return save_file_path

    elif file_type == 'pdf':
        return pdf_parse(hparams.file_path, save_path, progress_callback)
    else:
        raise ValueError(f"Unsupported file type: {file_type}")