def existing_bugs_history()

in jobs/webcompat-kb/webcompat_kb/bugzilla.py [0:0]


    def existing_bugs_history(self, existing_bugs: BugsById) -> HistoryByBug:
        last_import_time = self.bigquery_last_import()

        if last_import_time is None:
            logging.info("No previous history update found")
            return {}

        updated_bugs = {
            bug_id
            for bug_id, bug in existing_bugs.items()
            if bug.last_change_time > last_import_time
        }

        if not updated_bugs:
            logging.info(f"No updated bugs since {last_import_time.isoformat()}")
            return {}

        logging.info(
            f"Fetching history for bugs {','.join(str(item) for item in updated_bugs)} updated since {last_import_time.isoformat()}"
        )

        bugs_full_history = self.bugzilla_fetch_history(updated_bugs)
        # Filter down to only recent updates, since we always get the full history
        bugs_history = {}
        for bug_id, bug_full_history in bugs_full_history.items():
            bug_history = [
                item for item in bug_full_history if item.change_time > last_import_time
            ]
            if bug_history:
                bugs_history[bug_id] = bug_history

        return bugs_history