(self, sources: Optional[List[Dict]] = None, save: bool = True)
| 143 | return screened_sources, report |
| 144 | |
| 145 | def refresh_candidate_pool(self, sources: Optional[List[Dict]] = None, save: bool = True) -> Tuple[List[Dict], Dict]: |
| 146 | source_list = deepcopy(sources if sources is not None else self.load_candidate_sources()) |
| 147 | accepted, rejected, stats = self.policy.screen_sources(source_list) |
| 148 | candidates = self._sort_sources(self._dedupe_by_url(accepted)) |
| 149 | |
| 150 | report = { |
| 151 | "timestamp": datetime.now().isoformat(), |
| 152 | "input": len(source_list), |
| 153 | "candidates": len(candidates), |
| 154 | "rejected": len(rejected), |
| 155 | "reasons": stats.get("reasons", {}), |
| 156 | } |
| 157 | |
| 158 | if save: |
| 159 | self._write_json(self.candidate_file, candidates) |
| 160 | self._write_json(self.candidate_report, report) |
| 161 | |
| 162 | return candidates, report |
| 163 | |
| 164 | def merge_validated_candidates( |
| 165 | self, |
no test coverage detected