(self, source_id: str, source: dict)
| 1723 | self.seed_mode = seed_mode |
| 1724 | |
| 1725 | def check(self, source_id: str, source: dict) -> Optional[dict]: |
| 1726 | url = source["url"] |
| 1727 | prev = self.state.get(source_id, {}) |
| 1728 | prev_ids = set(prev.get("seen_ids", [])) |
| 1729 | title_filter = source.get("title_filter") |
| 1730 | |
| 1731 | try: |
| 1732 | resp = self.session.get(url, timeout=30) |
| 1733 | resp.raise_for_status() |
| 1734 | except Exception as e: |
| 1735 | print(f" ERROR Atom feed: {e}") |
| 1736 | return None |
| 1737 | |
| 1738 | entries = self._parse_atom(resp.content) |
| 1739 | if not entries: |
| 1740 | print(f" WARNING: No entries parsed from Atom feed") |
| 1741 | return None |
| 1742 | |
| 1743 | print(f" INFO: Parsed {len(entries)} entries from Atom feed") |
| 1744 | |
| 1745 | new_items = [] |
| 1746 | all_ids = list(prev_ids) |
| 1747 | title_exclude = source.get("title_exclude") |
| 1748 | |
| 1749 | for entry in entries: |
| 1750 | eid = entry.get("id", entry.get("link", "")) |
| 1751 | title = entry.get("title", "") |
| 1752 | if not eid or eid in prev_ids: |
| 1753 | continue |
| 1754 | if title_filter and not re.search(title_filter, title, re.IGNORECASE): |
| 1755 | continue |
| 1756 | if title_exclude and re.search(title_exclude, title, re.IGNORECASE): |
| 1757 | print(f" EXCLUDED (medicine/drug): {title[:80]}") |
| 1758 | all_ids.append(eid) |
| 1759 | continue |
| 1760 | all_ids.append(eid) |
| 1761 | new_items.append({ |
| 1762 | "title": title, |
| 1763 | "link": entry.get("link", ""), |
| 1764 | "pub_date": entry.get("updated", "")[:10], |
| 1765 | "description": entry.get("summary", "")[:300], |
| 1766 | }) |
| 1767 | |
| 1768 | self.state[source_id] = { |
| 1769 | "url": url, |
| 1770 | "last_checked": datetime.now().isoformat(), |
| 1771 | "seen_ids": all_ids[-500:], |
| 1772 | } |
| 1773 | |
| 1774 | if new_items and (prev_ids or self.seed_mode): |
| 1775 | if self.seed_mode and not prev_ids: |
| 1776 | new_items = new_items[:10] |
| 1777 | print(f" INFO: Seed mode -- returning top {len(new_items)} entries as initial news") |
| 1778 | result = _make_update( |
| 1779 | source_id, source, "atom_feed", |
| 1780 | f"{len(new_items)} new Atom feed entry(ies) detected" |
| 1781 | ) |
| 1782 | result["new_items"] = new_items |
nothing calls this directly
no test coverage detected