Coverage for webapp/store/logic.py: 77%
234 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-10 22:08 +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 sboms = None
216 if "sboms" in channel:
217 sboms = channel["sboms"]
219 info = {
220 "released-at": convert_date(channel["channel"].get("released-at")),
221 "version": channel.get("version"),
222 "channel": channel["channel"].get("name"),
223 "risk": channel["channel"].get("risk"),
224 "confinement": channel.get("confinement"),
225 "size": channel["download"].get("size"),
226 "revision": channel["revision"],
227 "sboms": sboms,
228 }
230 channel_map_restruct[arch][track].append(info)
232 return channel_map_restruct
235def convert_date(date_to_convert):
236 """Convert date to human readable format: Month Day Year
238 If date is less than a day return: today or yesterday
240 Format of date to convert: 2019-01-12T16:48:41.821037+00:00
241 Output: Jan 12 2019
243 :param date_to_convert: Date to convert
244 :returns: Readable date
245 """
246 local_timezone = datetime.datetime.utcnow().tzinfo
247 date_parsed = parser.parse(date_to_convert).replace(tzinfo=local_timezone)
248 delta = datetime.datetime.utcnow() - datetime.timedelta(days=1)
250 if delta < date_parsed:
251 return humanize.naturalday(date_parsed).title()
252 else:
253 return date_parsed.strftime("%-d %B %Y")
256def is_snap_old(last_updated_date, old_threshold_years=2.0):
257 """Check if a snap is considered 'old' based on its last update date
259 A snap is considered old if it hasn't been updated in the specified
260 number of years (default: 2 years).
262 :param last_updated_date: The last updated date string in ISO format
263 :param old_threshold_years: Number of years to consider a snap old
264 (default: 2)
265 :returns: True if snap is old, False otherwise
266 """
267 if not last_updated_date:
268 return False
270 try:
271 date_parsed = parser.parse(last_updated_date)
272 if date_parsed.tzinfo is None:
273 date_parsed = date_parsed.replace(tzinfo=datetime.timezone.utc)
275 now = datetime.datetime.now(datetime.timezone.utc)
277 delta = relativedelta(now, date_parsed)
278 years_since_update = delta.years
280 return years_since_update >= old_threshold_years
281 except (ValueError, TypeError):
282 # If we can't parse the date, assume it's not old
283 return False
286categories_list = [
287 "development",
288 "games",
289 "social",
290 "productivity",
291 "utilities",
292 "photo-and-video",
293 "server-and-cloud",
294 "security",
295 "devices-and-iot",
296 "music-and-audio",
297 "entertainment",
298 "art-and-design",
299]
301blacklist = ["featured"]
304def format_category_name(slug):
305 """Format category name into a standard title format
307 :param slug: The hypen spaced, lowercase slug to be formatted
308 :return: The formatted string
309 """
310 return (
311 slug.title()
312 .replace("-", " ")
313 .replace("And", "and")
314 .replace("Iot", "IoT")
315 )
318def get_categories(categories_json):
319 """Retrieve and flatten the nested array from the legacy API response.
321 :param categories_json: The returned json
322 :returns: A list of categories
323 """
325 categories = []
327 if "categories" in categories_json:
328 for cat in categories_json["categories"]:
329 if cat["name"] not in categories_list:
330 if cat["name"] not in blacklist:
331 categories_list.append(cat["name"])
333 for category in categories_list:
334 categories.append(
335 {"slug": category, "name": format_category_name(category)}
336 )
338 return categories
341def get_snap_categories(snap_categories):
342 """Retrieve list of categories with names for a snap.
344 :param snap_categories: List of snap categories from snap info API
345 :returns: A list of categories with names
346 """
347 categories = []
349 for cat in snap_categories:
350 if cat["name"] not in blacklist:
351 categories.append(
352 {
353 "slug": cat["name"],
354 "name": format_category_name(cat["name"]),
355 }
356 )
358 return categories
361def get_latest_versions(
362 channel_maps, default_track, lowest_risk, supported_architectures=None
363):
364 """Get the latest versions of both default/stable and the latest of
365 all other channels, unless it's default/stable
367 :param channel_map: Channel map list
369 :returns: A tuple of default/stable, track/risk channel map objects
370 """
371 ordered_versions = get_last_updated_versions(channel_maps)
373 default_stable = None
374 other = None
375 for channel in ordered_versions:
376 if (
377 not supported_architectures
378 or channel["architecture"] in supported_architectures
379 ):
380 if (
381 channel["track"] == default_track
382 and channel["risk"] == lowest_risk
383 ):
384 if not default_stable:
385 default_stable = channel
386 elif not other:
387 other = channel
389 if default_stable:
390 default_stable["released-at-display"] = convert_date(
391 default_stable["released-at"]
392 )
393 if other:
394 other["released-at-display"] = convert_date(other["released-at"])
395 return default_stable, other
398def get_revisions(channel_maps: list) -> list:
399 """Gets a sorted list of unique revisions
401 :param channel_map: Channel map list
403 :returns: A sorted list of unique revisions
404 """
405 revisions = {channel_map["revision"] for channel_map in channel_maps}
406 return list(reversed(sorted(revisions)))
409def get_last_updated_versions(channel_maps):
410 """Get all channels in order of updates
412 :param channel_map: Channel map list
414 :returns: A list of channels ordered by last updated time
415 """
416 releases = []
417 for channel_map in channel_maps:
418 releases.append(channel_map["channel"])
420 return list(reversed(sorted(releases, key=lambda c: c["released-at"])))
423def get_last_updated_version(channel_maps):
424 """Get the oldest channel that was created
426 :param channel_map: Channel map list
428 :returns: The latest stable version, if no stable, the latest risk updated
429 """
430 newest_channel = None
431 for channel_map in channel_maps:
432 if not newest_channel:
433 newest_channel = channel_map
434 else:
435 if channel_map["channel"]["risk"] == "stable":
436 newest_channel = channel_map
438 if newest_channel["channel"]["risk"] == "stable":
439 break
441 return newest_channel
444def has_stable(channel_maps_list):
445 """Use the channel map to find out if the snap has a stable release
447 :param channel_maps_list: Channel map list
449 :returns: True or False
450 """
451 if channel_maps_list:
452 for arch in channel_maps_list:
453 for track in channel_maps_list[arch]:
454 for release in channel_maps_list[arch][track]:
455 if release["risk"] == "stable":
456 return True
458 return False
461def get_lowest_available_risk(channel_map, track):
462 """Get the lowest available risk for the default track
464 :param channel_map: Channel map list
465 :param track: The track of the channel
467 :returns: The lowest available risk
468 """
469 risk_order = ["stable", "candidate", "beta", "edge"]
470 lowest_available_risk = None
471 for arch in channel_map:
472 if arch in channel_map and track in channel_map[arch]:
473 releases = channel_map[arch][track]
474 for release in releases:
475 if not lowest_available_risk:
476 lowest_available_risk = release["risk"]
477 else:
478 risk_index = risk_order.index(release["risk"])
479 lowest_index = risk_order.index(lowest_available_risk)
480 if risk_index < lowest_index:
481 lowest_available_risk = release["risk"]
483 return lowest_available_risk
486def get_default_architecture(architectures):
487 """Pick the default architecture from a list of architectures.
489 Prefer ``amd64`` when published, otherwise the first architecture
490 sorted. This is the single source of truth for the default-architecture
491 preference, shared by the provenance badge endpoint and the security tab.
493 :param architectures: Iterable of architecture names
495 :returns: The default architecture, or ``None`` if the list is empty
496 """
497 architectures = list(architectures)
498 if not architectures:
499 return None
500 if "amd64" in architectures:
501 return "amd64"
502 return sorted(architectures)[0]
505def extract_info_channel_map(channel_map, track, risk):
506 """Get the confinement and version for a channel
508 :param channel_map: Channel map list
509 :param track: The track of the channel
510 :param risk: The risk of the channel
512 :returns: Dict containing confinement and version
513 """
514 context = {
515 "confinement": None,
516 "version": None,
517 }
519 for arch in channel_map:
520 if track in channel_map[arch]:
521 releases = channel_map[arch][track]
522 for release in releases:
523 if release["risk"] == risk:
524 context["confinement"] = release.get("confinement")
525 context["version"] = release.get("version")
527 return context
529 return context
532def get_video_embed_code(url):
533 """Get the embed code for videos
535 :param url: The url of the video
537 :returns: Embed code
538 """
539 if "youtube" in url:
540 return {
541 "type": "youtube",
542 "url": url.replace("watch?v=", "embed/"),
543 "id": url.rsplit("?v=", 1)[-1],
544 }
545 if "youtu.be" in url:
546 return {
547 "type": "youtube",
548 "url": url.replace("youtu.be/", "youtube.com/embed/"),
549 "id": url.rsplit("/", 1)[-1],
550 }
551 if "vimeo" in url:
552 return {
553 "type": "vimeo",
554 "url": url.replace("vimeo.com/", "player.vimeo.com/video/"),
555 "id": url.rsplit("/", 1)[-1],
556 }
557 if "asciinema" in url:
558 return {
559 "type": "asciinema",
560 "url": url + ".js",
561 "id": url.rsplit("/", 1)[-1],
562 }
565def filter_screenshots(media):
566 banner_regex = r"/banner(\-icon)?(_.*)?\.(png|jpg)"
568 return [
569 m
570 for m in media
571 if m["type"] == "screenshot" and not re.search(banner_regex, m["url"])
572 ][:5]
575def get_video(media):
576 video = None
577 for m in media:
578 if m["type"] == "video":
579 video = get_video_embed_code(m["url"])
580 break
581 return video
584def promote_snap_with_icon(snaps):
585 """Move the first snap with an icon to the front of the list
587 :param snaps: The list of snaps
589 :returns: A list of snaps
590 """
591 try:
592 snap_with_icon = next(snap for snap in snaps if snap["icon_url"] != "")
594 if snap_with_icon:
595 snap_with_icon_index = snaps.index(snap_with_icon)
597 snaps.insert(0, snaps.pop(snap_with_icon_index))
598 except StopIteration:
599 pass
601 return snaps
604def get_snap_developer(snap_name):
605 """Is this a special snap published by Canonical?
606 Show some developer information
608 :param snap_name: The name of a snap
610 :returns: a list of [display_name, url]
612 """
613 filename = "store/content/developers/snaps.yaml"
614 snaps = helpers.get_yaml(filename, typ="rt")
616 if snaps and snap_name in snaps:
617 return snaps[snap_name]
619 return None