Spaces:
Running
Running
File size: 6,337 Bytes
a96145c 3c9e07f a96145c 9875b31 e697eec a96145c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 | import logging
import time
from email.utils import parsedate_to_datetime
from urllib.parse import urlparse, urlencode, parse_qs
import feedparser
import requests
from .models import RedditPost
logger = logging.getLogger(__name__)
TRACKING_PARAMS = {"at_medium", "at_campaign", "ref", "utm_source", "utm_medium", "utm_campaign"}
DEFAULT_FEEDS = [
# Geopolitical
"https://feeds.bbci.co.uk/news/world/rss.xml",
"https://rss.nytimes.com/services/xml/rss/nyt/World.xml",
"https://rss.nytimes.com/services/xml/rss/nyt/Politics.xml",
"https://feeds.npr.org/1001/rss.xml",
"https://www.aljazeera.com/xml/rss/all.xml",
"https://www.theguardian.com/world/rss",
# World Health
"https://www.statnews.com/feed/",
"https://www.sciencedaily.com/rss/all.xml",
"https://www.sciencenews.org/feed",
"https://www.medscape.com/cx/rssfeeds/2700.xml",
"https://www.nih.gov/news-events/news-releases/rss.xml",
"https://news.harvard.edu/gazette/feed/",
"https://news.harvard.edu/gazette/section/health-medicine/feed/",
# Tech
"https://feeds.bbci.co.uk/news/technology/rss.xml",
"https://rss.nytimes.com/services/xml/rss/nyt/Technology.xml",
"https://techcrunch.com/feed/",
"https://www.wired.com/feed/rss",
"https://www.theverge.com/rss/index.xml",
"https://arstechnica.com/feed/",
# Cybersecurity
"https://feeds.feedburner.com/TheHackerNews",
"https://krebsonsecurity.com/feed/",
"https://www.bleepingcomputer.com/feed/",
"https://threatpost.com/feed/",
"https://therecord.media/feed/",
# Funny / Weird
"https://www.theonion.com/rss",
"https://www.reddit.com/r/nottheonion/.rss",
"https://www.thedailymash.co.uk/feed",
"https://babylonbee.com/feed",
# Gaming
"https://feeds.ign.com/ign/all",
"https://www.eurogamer.net/feed",
"https://www.pcgamer.com/rss/",
"https://www.kotaku.com/rss",
"https://www.gamespot.com/feeds/news/",
"https://www.polygon.com/rss/index.xml",
# Movies
"https://variety.com/feed/",
"https://www.hollywoodreporter.com/feed/",
"https://deadline.com/feed/",
"https://screenrant.com/feed/",
# Arab World
"https://www.arabnews.com/rss.xml",
"https://www.middleeasteye.net/rss",
"https://www.newarab.com/rss.xml",
"https://www.france24.com/en/middle-east/rss",
# Tunisia
"https://www.tunisiaonlinenews.com/feed/",
"https://northafricapost.com/feed/",
"https://www.africanews.com/feed/",
"https://nawaat.org/feed/",
"https://www.tunisienumerique.com/feed-actualites-tunisie.xml",
"https://lapresse.tn/feed/",
"https://allafrica.com/tools/headlines/rdf/tunisia/headlines.rdf",
]
class RSSFeedScraper:
def __init__(self, config):
self.config = config
self.feeds = getattr(config, "rss_feeds", DEFAULT_FEEDS)
self.session = requests.Session()
self.session.headers.update({
"User-Agent": "Mozilla/5.0 (compatible; newsapp/1.0)",
})
def fetch_posts(self) -> list[RedditPost]:
seen_titles = set()
posts = []
for url in self.feeds:
logger.info("Fetching RSS: %s", url)
try:
if "reddit.com" in url:
time.sleep(2.5)
resp = self.session.get(url, timeout=15)
resp.raise_for_status()
feed = feedparser.parse(resp.content)
for entry in feed.entries[: self.config.posts_per_subreddit]:
post = self._entry_to_post(entry, url)
if post and post.title not in seen_titles:
seen_titles.add(post.title)
posts.append(post)
except Exception as exc:
logger.warning("Failed RSS %s: %s", url, exc)
time.sleep(1.0)
return posts
@staticmethod
def _clean_url(url: str) -> str:
parsed = urlparse(url)
if not parsed.query:
return url
params = parse_qs(parsed.query)
clean = {k: v[0] for k, v in params.items() if k.lower() not in TRACKING_PARAMS}
if not clean:
return f"{parsed.scheme}://{parsed.netloc}{parsed.path}"
return f"{parsed.scheme}://{parsed.netloc}{parsed.path}?{urlencode(clean)}"
@staticmethod
def _extract_image(entry) -> str:
for key in ("media_content", "media_thumbnail"):
items = entry.get(key) or []
for item in items:
url = item.get("url", "")
if url:
return url
for link in entry.get("links", []):
if link.get("rel") == "enclosure" and "image" in link.get("type", ""):
return link.get("href", "")
return ""
def _entry_to_post(self, entry, feed_url: str) -> RedditPost | None:
title = entry.get("title", "")
if not title:
return None
link = self._clean_url(entry.get("link", ""))
domain = urlparse(link).netloc or urlparse(feed_url).netloc
image_url = self._extract_image(entry)
published, published_iso = self._format_date(entry)
return RedditPost(
id=entry.get("id", entry.get("guid", link)),
title=title,
url=link,
subreddit=domain,
score=0, num_comments=0,
source_domain=domain,
image_url=image_url,
published=published,
published_iso=published_iso,
)
@staticmethod
def _format_date(entry) -> tuple[str, str]:
raw = entry.get("published") or entry.get("updated") or ""
if not raw:
return ("", "")
try:
dt = parsedate_to_datetime(raw)
display = dt.strftime("%d %b %Y")
iso = dt.strftime("%Y-%m-%d")
return (display, iso)
except Exception:
if len(raw) >= 10 and raw[4] == "-" and raw[7] == "-":
iso = raw[:10]
from datetime import datetime
try:
dt = datetime.strptime(iso, "%Y-%m-%d")
display = dt.strftime("%d %b %Y")
except Exception:
display = iso
return (display, iso)
return (raw[:16], "")
|