Coverage for webapp / store / views.py: 53%

1from math import floor 

2 

3import requests as api_requests 

4import flask 

5from dateutil import parser 

6from webapp.decorators import exchange_required, login_required 

7import webapp.helpers as helpers 

8import webapp.store.logic as logic 

9from webapp.api import requests 

10 

11from canonicalwebteam.exceptions import StoreApiError 

12from canonicalwebteam.store_api.dashboard import Dashboard 

13from canonicalwebteam.store_api.publishergw import PublisherGW 

14from canonicalwebteam.store_api.devicegw import DeviceGW 

15from canonicalwebteam.snap_recommendations import SnapRecommendations 

16 

17from webapp.api.exceptions import ApiError 

18from webapp.store.snap_details_views import snap_details_views 

19from webapp.helpers import api_publisher_session, api_session 

20from flask.json import jsonify 

21import os 

22from webapp.extensions import csrf 

23from webapp.store.logic import ( 

24 get_categories, 

25) 

26from cache.cache_utility import redis_cache 

27 

28session = requests.Session() 

29 

30YOUTUBE_API_KEY = os.getenv("YOUTUBE_API_KEY") 

31dashboard = Dashboard(api_session) 

32publisher_gateway = PublisherGW("snap", api_publisher_session) 

33device_gateway = DeviceGW("snap", api_session) 

34snap_recommendations = SnapRecommendations(session) 

35 

36 

37def store_blueprint(store_query=None): 

38 store = flask.Blueprint( 

39 "store", 

40 __name__, 

41 template_folder="/templates", 

42 static_folder="/static", 

43 ) 

44 snap_details_views(store) 

45 

46 @store.route("/validation-sets", defaults={"path": ""}) 

47 @store.route("/validation-sets/<path:path>") 

48 @login_required 

49 def validation_sets(path): 

50 return flask.render_template("store/publisher.html") 

51 

52 @store.route("/discover") 

53 def discover(): 

54 return flask.redirect(flask.url_for(".homepage")) 

55 

56 def brand_store_view(): 

57 error_info = {} 

58 status_code = 200 

59 

60 try: 

61 snaps = device_gateway.get_all_items(size=16)["results"] 

62 except (StoreApiError, ApiError): 

63 snaps = [] 

64 

65 for snap in snaps: 

66 if "media" in snap: 

67 snap["icon_url"] = helpers.get_icon(snap["media"]) 

68 

69 return ( 

70 flask.render_template( 

71 "brand-store/store.html", snaps=snaps, error_info=error_info 

72 ), 

73 status_code, 

74 ) 

75 

76 def brand_search_snap(): 

77 status_code = 200 

78 snap_searched = flask.request.args.get("q", default="", type=str) 

79 

80 if not snap_searched: 

81 return flask.redirect(flask.url_for(".homepage")) 

82 

83 size = flask.request.args.get("limit", default=25, type=int) 

84 offset = flask.request.args.get("offset", default=0, type=int) 

85 

86 try: 

87 page = floor(offset / size) + 1 

88 except ZeroDivisionError: 

89 size = 10 

90 page = floor(offset / size) + 1 

91 

92 error_info = {} 

93 searched_results = [] 

94 

95 searched_results = device_gateway.search( 

96 snap_searched, size=size, page=page 

97 ) 

98 

99 snaps_results = searched_results["results"] 

100 

101 for snap in snaps_results: 

102 snap["icon_url"] = helpers.get_icon(snap["media"]) 

103 

104 links = logic.get_pages_details( 

105 flask.request.base_url, 

106 ( 

107 searched_results["_links"] 

108 if "_links" in searched_results 

109 else [] 

110 ), 

111 ) 

112 

113 context = { 

114 "query": snap_searched, 

115 "snaps": snaps_results, 

116 "links": links, 

117 "error_info": error_info, 

118 } 

119 

120 return ( 

121 flask.render_template("brand-store/search.html", **context), 

122 status_code, 

123 ) 

