Method load_file_content

scrapegraphai/nodes/fetch_node.py:171–221

Loads the content of a file based on its input type. Parameters: source (str): The path to the source file. input_type (str): The type of the input file (e.g., "pdf", "csv", "json", "xml", "md"). Returns: list: A list containing a Document object wi

(self, source, input_type)

        return state

    def load_file_content(self, source, input_type):
        """
        Loads the content of a file based on its input type.

        Parameters:
        source (str): The path to the source file.
        input_type (str): The type of the input file (e.g., "pdf", "csv", "json", "xml", "md").

        Returns:
        list: A list containing a Document object with the loaded content and metadata.
        """

        if input_type == "pdf":
            from langchain_community.document_loaders import PyPDFLoader
            loader = PyPDFLoader(source)
            # PyPDFLoader.load() can be blocking for large PDFs. Run it in a thread and
            # enforce the configured timeout if provided.
            if self.timeout is None:
                return loader.load()
            else:
                with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
                    future = executor.submit(loader.load)
                    try:
                        return future.result(timeout=self.timeout)
                    except concurrent.futures.TimeoutError:
                        raise TimeoutError(
                            f"PDF parsing exceeded timeout of {self.timeout} seconds"
                        )
        elif input_type == "csv":
            try:
                import pandas as pd
            except ImportError:
                raise ImportError(
                    "pandas is not installed. Please install it using `pip install pandas`."
                )
            return [
                Document(
                    page_content=str(pd.read_csv(source)), metadata={"source": "csv"}
                )
            ]
        elif input_type == "json":
            with open(source, encoding="utf-8") as f:
                return [
                    Document(
                        page_content=str(json.load(f)), metadata={"source": "json"}
                    )
                ]
        elif input_type == "xml" or input_type == "md":
            with open(source, "r", encoding="utf-8") as f:
                data = f.read()
            return [Document(page_content=data, metadata={"source": input_type})]

    def handle_local_source(self, state, source):
        """