File size: 4,309 Bytes
d54db1a | 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 105 106 107 108 | # knowledge/loader.py — Recursive document loader with full metadata tracking
import os
import re
import hashlib
from pathlib import Path
from typing import List, Dict, Tuple
from knowledge.models import Document, DocumentMetadata
class KnowledgeLoader:
"""Recursive loader for university knowledge files (.md, .txt)."""
def __init__(self, data_dir: str):
self.data_dir = Path(data_dir)
self.indexed_files: List[Document] = []
self.skipped_files: List[Dict[str, str]] = []
self.stats: Dict[str, int] = {
"indexed_count": 0,
"skipped_count": 0,
"total_chars": 0,
"total_files": 0
}
def _extract_title(self, content: str, default_name: str) -> str:
"""Extract top H1 header or fallback to filename."""
match = re.search(r'^#\s+(.+)$', content, re.MULTILINE)
if match:
return match.group(1).strip()
return default_name.replace('.md', '').replace('.txt', '').replace('_', ' ').title()
def _generate_doc_id(self, filepath: Path) -> str:
rel_path = filepath.relative_to(self.data_dir)
return hashlib.sha256(str(rel_path).encode('utf-8')).hexdigest()[:12]
def load_documents(self) -> List[Document]:
"""Recursively scan data directory for supported files (.md, .txt)."""
self.indexed_files = []
self.skipped_files = []
if not self.data_dir.exists():
print(f"⚠️ Knowledge directory does not exist: {self.data_dir}")
return []
# Recursively scan directory
all_paths = sorted(list(self.data_dir.rglob("*")))
for path in all_paths:
if not path.is_file():
continue
rel_path_str = str(path.relative_to(self.data_dir))
# Skip hidden files or backup folders
if path.name.startswith(".") or "backup" in rel_path_str.lower() or "venv" in rel_path_str.lower():
self.skipped_files.append({"path": rel_path_str, "reason": "Hidden or backup file"})
continue
suffix = path.suffix.lower()
if suffix not in [".md", ".txt"]:
self.skipped_files.append({"path": rel_path_str, "reason": f"Unsupported extension ({suffix})"})
continue
try:
with open(path, "r", encoding="utf-8") as f:
content = f.read().strip()
if not content:
self.skipped_files.append({"path": rel_path_str, "reason": "Empty file"})
continue
doc_id = self._generate_doc_id(path)
title = self._extract_title(content, path.name)
source_type = "markdown" if suffix == ".md" else "text"
metadata = DocumentMetadata(
doc_id=doc_id,
filename=path.name,
path=rel_path_str,
title=title,
section="",
subsection="",
language="ar",
source_type=source_type
)
doc = Document(metadata=metadata, content=content)
self.indexed_files.append(doc)
self.stats["total_chars"] += len(content)
except Exception as e:
self.skipped_files.append({"path": rel_path_str, "reason": f"Read error: {str(e)}"})
self.stats["indexed_count"] = len(self.indexed_files)
self.stats["skipped_count"] = len(self.skipped_files)
self.stats["total_files"] = len(self.indexed_files) + len(self.skipped_files)
print(f"📚 Recursive Loader: Indexed {len(self.indexed_files)} files ({self.stats['total_chars']} chars). Skipped {len(self.skipped_files)} files.")
return self.indexed_files
def get_summary(self) -> Dict:
return {
"indexed_count": self.stats["indexed_count"],
"skipped_count": self.stats["skipped_count"],
"skipped_files": self.skipped_files,
"total_chars": self.stats["total_chars"],
"indexed_filenames": [doc.metadata.filename for doc in self.indexed_files]
}
|