124 

125 @store.route("/store") 

126 def store_view(): 

127 return flask.render_template("store/store.html") 

128 

129 @store.route("/explore") 

130 def explore_view(): 

131 try: 

132 popular_snaps = redis_cache.get( 

133 "explore:popular-snaps", expected_type=list 

134 ) 

135 if not popular_snaps: 

136 popular_snaps = snap_recommendations.get_popular() 

137 redis_cache.set( 

138 "explore:popular-snaps", popular_snaps, ttl=3600 

139 ) 

140 except api_requests.exceptions.RequestException: 

141 popular_snaps = [] 

142 

143 try: 

144 recent_snaps = redis_cache.get( 

145 "explore:recent-snaps", expected_type=list 

146 ) 

147 if not recent_snaps: 

148 recent_snaps = snap_recommendations.get_recent() 

149 redis_cache.set("explore:recent-snaps", recent_snaps, ttl=3600) 

150 except api_requests.exceptions.RequestException: 

151 recent_snaps = [] 

152 

153 try: 

154 trending_snaps = redis_cache.get( 

155 "explore:trending-snaps", expected_type=list 

156 ) 

157 if not trending_snaps: 

158 trending_snaps = snap_recommendations.get_trending() 

159 redis_cache.set( 

160 "explore:trending-snaps", trending_snaps, ttl=3600 

161 ) 

162 except api_requests.exceptions.RequestException: 

163 trending_snaps = [] 

164 

165 try: 

166 top_rated_snaps = redis_cache.get( 

167 "explore:top-rated-snaps", expected_type=list 

168 ) 

169 if not top_rated_snaps: 

170 top_rated_snaps = snap_recommendations.get_top_rated() 

171 redis_cache.set( 

172 "explore:top-rated-snaps", top_rated_snaps, ttl=3600 

173 ) 

174 except api_requests.exceptions.RequestException: 

175 top_rated_snaps = [] 

176 

177 try: 

178 categories_results = redis_cache.get( 

179 "explore:categories", expected_type=list 

180 ) 

181 if not categories_results: 

182 categories_results = device_gateway.get_categories() 

183 redis_cache.set( 

184 "explore:categories", categories_results, ttl=3600 

185 ) 

186 except StoreApiError: 

187 categories_results = [] 

188 

189 categories = sorted( 

190 get_categories(categories_results), 

191 key=lambda category: category["slug"], 

192 ) 

193 

194 return flask.render_template( 

195 "explore/index.html", 

196 categories=categories, 

197 popular_snaps=popular_snaps, 

198 recent_snaps=recent_snaps, 

199 trending_snaps=trending_snaps, 

200 top_rated_snaps=top_rated_snaps, 

201 ) 

202 

203 @store.route("/youtube", methods=["POST"]) 

204 def get_video_thumbnail_data(): 

205 body = flask.request.form 

206 thumbnail_url = "https://www.googleapis.com/youtube/v3/videos" 

207 thumbnail_data = session.get( 

208 ( 

209 f"{thumbnail_url}?id={body['videoId']}" 

210 f"&part=snippet&key={YOUTUBE_API_KEY}" 

211 ) 

212 ) 

213 

214 if thumbnail_data: 

215 return thumbnail_data.json() 

216 

217 return {} 

218 

219 @store.route("/publisher/<regex('[a-z0-9-]*[a-z][a-z0-9-]*'):publisher>") 

220 def publisher_details(publisher): 

221 """ 

222 A view to display the publisher details page for specific publisher. 

223 """ 

224 

225 # 404 for the snap-quarantine publisher 

226 if publisher == "snap-quarantine": 

227 flask.abort(404) 

228 

229 publisher_content_path = flask.current_app.config["CONTENT_DIRECTORY"][ 

230 "PUBLISHER_PAGES" 

231 ] 

232 

233 # special handling for some featured publishers with custom pages 

