Mokhtar
feat(production): complete phase 1-4 refactor with hybrid retrieval CAG architecture, E5 dense embeddings, grounding validator, precomputed cold-start index, and 24/24 passing system tests
d54db1a | # 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] | |
| } | |