Coverage for webapp/feeds/feeds.py: 87%

1import flask 

2import requests 

3from datetime import datetime, timezone 

4from flask import Response 

5from feedgen.feed import FeedGenerator 

6from requests import Session 

7from html import escape 

8from urllib.parse import urlparse 

9 

10feeds = flask.Blueprint( 

11 "feeds", 

12 __name__, 

13) 

14 

15session = Session() 

16 

17 

18def is_safe_url(url): 

19 """Check if URL is safe (http/https only).""" 

20 if not url: 

21 return False 

22 try: 

23 parsed = urlparse(url) 

24 return parsed.scheme in ("http", "https") and parsed.netloc 

25 except Exception: 

26 return False 

27 

28 

29def parse_snap_date(date_string): 

30 """Parse date string from API to datetime object.""" 

31 try: 

32 dt = datetime.strptime(date_string, "%a, %d %b %Y %H:%M:%S %Z") 

33 return dt.replace(tzinfo=timezone.utc) 

34 except ValueError: 

35 return datetime.now(timezone.utc) 

36 

37 

38def create_snap_description(snap): 

39 """Create HTML description for RSS item.""" 

40 description_parts = [] 

41 

42 if snap.get("icon") and is_safe_url(snap["icon"]): 

43 icon_url = escape(snap["icon"]) 

44 description_parts.append( 

45 f'<img src="{icon_url}" alt="Snap icon" width="128" height="128">' 

46 ) 

47 

48 if snap.get("summary"): 

49 summary = escape(snap["summary"]) 

50 description_parts.append(f"<p>{summary}</p>") 

51 

52 additional_info = [] 

53 

54 if snap.get("publisher"): 

55 publisher = escape(snap["publisher"]) 

56 additional_info.append(f"<li>Developer: {publisher}</li>") 

57 

58 if snap.get("version"): 

59 version = escape(snap["version"]) 

60 additional_info.append(f"<li>Version: {version}</li>") 

61 

62 if additional_info: 

63 description_parts.append("<ul>" + "".join(additional_info) + "</ul>") 

64 

65 if snap.get("media"): 

66 for media in snap["media"]: 

67 if ( 

68 media.get("type") == "screenshot" 

69 and media.get("url") 

70 and is_safe_url(media["url"]) 

71 ): 

72 media_url = escape(media["url"]) 

73 description_parts.append(f'<img src="{media_url}" alt="">') 

74 

75 return "".join(description_parts) 

76 

77 

78@feeds.route("/feeds/updates") 

79def recently_updated_feed(): 

80 """Generate RSS feed for recently updated snaps.""" 

81 

82 fg = FeedGenerator() 

83 fg.title("Snapcraft - recently updated snaps") 

84 fg.link(href="https://snapcraft.io/store", rel="alternate") 

85 fg.description("Recently updated snaps published on Snapcraft") 

86 fg.language("en") 

87 fg.docs("http://www.rssboard.org/rss-specification") 

88 fg.generator("python-feedgen") 

89 

90 size = int(flask.request.args.get("size", "50")) 

91 

92 page = int(flask.request.args.get("page", "1")) 

93 

94 try: 

95 api_url = flask.current_app.config.get( 

96 "RECOMMENDATION_API_URL", 

97 "https://recommendations.snapcraft.io/api/recently-updated", 

98 ) 

99 params = {"size": size, "page": page} 

100 response = session.get(api_url, params=params, timeout=10) 

101 response.raise_for_status() 

102 

103 data = response.json() 

104 snaps = data.get("snaps", []) 

105 

106 except (requests.RequestException, ValueError) as e: 

107 flask.current_app.logger.error(f"Failed to fetch recommendations: {e}") 

108 snaps = [] 

109 

110 for snap in snaps: 

111 try: 

112 fe = fg.add_entry() 

113 

114 title = escape(snap.get("title")) 

115 

116 fe.title(title) 

117 

118 snap_name = snap.get("name") 

119 snap_url = f"https://snapcraft.io/{snap_name}" 

120 fe.link(href=snap_url) 

121 

122 description = create_snap_description(snap) 

123 fe.description(description) 

124 

125 pub_date = parse_snap_date(snap["last_updated"]) 

126 fe.pubDate(pub_date) 

127 

128 except Exception as e: 

129 flask.current_app.logger.error( 

130 f"Failed to add snap to RSS feed: {e}" 

131 ) 

132 continue 

133 

134 rss_str = fg.rss_str(pretty=True) 

135 

136 response = Response(rss_str, mimetype="application/rss+xml") 

137 response.headers["Cache-Control"] = "public, max-age=86400" 

138 

139 return response