# 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] }