(self, advisories)
| 63 | logger.info(f"Finished import for {importer_name}. Imported {count} advisories.") |
| 64 | |
| 65 | def do_import(self, advisories) -> None: |
| 66 | advisory_importer = DefaultImporter(advisories=advisories) |
| 67 | logger.info(f"Running importer: {advisory_importer.qualified_name}") |
| 68 | importer_name = advisory_importer.qualified_name |
| 69 | advisories = [] |
| 70 | for advisory in advisory_importer.interesting_advisories: |
| 71 | if advisory.date_imported: |
| 72 | continue |
| 73 | logger.info(f"Processing advisory: {advisory!r}") |
| 74 | advisory_data = None |
| 75 | inferences = None |
| 76 | try: |
| 77 | advisory_data = advisory.to_advisory_data() |
| 78 | inferences = advisory_importer.get_inferences(advisory_data=advisory_data) |
| 79 | process_inferences( |
| 80 | inferences=inferences, |
| 81 | advisory=advisory, |
| 82 | improver_name=importer_name, |
| 83 | ) |
| 84 | except Exception: |
| 85 | from pprint import pformat |
| 86 | |
| 87 | logger.warning( |
| 88 | f"Failed to process advisory:\n{pformat(advisory_data.to_dict())}\n\n" |
| 89 | f"with error:\n{traceback_format_exc()}\n\n" |
| 90 | ) |
| 91 | logger.info("Finished importing using %s.", advisory_importer.__class__.qualified_name) |
| 92 | |
| 93 | def process_advisories( |
| 94 | self, advisory_datas: Iterable[AdvisoryData], importer_name: str |
no test coverage detected