(output_dir: Path)
| 167 | |
| 168 | |
| 169 | def scan_mirror(output_dir: Path): |
| 170 | mirror_pth = mirror.LocalMirror.get_mirror_path() |
| 171 | click.echo("Collecting package names from a mirror") |
| 172 | pkgs = list(x.name for x in Path(mirror_pth / "json").iterdir()) |
| 173 | click.echo(f"Collected {len(pkgs)} packages") |
| 174 | click.echo("Spawning scanning workers") |
| 175 | |
| 176 | with click.progressbar(pkgs) as bar: |
| 177 | for idx, pkg in enumerate(bar): |
| 178 | uri = f"mirror://{pkg}" |
| 179 | |
| 180 | out_pth = output_dir / f"{pkg}.scan_results.json" |
| 181 | |
| 182 | metadata = { |
| 183 | "format": "json", |
| 184 | "output_path": os.fspath(out_pth), |
| 185 | "fork": True |
| 186 | } |
| 187 | scan_uri(uri=uri, metadata=metadata) |
| 188 | # executor.apply_async( |
| 189 | # func=scan_uri, |
| 190 | # kwds={"uri": uri, "metadata": metadata} |
| 191 | # ) |
| 192 | |
| 193 | |
| 194 | def parse_ast(path: Union[str, Path], stages: Optional[Tuple[str,...]]=None, format="text"): |
nothing calls this directly
no test coverage detected