234 if publisher in ["kde", "snapcrafters", "jetbrains"]: 

235 context = helpers.get_yaml( 

236 publisher_content_path + publisher + ".yaml", typ="safe" 

237 ) 

238 

239 if not context: 

240 flask.abort(404) 

241 

242 popular_snaps = helpers.get_yaml( 

243 publisher_content_path + publisher + "-snaps.yaml", 

244 typ="safe", 

245 ) 

246 

247 context["popular_snaps"] = ( 

248 popular_snaps["snaps"] if popular_snaps else [] 

249 ) 

250 

251 context["snaps"] = [] 

252 snaps_results = [] 

253 try: 

254 snaps_results = device_gateway.find( 

255 publisher=publisher, 

256 fields=[ 

257 "title", 

258 "summary", 

259 "media", 

260 "publisher", 

261 ], 

262 )["results"] 

263 except StoreApiError: 

264 pass # proceed with an empty list 

265 

266 for snap in snaps_results: 

267 item = snap["snap"] 

268 item["package_name"] = snap["name"] 

269 item["icon_url"] = helpers.get_icon(item["media"]) 

270 context["snaps"].append(item) 

271 

272 featured_snaps = [ 

273 snap["package_name"] for snap in context["featured_snaps"] 

274 ] 

275 

276 context["snaps"] = [ 

277 snap 

278 for snap in context["snaps"] 

279 if snap["package_name"] not in featured_snaps 

280 ] 

281 

282 context["snaps_count"] = len(context["snaps"]) + len( 

283 featured_snaps 

284 ) 

285 

286 return flask.render_template( 

287 "store/publisher-details.html", **context 

288 ) 

289 

290 # standard page for all community publishers 

291 status_code = 200 

292 error_info = {} 

293 snaps_results = [] 

294 snaps = [] 

295 snaps_count = 0 

296 publisher_details = {"display-name": publisher, "username": publisher} 

297 

298 snaps_results = device_gateway.find( 

299 publisher=publisher, 

300 fields=[ 

301 "title", 

302 "summary", 

303 "media", 

304 "publisher", 

305 ], 

306 )["results"] 

307 

308 for snap in snaps_results: 

309 item = snap["snap"] 

310 item["package_name"] = snap["name"] 

311 item["icon_url"] = helpers.get_icon(item["media"]) 

312 snaps.append(item) 

313 

314 snaps_count = len(snaps) 

315 

316 if snaps_count > 0: 

317 publisher_details = snaps[0]["publisher"] 

318 

319 context = { 

320 "snaps": snaps, 

321 "snaps_count": snaps_count, 

322 "publisher": publisher_details, 

323 "error_info": error_info, 

324 } 

325 

326 return ( 

327 flask.render_template( 

328 "store/community-publisher-details.html", **context 

329 ), 

330 status_code, 

331 ) 

332 

333 @store.route("/store/categories/<category>") 

334 def store_category(category): 

335 status_code = 200 

336 error_info = {} 

337 snaps_results = [] 

338 

339 snaps_results = device_gateway.get_category_items( 

340 category=category, size=10, page=1 

341 )["results"] 

342 for snap in snaps_results: 

343 snap["icon_url"] = helpers.get_icon(snap["media"]) 

344 

345 # if the first snap (banner snap) doesn't have an icon, remove the last 

346 # snap from the list to avoid a hanging snap (grid of 9) 

347 if len(snaps_results) == 10 and snaps_results[0]["icon_url"] == "": 

348 snaps_results = snaps_results[:-1] 

349 

350 for index in range(len(snaps_results)): 

351 snaps_results[index] = logic.get_snap_banner_url( 

352 snaps_results[index] 

353 ) 

354 

355 context = { 

356 "category": category, 

357 "has_featured": True, 

358 "snaps": snaps_results, 

359 "error_info": error_info, 

360 } 

361 

362 return ( 

363 flask.render_template("store/_category-partial.html", **context), 

364 status_code, 

365 ) 

