LlamaIndex SimpleDirectoryReader — file type not recognized or metadata lost

from llama_index.core import SimpleDirectoryReader, Document
from llama_index.readers.file import PyMuPDFReader
from typing import List
import json

# Custom reader for .json files (not in defaults)
from llama_index.core.readers.base import BaseReader

class JSONReader(BaseReader):
    def load_data(self, file, extra_info=None) -> List[Document]:
        with open(file, "r", encoding="utf-8") as f:
            data = json.load(f)
        # Serialise back to text for embedding
        text = json.dumps(data, indent=2, ensure_ascii=False)
        return [Document(text=text, metadata={"source": str(file), **(extra_info or {})})]

# Custom reader for .parquet
class ParquetReader(BaseReader):
    def load_data(self, file, extra_info=None) -> List[Document]:
        import pyarrow.parquet as pq
        table = pq.read_table(file)
        df = table.to_pandas()
        # Emit one document per row (or aggregate — your choice)
        docs = []
        for i, row in df.iterrows():
            text = "\n".join(f"{col}: {val}" for col, val in row.items())
            docs.append(Document(text=text, metadata={
                "source": str(file),
                "row_index": i,
                **(extra_info or {}),
            }))
        return docs

reader = SimpleDirectoryReader(
    input_dir="./data",
    recursive=True,
    file_extractor={
        ".json": JSONReader(),
        ".parquet": ParquetReader(),
        ".pdf": PyMuPDFReader(),        # override default with PyMuPDF for better extraction
    },
    required_exts=[".pdf", ".docx", ".md", ".txt", ".json", ".parquet"],
    exclude_hidden=True,
    filename_as_id=True,
)

documents = reader.load_data()
print(f"Loaded {len(documents)} documents")
for doc in documents[:3]:
    print(f"  {doc.metadata.get('source', '?')} -> {len(doc.text)} chars")

from llama_index.core import SimpleDirectoryReader, Document
from llama_index.core.readers.base import BaseReader
import chardet

class SafeTextReader(BaseReader):
    """Text reader that detects encoding and falls back gracefully."""

    def load_data(self, file, extra_info=None):
        # First, detect encoding
        with open(file, "rb") as f:
            raw = f.read()
        detected = chardet.detect(raw)
        encoding = detected["encoding"] or "utf-8"
        confidence = detected["confidence"]

        try:
            text = raw.decode(encoding)
        except UnicodeDecodeError:
            # Fallback: replace un-decodable chars
            text = raw.decode(encoding or "utf-8", errors="replace")

        return [Document(
            text=text,
            metadata={
                "source": str(file),
                "encoding": encoding,
                "encoding_confidence": confidence,
                **(extra_info or {}),
            },
        )]

reader = SimpleDirectoryReader(
    input_dir="./legacy_data",
    recursive=True,
    file_extractor={".txt": SafeTextReader(), ".log": SafeTextReader()},
)

docs = reader.load_data()

# Audit encoding detection
low_confidence = [d for d in docs if d.metadata.get("encoding_confidence", 1) < 0.9]
if low_confidence:
    print(f"⚠ {len(low_confidence)} documents had low-confidence encoding detection")
    for d in low_confidence[:5]:
        print(f"  {d.metadata['source']} -> {d.metadata['encoding']} "
              f"({d.metadata['encoding_confidence']:.0%})")

from pathlib import Path
from datetime import datetime
from llama_index.core import SimpleDirectoryReader

def file_metadata_fn(file_path: str) -> dict:
    """Attach metadata based on file location and stats."""
    p = Path(file_path)
    stat = p.stat()

    # Derive category from directory structure
    parts = p.parts
    category = parts[-3] if len(parts) >= 3 else "uncategorized"

    return {
        "file_name": p.name,
        "file_path": str(p.resolve()),
        "file_type": p.suffix.lstrip("."),
        "file_size_bytes": stat.st_size,
        "created_at": datetime.fromtimestamp(stat.st_ctime).isoformat(),
        "modified_at": datetime.fromtimestamp(stat.st_mtime).isoformat(),
        "category": category,
        # Add your own domain-specific tags
        "department": "engineering" if "eng" in str(p).lower() else "other",
    }

reader = SimpleDirectoryReader(
    input_dir="./docs",
    recursive=True,
    file_metadata=file_metadata_fn,   # ← attaches to every doc
    exclude_hidden=True,
    exclude=[".git", "node_modules", "__pycache__", "*.tmp"],
    required_exts=[".md", ".pdf", ".txt"],
)

docs = reader.load_data()

# Metadata flows through the whole pipeline —
# available on retrieved nodes, so downstream can filter/cite by category etc.
for doc in docs[:3]:
    print(doc.metadata)

# On queries, metadata is preserved on the nodes for filter push-down:
from llama_index.core.vector_stores import MetadataFilter, MetadataFilters
filters = MetadataFilters(filters=[
    MetadataFilter(key="category", value="policies"),
    MetadataFilter(key="department", value="engineering"),
])
# Use filters on the query engine (see page 109)