excel_processor.py

"""
Excel ingestion: each non-empty data row becomes one Document with
header:value pipe-delimited text.
"""
import os
import logging

from langchain_core.documents import Document
import openpyxl

from config.settings import RAW_EXCEL_PATH
from processors.source_metadata import build_source_metadata

logger = logging.getLogger(__name__)


def process_excel() -> list[Document]:
    """Load Excel files as Document objects (not yet chunked)."""
    documents: list[Document] = []
    if not os.path.exists(RAW_EXCEL_PATH):
        print("Excel folder not found")
        return documents

    for filename in os.listdir(RAW_EXCEL_PATH):
        if not filename.endswith((".xlsx", ".xls")):
            continue
        filepath = os.path.join(RAW_EXCEL_PATH, filename)
        print(f"Processing: {filename}")

        try:
            wb = openpyxl.load_workbook(filepath)
        except Exception as e:
            logger.warning("Cannot open workbook %s: %s", filename, e)
            print(f"[WARN] Skipping unreadable Excel file: {filename} ({e})")
            continue

        for sheet_name in wb.sheetnames:
            sheet = wb[sheet_name]
            rows = list(sheet.iter_rows(values_only=True))
            if not rows:
                continue

            headers = [str(h) if h else "" for h in rows[0]]
            for row in rows[1:]:
                row_text = " | ".join(
                    f"{headers[i]}: {str(cell)}"
                    for i, cell in enumerate(row) if cell
                )
                if row_text.strip():
                    documents.append(
                        Document(
                            page_content=row_text,
                            metadata=build_source_metadata(
                                filepath, "excel", sheet=sheet_name
                            ),
                        )
                    )

    print(f"Excel loading complete: {len(documents)} rows")
    return documents


if __name__ == "__main__":
    docs = process_excel()