366 

367 @store.route("/store/featured-snaps/<category>") 

368 def featured_snaps_in_category(category): 

369 snaps_results = [] 

370 

371 snaps_results = device_gateway.get_category_items( 

372 category=category, size=3, page=1 

373 )["_embedded"]["clickindex:package"] 

374 

375 for snap in snaps_results: 

376 snap["icon_url"] = helpers.get_icon(snap["media"]) 

377 

378 return flask.jsonify(snaps_results) 

379 

380 @store.route("/store/sitemap.xml") 

381 def sitemap(): 

382 sitemap = redis_cache.get("sitemap:xml", expected_type=str) 

383 if sitemap: 

384 response = flask.make_response(sitemap) 

385 response.headers["Content-Type"] = "application/xml" 

386 response.headers["Cache-Control"] = "public, max-age=43200" 

387 return response 

388 

389 base_url = "https://snapcraft.io/store" 

390 

391 snaps = [] 

392 page = 0 

393 url = f"https://api.snapcraft.io/api/v1/snaps/search?page={page}" 

394 while url: 

395 response = session.get(url) 

396 try: 

397 snaps_response = response.json() 

398 except Exception: 

399 continue 

400 

401 for snap in snaps_response["_embedded"]["clickindex:package"]: 

402 try: 

403 last_udpated = ( 

404 parser.parse(snap["last_updated"]) 

405 .replace(tzinfo=None) 

406 .strftime("%Y-%m-%d") 

407 ) 

408 snaps.append( 

409 { 

410 "url": "https://snapcraft.io/" 

411 + snap["package_name"], 

412 "last_udpated": last_udpated, 

413 } 

414 ) 

415 except Exception: 

416 continue 

417 if "next" in snaps_response["_links"]: 

418 url = snaps_response["_links"]["next"]["href"] 

419 else: 

420 url = None 

421 

422 xml_sitemap = flask.render_template( 

423 "sitemap/sitemap.xml", 

424 base_url=base_url, 

425 links=snaps, 

426 ) 

427 

428 # Cache the generated sitemap for 12 hours 

429 redis_cache.set("sitemap:xml", xml_sitemap, ttl=43200) 

430 

431 response = flask.make_response(xml_sitemap) 

432 response.headers["Content-Type"] = "application/xml" 

433 response.headers["Cache-Control"] = "public, max-age=43200" 

434 

435 return response 

436 

437 if store_query: 

438 store.add_url_rule("/", "homepage", brand_store_view) 

439 store.add_url_rule("/search", "search", brand_search_snap) 

440 else: 

441 store.add_url_rule("/store", "homepage", store_view) 

442 

443 store.add_url_rule("/explore", "explore", explore_view) 

444 

445 @store.route("/<snap_name>/create-track", methods=["POST"]) 

446 @login_required 

447 @csrf.exempt 

448 @exchange_required 

449 def post_create_track(snap_name): 

450 track_name = flask.request.form["track-name"] 

451 version_pattern = flask.request.form.get("version-pattern") 

452 auto_phasing_percentage = flask.request.form.get( 

453 "automatic-phasing-percentage" 

454 ) 

455 

456 if auto_phasing_percentage is not None: 

457 auto_phasing_percentage = float(auto_phasing_percentage) 

458 

459 response = publisher_gateway.create_track( 

460 flask.session, 

461 snap_name, 

462 track_name, 

463 version_pattern, 

464 auto_phasing_percentage, 

465 ) 

466 if response.status_code == 201: 

467 return response.json(), response.status_code 

468 if response.status_code == 409: 

469 return ( 

470 jsonify({"error": "Track already exists."}), 

471 response.status_code, 

472 ) 

473 if "error-list" in response.json(): 

474 return ( 

475 jsonify( 

476 {"error": response.json()["error-list"][0]["message"]} 

477 ), 

478 response.status_code, 

479 ) 

480 return response.json(), response.status_code 

481 

482 return store