(self, source_id: str, source: dict)
| 2473 | self.state = state |
| 2474 | |
| 2475 | def check(self, source_id: str, source: dict) -> Optional[dict]: |
| 2476 | url = source["url"] |
| 2477 | prev = self.state.get(source_id, {}) |
| 2478 | prev_titles = set(prev.get("seen_titles", [])) |
| 2479 | title_filter = source.get("title_filter") |
| 2480 | |
| 2481 | try: |
| 2482 | resp = self.session.get(url, timeout=30) |
| 2483 | resp.raise_for_status() |
| 2484 | except Exception as e: |
| 2485 | print(f" ERROR generic page: {e}") |
| 2486 | return None |
| 2487 | |
| 2488 | entries = self._parse_links(resp.text, url) |
| 2489 | if not entries: |
| 2490 | print(f" WARNING: No entries from page {url}") |
| 2491 | return None |
| 2492 | |
| 2493 | print(f" INFO: Parsed {len(entries)} entries from page") |
| 2494 | |
| 2495 | new_items = [] |
| 2496 | all_titles = list(prev_titles) |
| 2497 | |
| 2498 | for entry in entries: |
| 2499 | title = entry.get("title", "") |
| 2500 | if not title or title in prev_titles: |
| 2501 | continue |
| 2502 | if title_filter and not re.search(title_filter, title, re.IGNORECASE): |
| 2503 | continue |
| 2504 | all_titles.append(title) |
| 2505 | new_items.append(entry) |
| 2506 | |
| 2507 | self.state[source_id] = { |
| 2508 | "url": url, |
| 2509 | "last_checked": datetime.now().isoformat(), |
| 2510 | "seen_titles": all_titles[-300:], |
| 2511 | } |
| 2512 | |
| 2513 | if new_items and prev_titles: |
| 2514 | result = _make_update( |
| 2515 | source_id, source, "generic_page", |
| 2516 | f"{len(new_items)} new page item(s) detected" |
| 2517 | ) |
| 2518 | result["new_items"] = new_items |
| 2519 | return result |
| 2520 | elif not prev_titles: |
| 2521 | print(f" INFO: Baseline established ({len(entries)} page entries)") |
| 2522 | |
| 2523 | return None |
| 2524 | |
| 2525 | def _parse_links(self, html: str, base_url: str) -> list[dict]: |
| 2526 | """Extract meaningful links from an HTML page.""" |
nothing calls this directly
no test coverage detected