Spaces:
Sleeping
Sleeping
File size: 3,464 Bytes
f63b452 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | 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 |