diff --git a/requirements.txt b/requirements.txt index 2c361251a3..cb3874124d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -12,6 +12,7 @@ Flask-OpenID==1.3.0 Flask-WTF==1.2.1 bleach==6.1.0 emoji==2.9.0 +feedgen==1.0.0 humanize==4.9.0 mistune==2.0.5 pybadges==3.0.1 diff --git a/tests/tests_feeds.py b/tests/tests_feeds.py new file mode 100644 index 0000000000..628bf06422 --- /dev/null +++ b/tests/tests_feeds.py @@ -0,0 +1,72 @@ +import unittest +import xml.etree.ElementTree as ET +from unittest.mock import patch, Mock +from webapp.app import create_app + + +class TestFeeds(unittest.TestCase): + def setUp(self): + self.app = create_app(testing=True) + self.client = self.app.test_client() + + @patch("webapp.feeds.feeds.session.get") + def test_feeds_updates_success(self, mock_get): + """Test successful RSS feed generation using feedgen""" + mock_response = Mock() + mock_response.json.return_value = { + "page": 1, + "size": 2, + "snaps": [ + { + "name": "test-snap", + "title": "Test Snap", + "summary": "A test snap", + "publisher": "Test Publisher", + "license": "MIT", + "version": "1.0.0", + "last_updated": "Thu, 02 Oct 2025 22:07:58 GMT", + "icon": "https://example.com/icon.png", + "snap_id": "test-snap-id-123", + "media": [ + { + "type": "screenshot", + "url": "https://example.com/screenshot.png", + } + ], + } + ], + } + mock_response.raise_for_status.return_value = None + mock_get.return_value = mock_response + + response = self.client.get("/feeds/updates") + self.assertEqual(response.status_code, 200) + self.assertEqual( + response.content_type, "application/rss+xml; charset=utf-8" + ) + content = response.get_data(as_text=True) + self.assertIn("", content) + self.assertIn( + "Snapcraft - recently updated snaps", content + ) + + try: + root = ET.fromstring(content) + except ET.ParseError as e: + self.fail(f"Returned content is not valid XML: {e}") + + self.assertEqual(root.tag, "rss") + channel = root.find("channel") + self.assertIsNotNone(channel, "No element found in RSS feed") + + items = channel.findall("item") + self.assertTrue( + any(item.findtext("title") == "Test Snap" for item in items) + ) + self.assertIn("", content) + self.assertIn("Test Publisher", content) + self.assertIn("https://snapcraft.io/test-snap", content) + + +if __name__ == "__main__": + unittest.main() diff --git a/webapp/app.py b/webapp/app.py index bf8f4c59a1..00082e7764 100644 --- a/webapp/app.py +++ b/webapp/app.py @@ -34,6 +34,7 @@ from webapp.endpoints.validation_sets import validation_sets from webapp.endpoints.invites import invites from webapp.endpoints.settings import settings +from webapp.feeds.feeds import feeds from webapp.config import SENTRY_DSN @@ -81,6 +82,7 @@ def inject_csrf_token(): app.register_blueprint(validation_sets) app.register_blueprint(invites) app.register_blueprint(settings) + app.register_blueprint(feeds) init_docs(app, "/docs") init_blog(app, "/blog") init_tutorials(app, "/tutorials") diff --git a/webapp/feeds/__init__.py b/webapp/feeds/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/webapp/feeds/feeds.py b/webapp/feeds/feeds.py new file mode 100644 index 0000000000..cbe6c7bcc7 --- /dev/null +++ b/webapp/feeds/feeds.py @@ -0,0 +1,139 @@ +import flask +import requests +from datetime import datetime, timezone +from flask import Response +from feedgen.feed import FeedGenerator +from requests import Session +from html import escape +from urllib.parse import urlparse + +feeds = flask.Blueprint( + "feeds", + __name__, +) + +session = Session() + + +def is_safe_url(url): + """Check if URL is safe (http/https only).""" + if not url: + return False + try: + parsed = urlparse(url) + return parsed.scheme in ("http", "https") and parsed.netloc + except Exception: + return False + + +def parse_snap_date(date_string): + """Parse date string from API to datetime object.""" + try: + dt = datetime.strptime(date_string, "%a, %d %b %Y %H:%M:%S %Z") + return dt.replace(tzinfo=timezone.utc) + except ValueError: + return datetime.now(timezone.utc) + + +def create_snap_description(snap): + """Create HTML description for RSS item.""" + description_parts = [] + + if snap.get("icon") and is_safe_url(snap["icon"]): + icon_url = escape(snap["icon"]) + description_parts.append( + f'Snap icon' + ) + + if snap.get("summary"): + summary = escape(snap["summary"]) + description_parts.append(f"

{summary}

") + + additional_info = [] + + if snap.get("publisher"): + publisher = escape(snap["publisher"]) + additional_info.append(f"
  • Developer: {publisher}
    • ") + + if snap.get("version"): + version = escape(snap["version"]) + additional_info.append(f"
  • Version: {version}
    • ") + + if additional_info: + description_parts.append("") + + if snap.get("media"): + for media in snap["media"]: + if ( + media.get("type") == "screenshot" + and media.get("url") + and is_safe_url(media["url"]) + ): + media_url = escape(media["url"]) + description_parts.append(f'') + + return "".join(description_parts) + + +@feeds.route("/feeds/updates") +def recently_updated_feed(): + """Generate RSS feed for recently updated snaps.""" + + fg = FeedGenerator() + fg.title("Snapcraft - recently updated snaps") + fg.link(href="https://snapcraft.io/store", rel="alternate") + fg.description("Recently updated snaps published on Snapcraft") + fg.language("en") + fg.docs("http://www.rssboard.org/rss-specification") + fg.generator("python-feedgen") + + size = int(flask.request.args.get("size", "50")) + + page = int(flask.request.args.get("page", "1")) + + try: + api_url = flask.current_app.config.get( + "RECOMMENDATION_API_URL", + "https://recommendations.snapcraft.io/api/recently-updated", + ) + params = {"size": size, "page": page} + response = session.get(api_url, params=params, timeout=10) + response.raise_for_status() + + data = response.json() + snaps = data.get("snaps", []) + + except (requests.RequestException, ValueError) as e: + flask.current_app.logger.error(f"Failed to fetch recommendations: {e}") + snaps = [] + + for snap in snaps: + try: + fe = fg.add_entry() + + title = escape(snap.get("title")) + + fe.title(title) + + snap_name = snap.get("name") + snap_url = f"https://snapcraft.io/{snap_name}" + fe.link(href=snap_url) + + description = create_snap_description(snap) + fe.description(description) + + pub_date = parse_snap_date(snap["last_updated"]) + fe.pubDate(pub_date) + + except Exception as e: + flask.current_app.logger.error( + f"Failed to add snap to RSS feed: {e}" + ) + continue + + rss_str = fg.rss_str(pretty=True) + + response = Response(rss_str, mimetype="application/rss+xml") + response.headers["Cache-Control"] = "public, max-age=86400" + + return response