Manages documentation cache.
| 14 | |
| 15 | |
| 16 | class CacheManager: |
| 17 | """Manages documentation cache.""" |
| 18 | |
| 19 | def __init__(self, cache_dir: str = None, cache_expiry_days: int = None): |
| 20 | self.cache_dir = Path(cache_dir or WebAppConfig.CACHE_DIR) |
| 21 | self.cache_expiry_days = cache_expiry_days or WebAppConfig.CACHE_EXPIRY_DAYS |
| 22 | self.cache_dir.mkdir(parents=True, exist_ok=True) |
| 23 | self.cache_index: Dict[str, CacheEntry] = {} |
| 24 | self.load_cache_index() |
| 25 | |
| 26 | def load_cache_index(self): |
| 27 | """Load cache index from disk.""" |
| 28 | index_file = self.cache_dir / "cache_index.json" |
| 29 | if index_file.exists(): |
| 30 | try: |
| 31 | data = file_manager.load_json(index_file) |
| 32 | for key, value in data.items(): |
| 33 | self.cache_index[key] = CacheEntry( |
| 34 | repo_url=value['repo_url'], |
| 35 | repo_url_hash=value['repo_url_hash'], |
| 36 | docs_path=value['docs_path'], |
| 37 | created_at=datetime.fromisoformat(value['created_at']), |
| 38 | last_accessed=datetime.fromisoformat(value['last_accessed']) |
| 39 | ) |
| 40 | except Exception as e: |
| 41 | print(f"Error loading cache index: {e}") |
| 42 | |
| 43 | def save_cache_index(self): |
| 44 | """Save cache index to disk.""" |
| 45 | index_file = self.cache_dir / "cache_index.json" |
| 46 | try: |
| 47 | data = {} |
| 48 | for key, entry in self.cache_index.items(): |
| 49 | data[key] = { |
| 50 | 'repo_url': entry.repo_url, |
| 51 | 'repo_url_hash': entry.repo_url_hash, |
| 52 | 'docs_path': entry.docs_path, |
| 53 | 'created_at': entry.created_at.isoformat(), |
| 54 | 'last_accessed': entry.last_accessed.isoformat() |
| 55 | } |
| 56 | |
| 57 | file_manager.save_json(data, index_file) |
| 58 | except Exception as e: |
| 59 | print(f"Error saving cache index: {e}") |
| 60 | |
| 61 | def get_repo_hash(self, repo_url: str) -> str: |
| 62 | """Generate hash for repository URL.""" |
| 63 | return hashlib.sha256(repo_url.encode()).hexdigest()[:16] |
| 64 | |
| 65 | def get_cached_docs(self, repo_url: str) -> Optional[str]: |
| 66 | """Get cached documentation path if available.""" |
| 67 | repo_hash = self.get_repo_hash(repo_url) |
| 68 | |
| 69 | if repo_hash in self.cache_index: |
| 70 | entry = self.cache_index[repo_hash] |
| 71 | |
| 72 | # Check if cache is still valid |
| 73 | if datetime.now() - entry.created_at < timedelta(days=self.cache_expiry_days): |