(self, source_id: str, source: dict)
| 2331 | self.seed_mode = seed_mode |
| 2332 | |
| 2333 | def check(self, source_id: str, source: dict) -> Optional[dict]: |
| 2334 | url = source["url"] |
| 2335 | prev = self.state.get(source_id, {}) |
| 2336 | prev_titles = set(prev.get("seen_titles", [])) |
| 2337 | title_filter = source.get("title_filter") |
| 2338 | |
| 2339 | try: |
| 2340 | resp = self.session.get(url, timeout=30) |
| 2341 | resp.raise_for_status() |
| 2342 | except Exception as e: |
| 2343 | print(f" ERROR MFDS: {e}") |
| 2344 | return None |
| 2345 | |
| 2346 | base_url = self._detect_base_url(url) |
| 2347 | entries = self._parse_news(resp.text, base_url) |
| 2348 | if not entries: |
| 2349 | print(f" WARNING: No entries from MFDS page ({url})") |
| 2350 | return None |
| 2351 | |
| 2352 | print(f" INFO: Parsed {len(entries)} entries from MFDS page") |
| 2353 | |
| 2354 | if title_filter: |
| 2355 | before = len(entries) |
| 2356 | entries = [e for e in entries |
| 2357 | if re.search(title_filter, e.get("title", ""), re.IGNORECASE)] |
| 2358 | print(f" INFO: {len(entries)} entries after title filter (from {before})") |
| 2359 | |
| 2360 | new_items = [] |
| 2361 | all_titles = list(prev_titles) |
| 2362 | |
| 2363 | for entry in entries: |
| 2364 | title = entry.get("title", "") |
| 2365 | if not title or title in prev_titles: |
| 2366 | continue |
| 2367 | all_titles.append(title) |
| 2368 | new_items.append(entry) |
| 2369 | |
| 2370 | self.state[source_id] = { |
| 2371 | "url": url, |
| 2372 | "last_checked": datetime.now().isoformat(), |
| 2373 | "seen_titles": all_titles[-300:], |
| 2374 | } |
| 2375 | |
| 2376 | if new_items and (prev_titles or self.seed_mode): |
| 2377 | if self.seed_mode and not prev_titles: |
| 2378 | new_items = new_items[:10] |
| 2379 | print(f" INFO: Seed mode -- returning top {len(new_items)} entries as initial news") |
| 2380 | result = _make_update( |
| 2381 | source_id, source, "mfds_page", |
| 2382 | f"{len(new_items)} new MFDS update(s) detected" |
| 2383 | ) |
| 2384 | result["new_items"] = new_items |
| 2385 | return result |
| 2386 | elif not prev_titles: |
| 2387 | print(f" INFO: Baseline established ({len(entries)} MFDS entries)") |
| 2388 | |
| 2389 | return None |
| 2390 |
nothing calls this directly
no test coverage detected