Coverage for webapp/store/logic.py: 77%
231 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-20 22:09 +0000
1import datetime
2import random
3import re
4from urllib.parse import parse_qs, urlparse
6import humanize
7from dateutil import parser
8from dateutil.relativedelta import relativedelta
9from canonicalwebteam.exceptions import StoreApiError
10from cache.cache_utility import redis_cache
11from webapp import helpers
14def get_n_random_snaps(snaps, choice_number):
15 if len(snaps) > choice_number:
16 return random.sample(snaps, choice_number)
18 return snaps
21def get_publisher_snaps(device_gateway, publisher):
22 """Return a publisher's snaps from the store API, cached per publisher.
24 Uses the v2 "find" endpoint, which only returns currently-listed
25 snaps (unlisted/removed snaps are excluded). The result is cached so
26 we don't fetch the full publisher catalogue on every page view.
27 """
28 cache_key = f"publisher-snaps:{publisher}"
29 snaps = redis_cache.get(cache_key, expected_type=list)
30 if not snaps:
31 try:
32 snaps = device_gateway.find(
33 publisher=publisher,
34 fields=["title", "summary", "media", "publisher"],
35 ).get("results", [])
36 except StoreApiError:
37 snaps = []
38 if snaps:
39 redis_cache.set(cache_key, snaps, ttl=3600)
40 return snaps
43def hydrate_featured_snaps(featured_snaps, snaps_by_name):
44 """Hydrate curated featured snaps with live store API data.
46 'featured_snaps' is the editorial list from a publisher's YAML
47 (package_name, background, description). title/summary/icon come from
48 'snaps_by_name' (built from the API). Snaps missing from the API
49 (unlisted/private/removed) are dropped.
50 """
51 return [
52 {
53 **snaps_by_name[snap["package_name"]],
54 "background": snap.get("background"),
55 "description": snap.get("description"),
56 }
57 for snap in featured_snaps or []
58 if snap["package_name"] in snaps_by_name
59 ]
62def get_snap_banner_url(snap_result):
63 """Get snaps banner url from media object
65 :param snap_result: the snap dictionnary
66 :returns: the snap dict with banner key
67 """
68 for media in snap_result["media"]:
69 if media["type"] == "banner":
70 snap_result["banner_url"] = media["url"]
71 break
73 return snap_result
76def get_pages_details(url, links):
77 """Transform returned navigation links from search API from limit/offset
78 to size/page
80 :param url: The url to build
81 :param links: The links returned by the API
83 :returns: A dictionnary with all the navigation links
84 """
85 links_result = {}
87 if "first" in links:
88 links_result["first"] = convert_navigation_url(
89 url, links["first"]["href"]
90 )
92 if "last" in links:
93 links_result["last"] = convert_navigation_url(
94 url, links["last"]["href"]
95 )
97 if "next" in links:
98 links_result["next"] = convert_navigation_url(
99 url, links["next"]["href"]
100 )
102 if "prev" in links:
103 links_result["prev"] = convert_navigation_url(
104 url, links["prev"]["href"]
105 )
107 if "self" in links:
108 links_result["self"] = convert_navigation_url(
109 url, links["self"]["href"]
110 )
112 return links_result
115def convert_navigation_url(url, link):
116 """Convert navigation link from offest/limit to size/page
118 Example:
119 - input: http://example.com?q=test&category=finance&size=10&page=3
120 - output: http://example2.com?q=test&category=finance&limit=10&offset=30
122 :param url: The new url
123 :param link: The navigation url returned by the API
125 :returns: The new navigation link
126 """
127 url_parsed = urlparse(link)
128 host_url = "{base_url}" "?q={q}&limit={limit}&offset={offset}"
130 url_queries = parse_qs(url_parsed.query)
132 if "q" in url_queries:
133 q = url_queries["q"][0]
134 else:
135 q = ""
137 if "section" in url_queries:
138 category = url_queries["section"][0]
139 else:
140 category = ""
142 size = int(url_queries["size"][0])
143 page = int(url_queries["page"][0])
145 url = host_url.format(
146 base_url=url, q=q, limit=size, offset=size * (page - 1)
147 )
149 if category != "":
150 url += "&category=" + category
152 return url
155def build_pagination_link(snap_searched, snap_category, page):
156 """Build pagination link
158 :param snap_searched: Name of the search query
159 :param snap_category: The category being searched in
160 :param page: The page of results
162 :returns: A url string
163 """
164 params = []
166 if snap_searched:
167 params.append("q=" + snap_searched)
169 if snap_category:
170 params.append("category=" + snap_category)
172 if page:
173 params.append("page=" + str(page))
175 return "/search?" + "&".join(params)
178def convert_channel_maps(channel_map):
179 """Converts channel maps list to format easier to manipulate
181 Example:
182 - Input:
183 [
184 {
185 'architecture': 'arch'
186 'map': [{'info': 'release', ...}, ...],
187 'track': 'track 1'
188 },
189 ...
190 ]
191 - Output:
192 {
193 'arch': {
194 'track 1': [{'info': 'release', ...}, ...],
195 ...
196 },
197 ...
198 }
200 :param channel_maps_list: The channel maps list returned by the API
202 :returns: The channel maps reshaped
203 """
204 channel_map_restruct = {}
206 for channel in channel_map:
207 arch = channel.get("channel").get("architecture")
208 track = channel.get("channel").get("track")
209 if arch not in channel_map_restruct:
210 channel_map_restruct[arch] = {}
211 if track not in channel_map_restruct[arch]:
212 channel_map_restruct[arch][track] = []
214 info = {
215 "released-at": convert_date(channel["channel"].get("released-at")),
216 "version": channel.get("version"),
217 "channel": channel["channel"].get("name"),
218 "risk": channel["channel"].get("risk"),
219 "confinement": channel.get("confinement"),
220 "size": channel["download"].get("size"),
221 "revision": channel["revision"],
222 }
224 channel_map_restruct[arch][track].append(info)
226 return channel_map_restruct
229def convert_date(date_to_convert):
230 """Convert date to human readable format: Month Day Year
232 If date is less than a day return: today or yesterday
234 Format of date to convert: 2019-01-12T16:48:41.821037+00:00
235 Output: Jan 12 2019
237 :param date_to_convert: Date to convert
238 :returns: Readable date
239 """
240 local_timezone = datetime.datetime.utcnow().tzinfo
241 date_parsed = parser.parse(date_to_convert).replace(tzinfo=local_timezone)
242 delta = datetime.datetime.utcnow() - datetime.timedelta(days=1)
244 if delta < date_parsed:
245 return humanize.naturalday(date_parsed).title()
246 else:
247 return date_parsed.strftime("%-d %B %Y")
250def is_snap_old(last_updated_date, old_threshold_years=2.0):
251 """Check if a snap is considered 'old' based on its last update date
253 A snap is considered old if it hasn't been updated in the specified
254 number of years (default: 2 years).
256 :param last_updated_date: The last updated date string in ISO format
257 :param old_threshold_years: Number of years to consider a snap old
258 (default: 2)
259 :returns: True if snap is old, False otherwise
260 """
261 if not last_updated_date:
262 return False
264 try:
265 date_parsed = parser.parse(last_updated_date)
266 if date_parsed.tzinfo is None:
267 date_parsed = date_parsed.replace(tzinfo=datetime.timezone.utc)
269 now = datetime.datetime.now(datetime.timezone.utc)
271 delta = relativedelta(now, date_parsed)
272 years_since_update = delta.years
274 return years_since_update >= old_threshold_years
275 except (ValueError, TypeError):
276 # If we can't parse the date, assume it's not old
277 return False
280categories_list = [
281 "development",
282 "games",
283 "social",
284 "productivity",
285 "utilities",
286 "photo-and-video",
287 "server-and-cloud",
288 "security",
289 "devices-and-iot",
290 "music-and-audio",
291 "entertainment",
292 "art-and-design",
293]
295blacklist = ["featured"]
298def format_category_name(slug):
299 """Format category name into a standard title format
301 :param slug: The hypen spaced, lowercase slug to be formatted
302 :return: The formatted string
303 """
304 return (
305 slug.title()
306 .replace("-", " ")
307 .replace("And", "and")
308 .replace("Iot", "IoT")
309 )
312def get_categories(categories_json):
313 """Retrieve and flatten the nested array from the legacy API response.
315 :param categories_json: The returned json
316 :returns: A list of categories
317 """
319 categories = []
321 if "categories" in categories_json:
322 for cat in categories_json["categories"]:
323 if cat["name"] not in categories_list:
324 if cat["name"] not in blacklist:
325 categories_list.append(cat["name"])
327 for category in categories_list:
328 categories.append(
329 {"slug": category, "name": format_category_name(category)}
330 )
332 return categories
335def get_snap_categories(snap_categories):
336 """Retrieve list of categories with names for a snap.
338 :param snap_categories: List of snap categories from snap info API
339 :returns: A list of categories with names
340 """
341 categories = []
343 for cat in snap_categories:
344 if cat["name"] not in blacklist:
345 categories.append(
346 {
347 "slug": cat["name"],
348 "name": format_category_name(cat["name"]),
349 }
350 )
352 return categories
355def get_latest_versions(
356 channel_maps, default_track, lowest_risk, supported_architectures=None
357):
358 """Get the latest versions of both default/stable and the latest of
359 all other channels, unless it's default/stable
361 :param channel_map: Channel map list
363 :returns: A tuple of default/stable, track/risk channel map objects
364 """
365 ordered_versions = get_last_updated_versions(channel_maps)
367 default_stable = None
368 other = None
369 for channel in ordered_versions:
370 if (
371 not supported_architectures
372 or channel["architecture"] in supported_architectures
373 ):
374 if (
375 channel["track"] == default_track
376 and channel["risk"] == lowest_risk
377 ):
378 if not default_stable:
379 default_stable = channel
380 elif not other:
381 other = channel
383 if default_stable:
384 default_stable["released-at-display"] = convert_date(
385 default_stable["released-at"]
386 )
387 if other:
388 other["released-at-display"] = convert_date(other["released-at"])
389 return default_stable, other
392def get_revisions(channel_maps: list) -> list:
393 """Gets a sorted list of unique revisions
395 :param channel_map: Channel map list
397 :returns: A sorted list of unique revisions
398 """
399 revisions = {channel_map["revision"] for channel_map in channel_maps}
400 return list(reversed(sorted(revisions)))
403def get_last_updated_versions(channel_maps):
404 """Get all channels in order of updates
406 :param channel_map: Channel map list
408 :returns: A list of channels ordered by last updated time
409 """
410 releases = []
411 for channel_map in channel_maps:
412 releases.append(channel_map["channel"])
414 return list(reversed(sorted(releases, key=lambda c: c["released-at"])))
417def get_last_updated_version(channel_maps):
418 """Get the oldest channel that was created
420 :param channel_map: Channel map list
422 :returns: The latest stable version, if no stable, the latest risk updated
423 """
424 newest_channel = None
425 for channel_map in channel_maps:
426 if not newest_channel:
427 newest_channel = channel_map
428 else:
429 if channel_map["channel"]["risk"] == "stable":
430 newest_channel = channel_map
432 if newest_channel["channel"]["risk"] == "stable":
433 break
435 return newest_channel
438def has_stable(channel_maps_list):
439 """Use the channel map to find out if the snap has a stable release
441 :param channel_maps_list: Channel map list
443 :returns: True or False
444 """
445 if channel_maps_list:
446 for arch in channel_maps_list:
447 for track in channel_maps_list[arch]:
448 for release in channel_maps_list[arch][track]:
449 if release["risk"] == "stable":
450 return True
452 return False
455def get_lowest_available_risk(channel_map, track):
456 """Get the lowest available risk for the default track
458 :param channel_map: Channel map list
459 :param track: The track of the channel
461 :returns: The lowest available risk
462 """
463 risk_order = ["stable", "candidate", "beta", "edge"]
464 lowest_available_risk = None
465 for arch in channel_map:
466 if arch in channel_map and track in channel_map[arch]:
467 releases = channel_map[arch][track]
468 for release in releases:
469 if not lowest_available_risk:
470 lowest_available_risk = release["risk"]
471 else:
472 risk_index = risk_order.index(release["risk"])
473 lowest_index = risk_order.index(lowest_available_risk)
474 if risk_index < lowest_index:
475 lowest_available_risk = release["risk"]
477 return lowest_available_risk
480def get_default_architecture(architectures):
481 """Pick the default architecture from a list of architectures.
483 Prefer ``amd64`` when published, otherwise the first architecture
484 sorted. This is the single source of truth for the default-architecture
485 preference, shared by the provenance badge endpoint and the security tab.
487 :param architectures: Iterable of architecture names
489 :returns: The default architecture, or ``None`` if the list is empty
490 """
491 architectures = list(architectures)
492 if not architectures:
493 return None
494 if "amd64" in architectures:
495 return "amd64"
496 return sorted(architectures)[0]
499def extract_info_channel_map(channel_map, track, risk):
500 """Get the confinement and version for a channel
502 :param channel_map: Channel map list
503 :param track: The track of the channel
504 :param risk: The risk of the channel
506 :returns: Dict containing confinement and version
507 """
508 context = {
509 "confinement": None,
510 "version": None,
511 }
513 for arch in channel_map:
514 if track in channel_map[arch]:
515 releases = channel_map[arch][track]
516 for release in releases:
517 if release["risk"] == risk:
518 context["confinement"] = release.get("confinement")
519 context["version"] = release.get("version")
521 return context
523 return context
526def get_video_embed_code(url):
527 """Get the embed code for videos
529 :param url: The url of the video
531 :returns: Embed code
532 """
533 if "youtube" in url:
534 return {
535 "type": "youtube",
536 "url": url.replace("watch?v=", "embed/"),
537 "id": url.rsplit("?v=", 1)[-1],
538 }
539 if "youtu.be" in url:
540 return {
541 "type": "youtube",
542 "url": url.replace("youtu.be/", "youtube.com/embed/"),
543 "id": url.rsplit("/", 1)[-1],
544 }
545 if "vimeo" in url:
546 return {
547 "type": "vimeo",
548 "url": url.replace("vimeo.com/", "player.vimeo.com/video/"),
549 "id": url.rsplit("/", 1)[-1],
550 }
551 if "asciinema" in url:
552 return {
553 "type": "asciinema",
554 "url": url + ".js",
555 "id": url.rsplit("/", 1)[-1],
556 }
559def filter_screenshots(media):
560 banner_regex = r"/banner(\-icon)?(_.*)?\.(png|jpg)"
562 return [
563 m
564 for m in media
565 if m["type"] == "screenshot" and not re.search(banner_regex, m["url"])
566 ][:5]
569def get_video(media):
570 video = None
571 for m in media:
572 if m["type"] == "video":
573 video = get_video_embed_code(m["url"])
574 break
575 return video
578def promote_snap_with_icon(snaps):
579 """Move the first snap with an icon to the front of the list
581 :param snaps: The list of snaps
583 :returns: A list of snaps
584 """
585 try:
586 snap_with_icon = next(snap for snap in snaps if snap["icon_url"] != "")
588 if snap_with_icon:
589 snap_with_icon_index = snaps.index(snap_with_icon)
591 snaps.insert(0, snaps.pop(snap_with_icon_index))
592 except StopIteration:
593 pass
595 return snaps
598def get_snap_developer(snap_name):
599 """Is this a special snap published by Canonical?
600 Show some developer information
602 :param snap_name: The name of a snap
604 :returns: a list of [display_name, url]
606 """
607 filename = "store/content/developers/snaps.yaml"
608 snaps = helpers.get_yaml(filename, typ="rt")
610 if snaps and snap_name in snaps:
611 return snaps[snap_name]
613 return None