Repository: imartinez/privateGPT
File: scripts/ingest_folder.py (Python)

Code before:
import argparse
import sys
from pathlib import Path

from private_gpt.di import root_injector
from private_gpt.server.ingest.ingest_service import IngestService
from private_gpt.server.ingest.ingest_watcher import IngestWatcher

ingest_service = root_injector.get(IngestService)


parser = argparse.ArgumentParser(prog="ingest_folder.py")
parser.add_argument("folder", help="Folder to ingest")
parser.add_argument(
    "--watch",
    help="Watch for changes",
    action=argparse.BooleanOptionalAction,
    default=False,
)
args = parser.parse_args()


def _recursive_ingest_folder(folder_path: Path) -> None:
    for file_path in folder_path.iterdir():
        if file_path.is_file():
            _do_ingest(file_path)
        elif file_path.is_dir():
            _recursive_ingest_folder(file_path)


def _do_ingest(changed_path: Path) -> None:
    if changed_path.exists():
        print(f"\nIngesting {changed_path}")
        ingest_service.ingest(changed_path.name, changed_path)


path = Path(args.folder)
if not path.exists():
    raise ValueError(f"Path {args.folder} does not exist")

_recursive_ingest_folder(path)
if args.watch:
    print(f"Watching {args.folder} for changes, press Ctrl+C to stop...")
    watcher = IngestWatcher(args.folder, _do_ingest)
    watcher.start()


Review comment (bug):
Using `exception` instead of `error` is [adding the stack trace after the log](https://docs.python.org/3/library/logging.html#logging.Logger.exception) - I don't know if you want it, or if you just want the string representation of the error, so putting it as a suggestion:

```suggestion
    except Exception as e:
        logger.exception(f"Failed to ingest document: {changed_path}. Error: {e}")
```

Code after:
        if file_path.is_file():
            total_documents += 1
        elif file_path.is_dir():
            count_documents(file_path)


def _recursive_ingest_folder(folder_path: Path) -> None:
    global current_document_count, total_documents
    for file_path in folder_path.iterdir():
        if file_path.is_file():
            current_document_count += 1
            progress_msg = f"Document {current_document_count} of {total_documents} ({(current_document_count / total_documents) * 100:.2f}%)"
            logger.info(progress_msg)
            _do_ingest(file_path)
        elif file_path.is_dir():
            _recursive_ingest_folder(file_path)


def _do_ingest(changed_path: Path) -> None:
    try:
        if changed_path.exists():
            logger.info(f"Started ingesting {changed_path}")
            ingest_service.ingest(changed_path.name, changed_path)
            logger.info(f"Completed ingesting {changed_path}")
    except Exception as e:
        logger.error(f"Failed to ingest document: {changed_path}. Error: {e}")


path = Path(args.folder)
if not path.exists():
    raise ValueError(f"Path {args.folder} does not exist")

# Count total documents before ingestion
count_documents(path)

_recursive_ingest_folder(path)
if args.watch:
    logger.info(f"Watching {args.folder} for changes, press Ctrl+C to stop...")
    watcher = IngestWatcher(args.folder, _do_ingest)
    watcher.start()