(
self,
valid_sources: List[Dict],
invalid_sources: Optional[List[Dict]] = None,
*,
save: bool = True,
)
| 162 | return candidates, report |
| 163 | |
| 164 | def merge_validated_candidates( |
| 165 | self, |
| 166 | valid_sources: List[Dict], |
| 167 | invalid_sources: Optional[List[Dict]] = None, |
| 168 | *, |
| 169 | save: bool = True, |
| 170 | ) -> Tuple[List[Dict], Dict]: |
| 171 | existing = self.load_candidate_sources() |
| 172 | invalid_urls = {str(item.get("bookSourceUrl", "")).strip() for item in (invalid_sources or [])} |
| 173 | |
| 174 | merged = [source for source in existing if str(source.get("bookSourceUrl", "")).strip() not in invalid_urls] |
| 175 | now = datetime.now().isoformat() |
| 176 | for source in valid_sources: |
| 177 | candidate = deepcopy(source) |
| 178 | candidate["_validation_status"] = "valid" |
| 179 | candidate["_last_validated_at"] = now |
| 180 | merged.append(candidate) |
| 181 | |
| 182 | return self.refresh_candidate_pool(merged, save=save) |
| 183 | |
| 184 | def select_screened_validation_batch( |
| 185 | self, |
no test coverage detected