Validates the cache folder and returns the index path.
(path: str, model_path: str | None, content: Sequence[ContentType])
| 108 | |
| 109 | |
| 110 | def get_validated_cache(path: str, model_path: str | None, content: Sequence[ContentType]) -> Path | None: |
| 111 | """Validates the cache folder and returns the index path.""" |
| 112 | index_path = find_index_from_cache_folder(path) |
| 113 | if not index_path.exists(): |
| 114 | return None |
| 115 | |
| 116 | persistence_path = PersistencePath.from_path(index_path) |
| 117 | if persistence_path.non_existing(): |
| 118 | return None |
| 119 | |
| 120 | if model_path is None: |
| 121 | model_path = resolve_model_name() |
| 122 | with open(persistence_path.metadata, encoding="utf-8") as f: |
| 123 | metadata = json.load(f) |
| 124 | if not _metadata_matches(metadata, model_path, content): |
| 125 | return None |
| 126 | |
| 127 | if is_git_url(str(path)): |
| 128 | return index_path |
| 129 | |
| 130 | write_time = metadata["time"] |
| 131 | extensions = get_extensions(content) |
| 132 | |
| 133 | path_as_path = Path(path).resolve() |
| 134 | stored_files: list[str] = metadata.get("file_paths", []) |
| 135 | current_files = [] |
| 136 | for file_path in walk_files(path_as_path, extensions=extensions): |
| 137 | file_status = get_file_status(file_path, write_time) |
| 138 | if file_status == FileStatus.NEWER: |
| 139 | return None |
| 140 | if file_status != FileStatus.VALID: |
| 141 | continue |
| 142 | current_files.append(str(file_path.relative_to(path_as_path))) |
| 143 | |
| 144 | if set(current_files) != set(stored_files): |
| 145 | return None |
| 146 | |
| 147 | return index_path |