(uri, metadata: Union[list, dict]=None, download_only: bool=False)
| 74 | |
| 75 | |
| 76 | def scan_uri(uri, metadata: Union[list, dict]=None, download_only: bool=False) -> List[Detection]: |
| 77 | with utils.enrich_exception(uri, metadata): |
| 78 | start = time.time() |
| 79 | handler = None |
| 80 | metadata = metadata or {} |
| 81 | output_format = metadata.get("format", "text") |
| 82 | all_hits = [] |
| 83 | |
| 84 | if type(output_format) not in (list, tuple): |
| 85 | output_format = (output_format,) |
| 86 | |
| 87 | formatters = [ScanOutputBase.from_uri(x, opts=metadata.get("output_opts")) for x in output_format] |
| 88 | |
| 89 | try: |
| 90 | handler = URIHandler.from_uri(uri) |
| 91 | |
| 92 | if handler is None: |
| 93 | raise ValueError(f"Could not find a handler for provided URI: '{uri}'") |
| 94 | elif not handler.exists: |
| 95 | raise exceptions.InvalidLocation(f"Invalid location provided from URI: '{uri}'") |
| 96 | |
| 97 | metadata.update({ |
| 98 | "name": uri, |
| 99 | "uri_scheme": handler.scheme, |
| 100 | "uri_input": handler.metadata, |
| 101 | "depth": 0 |
| 102 | }) |
| 103 | |
| 104 | # FIXME: metadata=metadata |
| 105 | for x in handler.get_paths(metadata={"analyzers": metadata["analyzers"]}): # type: ScanLocation |
| 106 | if download_only: |
| 107 | continue |
| 108 | else: |
| 109 | all_hits.extend(scan_worker(x)) |
| 110 | |
| 111 | for formatter in formatters: |
| 112 | try: |
| 113 | filtered_hits = formatter.filtered(all_hits) |
| 114 | except exceptions.MinimumScoreNotReached: |
| 115 | pass |
| 116 | else: |
| 117 | with formatter: |
| 118 | formatter.output(hits=filtered_hits, scan_metadata=metadata) |
| 119 | |
| 120 | except exceptions.NoSuchPackage: |
| 121 | logger.warn(f"No such package: {uri}") |
| 122 | except Exception: |
| 123 | logger.exception(f"An error was thrown while processing URI: '{uri}'") |
| 124 | raise |
| 125 | finally: |
| 126 | if handler: |
| 127 | handler.cleanup() |
| 128 | |
| 129 | logger.info(f"Scan finished in {time.time() - start} s") |
| 130 | return all_hits |
| 131 | |
| 132 | |
| 133 | def data_diff(a_path: str, b_path: str, format_uri=("text",), output_opts=None): |
no test coverage detected