"""Safe read-only access to repository documentation and data files. Supports: .md, .txt, .docx, .xlsx, .csv, .json, .pdf (metadata only) """ from __future__ import annotations import csv import json from dataclasses import dataclass from datetime import datetime from enum import Enum from functools import lru_cache from importlib import import_module from pathlib import Path from typing import Any, Callable, cast from .models import JSONValue # Optional imports with proper fallbacks – using lower-case names to avoid Pylance constant redefinition warnings. DocxDocument: Any = None load_workbook: Callable[..., Any] | None = None pypdf: Any = None try: from docx import ( Document as _DocxDocument, # pyright: ignore[reportMissingTypeStubs] ) DocxDocument = _DocxDocument has_docx: bool = True except ImportError: has_docx = False try: from openpyxl import ( # type: ignore[import-untyped] load_workbook as _load_workbook, # pyright: ignore[reportMissingTypeStubs, reportUnknownVariableType] ) load_workbook = cast(Callable[..., Any], _load_workbook) has_openpyxl: bool = True except ImportError: has_openpyxl = False try: pypdf = import_module("pypdf") has_pypdf: bool = True except ImportError: has_pypdf = False __all__ = [ "DocumentationEntry", "DocumentationSource", "FileType", "create_docs_source", ] class FileType(Enum): """Supported document file types.""" MARKDOWN = "markdown" TEXT = "text" DOCX = "docx" XLSX = "xlsx" CSV = "csv" JSON = "json" PDF = "pdf" UNKNOWN = "unknown" @classmethod def from_extension(cls, ext: str) -> FileType: """Map file extension to FileType.""" ext = ext.lower().lstrip(".") mapping: dict[str, FileType] = { "md": cls.MARKDOWN, "markdown": cls.MARKDOWN, "txt": cls.TEXT, "text": cls.TEXT, "docx": cls.DOCX, "xlsx": cls.XLSX, "xls": cls.XLSX, "csv": cls.CSV, "json": cls.JSON, "pdf": cls.PDF, } return mapping.get(ext, cls.UNKNOWN) @property def is_editable(self) -> bool: """Whether this file type can be edited in a text editor.""" return self in {FileType.MARKDOWN, FileType.TEXT, FileType.CSV, FileType.JSON} @property def is_binary(self) -> bool: """Whether this file type is binary.""" return self in {FileType.DOCX, FileType.XLSX, FileType.PDF} @dataclass(frozen=True, slots=True) class DocumentationEntry: """Rich metadata for a documentation or data file.""" name: str path: str size_bytes: int file_type: FileType modified_time: str content_preview: str | None = None word_count: int | None = None line_count: int | None = None sheet_names: tuple[str, ...] | None = None supported: bool = True def to_json(self) -> dict[str, JSONValue]: """Return JSON-ready document metadata.""" result: dict[str, JSONValue] = { "name": self.name, "path": self.path, "size_bytes": self.size_bytes, "file_type": self.file_type.value, "modified_time": self.modified_time, "supported": self.supported, } if self.content_preview is not None: result["content_preview"] = self.content_preview if self.word_count is not None: result["word_count"] = self.word_count if self.line_count is not None: result["line_count"] = self.line_count if self.sheet_names is not None: result["sheet_names"] = list(self.sheet_names) return result class DocumentationSource: """Expose documentation and data files with multi-format support. Provides safe, read-only access to files below a fixed root directory. Supports content extraction for .md, .txt, .docx, .xlsx, .csv, .json. """ def __init__( self, docs_root: Path, max_preview_chars: int = 500, max_file_size_mb: int = 50, enable_caching: bool = True, ) -> None: """Initialize documentation source. Args: docs_root: Root directory for documentation files. max_preview_chars: Maximum characters for content preview. max_file_size_mb: Maximum file size to process (MB). enable_caching: Whether to cache file listings. """ self.docs_root = docs_root.resolve() self.max_preview_chars = max_preview_chars self.max_file_size_bytes = max_file_size_mb * 1024 * 1024 self.enable_caching = enable_caching def list_documents(self, recursive: bool = False) -> tuple[DocumentationEntry, ...]: """List all supported documents in stable order. Args: recursive: Whether to scan subdirectories recursively. Returns: Tuple of DocumentationEntry objects. """ if not self.docs_root.is_dir(): return () pattern = "**/*" if recursive else "*" entries: list[DocumentationEntry] = [] for path in sorted(self.docs_root.glob(pattern)): if not path.is_file(): continue if self._is_excluded(path): continue entry = self._build_entry(path) if entry is not None: entries.append(entry) return tuple(entries) def list_documents_by_type( self, file_type: FileType ) -> tuple[DocumentationEntry, ...]: """List documents filtered by file type.""" return tuple( entry for entry in self.list_documents(recursive=True) if entry.file_type == file_type ) def get_document(self, path: str) -> DocumentationEntry: """Get metadata for a specific document.""" resolved_path = self._resolve_path(path) if resolved_path is None or not resolved_path.is_file(): raise FileNotFoundError(f"Document not found: {path}") entry = self._build_entry(resolved_path) if entry is None: raise ValueError(f"Unsupported file type: {path}") return entry def read(self, path: str) -> str: """Read the full content of a document (convenience alias).""" return self.read_content(path) def read_content(self, path: str) -> str: """Read the full content of a document (text extraction).""" resolved_path = self._resolve_path(path) if resolved_path is None or not resolved_path.is_file(): raise FileNotFoundError(f"Document not found: {path}") file_type = FileType.from_extension(resolved_path.suffix) return self._extract_content(resolved_path, file_type, full=True) def read_preview(self, path: str) -> str: """Read a preview of the document content.""" resolved_path = self._resolve_path(path) if resolved_path is None or not resolved_path.is_file(): raise FileNotFoundError(f"Document not found: {path}") file_type = FileType.from_extension(resolved_path.suffix) return self._extract_content(resolved_path, file_type, full=False) def get_directory_structure(self) -> dict[str, Any]: """Get the full directory tree with metadata.""" if not self.docs_root.is_dir(): return {"path": str(self.docs_root), "children": []} return self._build_tree(self.docs_root) def _build_tree(self, path: Path, relative_path: str = "") -> dict[str, Any]: """Recursively build directory tree.""" result: dict[str, Any] = { "name": path.name if relative_path else str(path), "path": relative_path or ".", "type": "directory", "children": [], } for child in sorted(path.iterdir()): if self._is_excluded(child): continue if child.is_dir(): child_path = str(child.relative_to(self.docs_root)) result["children"].append(self._build_tree(child, child_path)) elif child.is_file(): entry = self._build_entry(child) if entry is not None: result["children"].append( { "name": child.name, "path": str(child.relative_to(self.docs_root)), "type": "file", "size_bytes": child.stat().st_size, "file_type": entry.file_type.value, "modified_time": entry.modified_time, } ) return result def _build_entry(self, path: Path) -> DocumentationEntry | None: """Build a DocumentationEntry from a file path.""" try: stat = path.stat() file_type = FileType.from_extension(path.suffix) supported = self._is_supported(file_type) # Base metadata name = path.name rel_path = str(path.relative_to(self.docs_root)) size_bytes = stat.st_size modified_time = datetime.fromtimestamp(stat.st_mtime).isoformat() entry_kwargs: dict[str, Any] = { "name": name, "path": rel_path, "size_bytes": size_bytes, "file_type": file_type, "modified_time": modified_time, "supported": supported, } # Skip content extraction for unsupported or large files if not supported or size_bytes > self.max_file_size_bytes: return DocumentationEntry(**entry_kwargs) # pyright: ignore[arg-type] # Extract content preview and metrics try: content = self._extract_content(path, file_type, full=False) if content: entry_kwargs["content_preview"] = content[: self.max_preview_chars] if file_type.is_editable: full_content = self._extract_content(path, file_type, full=True) if full_content: entry_kwargs["word_count"] = len(full_content.split()) entry_kwargs["line_count"] = full_content.count("\n") + 1 # Sheet names for Excel if ( file_type == FileType.XLSX and has_openpyxl and load_workbook is not None ): try: wb = load_workbook(path, read_only=True, data_only=True) entry_kwargs["sheet_names"] = tuple(wb.sheetnames) wb.close() except Exception: pass except Exception: pass return DocumentationEntry(**entry_kwargs) # pyright: ignore[arg-type] except Exception: return None def _extract_content( self, path: Path, file_type: FileType, full: bool = False ) -> str: """Extract text content from a file based on its type.""" if file_type == FileType.MARKDOWN or file_type == FileType.TEXT: return path.read_text(encoding="utf-8", errors="replace") if file_type == FileType.CSV: return self._extract_csv_content(path, full) if file_type == FileType.JSON: return self._extract_json_content(path, full) if file_type == FileType.DOCX and has_docx and DocxDocument is not None: return self._extract_docx_content(path, full) if file_type == FileType.XLSX and has_openpyxl and load_workbook is not None: return self._extract_xlsx_content(path, full) if file_type == FileType.PDF and has_pypdf and pypdf is not None: return self._extract_pdf_content(path, full) return f"Content extraction not available for {file_type.value} files." def _extract_csv_content(self, path: Path, full: bool) -> str: """Extract CSV content as readable text.""" try: lines: list[str] = [] with open(path, encoding="utf-8", errors="replace") as f: reader = csv.reader(f) if not full: for i, row in enumerate(reader): if i >= 10: lines.append("... (truncated)") break lines.append(" | ".join(row)) else: for row in reader: lines.append(" | ".join(row)) return "\n".join(lines) except Exception: return f"[Could not parse CSV: {path.name}]" def _extract_json_content(self, path: Path, full: bool) -> str: """Extract JSON content as formatted text.""" try: data = json.loads(path.read_text(encoding="utf-8", errors="replace")) if not full: if isinstance(data, list): # JSON data has arbitrary structure that Pylance cannot infer. preview = data[:2] # pyright: ignore if len(data) > 2: # pyright: ignore preview.append("...") # pyright: ignore return json.dumps(preview, indent=2, ensure_ascii=False) elif isinstance(data, dict): items = list(data.items()) # pyright: ignore preview_dict = dict(items[:5]) # pyright: ignore if len(items) > 5: # pyright: ignore preview_dict["..."] = ( f"({len(items) - 5} more keys)" # pyright: ignore ) return json.dumps(preview_dict, indent=2, ensure_ascii=False) return json.dumps(data, indent=2, ensure_ascii=False) except Exception: return f"[Could not parse JSON: {path.name}]" def _extract_docx_content(self, path: Path, full: bool) -> str: """Extract text from DOCX file.""" try: if DocxDocument is None: return "[python-docx not available]" doc = DocxDocument(str(path)) paragraphs: list[str] = [] for p in doc.paragraphs: if p.text.strip(): paragraphs.append(p.text) if not full: paragraphs = paragraphs[:20] if len(doc.paragraphs) > 20: paragraphs.append("... (truncated)") return "\n".join(paragraphs) except Exception: return f"[Could not read DOCX: {path.name}]" def _extract_xlsx_content(self, path: Path, full: bool) -> str: """Extract text from XLSX file.""" try: if load_workbook is None: return "[openpyxl not available]" wb = load_workbook(path, read_only=True, data_only=True) lines: list[str] = [] for sheet_name in wb.sheetnames: sheet = wb[sheet_name] lines.append(f"\n=== Sheet: {sheet_name} ===\n") max_rows = 50 if not full else 1000 row_count = 0 for row in sheet.iter_rows(values_only=True): if row_count >= max_rows: lines.append("... (truncated)") break row_str = " | ".join( str(cell) if cell is not None else "" for cell in row ) if row_str.strip(): lines.append(row_str) row_count += 1 wb.close() return "\n".join(lines) except Exception: return f"[Could not read XLSX: {path.name}]" def _extract_pdf_content(self, path: Path, full: bool) -> str: """Extract text from PDF file.""" try: if pypdf is None: return "[pypdf not available]" text_parts: list[str] = [] with open(path, "rb") as f: pdf = pypdf.PdfReader(f) max_pages = 5 if not full else len(pdf.pages) for i in range(min(max_pages, len(pdf.pages))): page = pdf.pages[i] page_text = page.extract_text() if page_text: text_parts.append(page_text) if not full and len(pdf.pages) > 5: text_parts.append("... (truncated)") return "\n".join(text_parts) except Exception: return f"[Could not read PDF: {path.name}]" def _resolve_path(self, path: str) -> Path | None: """Resolve a path relative to docs_root with security checks.""" # Basic security: prevent path traversal if ".." in path or path.startswith("/") or path.startswith("\\"): return None try: full_path = (self.docs_root / path).resolve() except ValueError: return None # Ensure path is within docs_root try: full_path.relative_to(self.docs_root) except ValueError: return None return full_path def _is_excluded(self, path: Path) -> bool: """Check if a path should be excluded.""" if path.name.startswith("."): return True exclude_dirs = {"__pycache__", ".git", ".venv", "node_modules", ".pytest_cache"} if path.name in exclude_dirs: return True return False def _is_supported(self, file_type: FileType) -> bool: """Check if a file type is supported for content extraction.""" if file_type in {FileType.MARKDOWN, FileType.TEXT, FileType.CSV, FileType.JSON}: return True if file_type == FileType.DOCX: return has_docx if file_type == FileType.XLSX: return has_openpyxl if file_type == FileType.PDF: return has_pypdf return False @lru_cache(maxsize=128) def _get_cached_list(self, recursive: bool) -> tuple[DocumentationEntry, ...]: """Cached version of list_documents for performance.""" return self.list_documents(recursive) def invalidate_cache(self) -> None: """Invalidate the internal cache.""" self._get_cached_list.cache_clear() def search_documents( self, query: str, file_types: list[FileType] | None = None ) -> tuple[DocumentationEntry, ...]: """Search documents by filename (basic search).""" query_lower = query.lower() results: list[DocumentationEntry] = [] for entry in self.list_documents(recursive=True): if file_types is not None and entry.file_type not in file_types: continue if query_lower in entry.name.lower(): results.append(entry) return tuple(results) def create_docs_source(docs_root: str | Path) -> DocumentationSource: """Factory function for creating a DocumentationSource.""" if isinstance(docs_root, str): docs_root = Path(docs_root) return DocumentationSource(docs_root)