(feed_url: str)
| 178 | } |
| 179 | |
| 180 | def fetch_one_feed(feed_url: str) -> List[Dict]: |
| 181 | try: |
| 182 | request = urllib.request.Request(feed_url, headers={"User-Agent": "PaperFlow Desktop"}) |
| 183 | with urllib.request.urlopen(request, timeout=20) as response: |
| 184 | raw = response.read() |
| 185 | root = ET.fromstring(raw) |
| 186 | except Exception as exc: |
| 187 | print(f" Custom RSS fetch error for {feed_url}: {exc}") |
| 188 | return [] |
| 189 | |
| 190 | items = root.findall(".//item") or root.findall(".//atom:entry", namespaces) |
| 191 | feed_papers: List[Dict] = [] |
| 192 | for item in items[:per_feed_limit]: |
| 193 | title = _strip_html(_xml_text(item, "title", "atom:title")) |
| 194 | link = _xml_text(item, "link") |
| 195 | if not link: |
| 196 | atom_link = item.find("atom:link", namespaces) |
| 197 | link = atom_link.get("href", "") if atom_link is not None else "" |
| 198 | abstract = _strip_html(_xml_text(item, "description", "summary", "atom:summary", "atom:content")) |
| 199 | author = _xml_text(item, "dc:creator", "author", "atom:author/atom:name") |
| 200 | published = _parse_feed_date(_xml_text(item, "pubDate", "published", "updated", "atom:published", "atom:updated")) |
| 201 | if not title: |
| 202 | continue |
| 203 | feed_papers.append( |
| 204 | { |
| 205 | "title": title, |
| 206 | "authors": [author] if author else [], |
| 207 | "abstract": abstract, |
| 208 | "url": link, |
| 209 | "paper_url": link, |
| 210 | "source": "custom_rss", |
| 211 | "venue": feed_url, |
| 212 | "publish_date": published, |
| 213 | "categories": ["custom_rss"], |
| 214 | "metadata": {"feed_url": feed_url}, |
| 215 | } |
| 216 | ) |
| 217 | return feed_papers |
| 218 | |
| 219 | papers: List[Dict] = [] |
| 220 | max_workers = min(_env_positive_int("PAPERFLOW_MAX_CONCURRENCY", default=5), len(urls)) |
nothing calls this directly
no test coverage detected