knowledge_base/loader.py

import json
import os

from langchain_core.documents import Document

from knowledge_base.formatters import json_to_text

# Root-level files and their source types
_ROOT_FILES: dict[str, str] = {
    "personal.json":     "personal",
    "education.json":    "education",
    "skills.json":       "skills",
    "experience.json":   "experience",
    "volunteering.json": "volunteering",
    "certificates.json": "certificates",
    "references.json":   "references",
}


def _load_json(filepath: str) -> dict | None:
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception as exc:
        print(f"❌ Error reading {filepath}: {exc}")
        return None


def _load_root_files(base_dir: str) -> list[Document]:
    docs = []
    for filename, source_type in _ROOT_FILES.items():
        filepath = os.path.join(base_dir, filename)
        if not os.path.exists(filepath):
            print(f"⚠️  Not found: {filepath}")
            continue
        if (data := _load_json(filepath)) is not None:
            docs.append(Document(
                page_content=json_to_text(data, source_type),
                metadata={"source": filename, "type": source_type},
            ))
            print(f"✅ Loaded: {filename}")
    return docs


def _load_projects(base_dir: str) -> list[Document]:
    docs = []
    projects_dir = os.path.join(base_dir, "projects")
    if not os.path.exists(projects_dir):
        return docs

    for fname in sorted(os.listdir(projects_dir)):
        if not fname.endswith(".json"):
            continue
        filepath = os.path.join(projects_dir, fname)
        if (data := _load_json(filepath)) is not None:
            docs.append(Document(
                page_content=json_to_text(data, "project"),
                metadata={
                    "source":     f"projects/{fname}",
                    "type":       "project",
                    "title":      data.get("title", fname),
                    "isFeatured": data.get("isFeatured", False),
                    "demoLink":   data.get("demoLink", ""),
                    "githubLink": data.get("githubLink", ""),
                    "category":   ", ".join(data.get("category", [])),
                },
            ))
            print(f"✅ Loaded project: {fname}")
    return docs


def _load_articles(base_dir: str) -> list[Document]:
    docs = []
    articles_dir = os.path.join(base_dir, "articles")
    if not os.path.exists(articles_dir):
        return docs

    for fname in sorted(os.listdir(articles_dir)):
        if not fname.endswith(".json"):
            continue
        filepath = os.path.join(articles_dir, fname)
        if (data := _load_json(filepath)) is not None:
            docs.append(Document(
                page_content=json_to_text(data, "article"),
                metadata={
                    "source": f"articles/{fname}",
                    "type":   "article",
                    "title":  data.get("title", fname),
                    "link":   data.get("link", data.get("demoLink", "")),
                },
            ))
            print(f"✅ Loaded article: {fname}")
    return docs


def load_knowledge_base(base_dir: str) -> list[Document]:
    docs = (
        _load_root_files(base_dir)
        + _load_projects(base_dir)
        + _load_articles(base_dir)
    )
    if not docs:
        raise ValueError("Knowledge base is empty.")
    return docs