Coverage for webapp/store/logic.py: 77%

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 )

101

102 if "prev" in links:

103 links_result["prev"] = convert_navigation_url(

104 url, links["prev"]["href"]

105 )

106

107 if "self" in links:

108 links_result["self"] = convert_navigation_url(

109 url, links["self"]["href"]

110 )

111

112 return links_result

113

114

115def convert_navigation_url(url, link):

116 """Convert navigation link from offest/limit to size/page

117

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

121

122 :param url: The new url

123 :param link: The navigation url returned by the API

124

125 :returns: The new navigation link

126 """

127 url_parsed = urlparse(link)

128 host_url = "{base_url}" "?q={q}&limit={limit}&offset={offset}"

129

130 url_queries = parse_qs(url_parsed.query)

131

132 if "q" in url_queries:

133 q = url_queries["q"][0]

134 else:

135 q = ""

136

137 if "section" in url_queries:

138 category = url_queries["section"][0]

139 else:

140 category = ""

141

142 size = int(url_queries["size"][0])

143 page = int(url_queries["page"][0])

144

145 url = host_url.format(

146 base_url=url, q=q, limit=size, offset=size * (page - 1)

147 )

148

149 if category != "":

150 url += "&category=" + category

151

152 return url

153

154

155def build_pagination_link(snap_searched, snap_category, page):

156 """Build pagination link

157

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

161

162 :returns: A url string

163 """

164 params = []

165

166 if snap_searched:

167 params.append("q=" + snap_searched)

168

169 if snap_category:

170 params.append("category=" + snap_category)

171

172 if page:

173 params.append("page=" + str(page))

174

175 return "/search?" + "&".join(params)

176

177

178def convert_channel_maps(channel_map):

179 """Converts channel maps list to format easier to manipulate

180

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 }

199

200 :param channel_maps_list: The channel maps list returned by the API

201

202 :returns: The channel maps reshaped

203 """

204 channel_map_restruct = {}

205

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] = []

213

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 }

223

224 channel_map_restruct[arch][track].append(info)

225

226 return channel_map_restruct

227

228

229def convert_date(date_to_convert):

230 """Convert date to human readable format: Month Day Year

231

232 If date is less than a day return: today or yesterday

233

234 Format of date to convert: 2019-01-12T16:48:41.821037+00:00

235 Output: Jan 12 2019

236

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)

243

244 if delta < date_parsed:

245 return humanize.naturalday(date_parsed).title()

246 else:

247 return date_parsed.strftime("%-d %B %Y")

248

249

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

252

253 A snap is considered old if it hasn't been updated in the specified

254 number of years (default: 2 years).

255

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

263

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)

268

269 now = datetime.datetime.now(datetime.timezone.utc)

270

271 delta = relativedelta(now, date_parsed)

272 years_since_update = delta.years

273

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

278

279

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]

294

295blacklist = ["featured"]

296

297

298def format_category_name(slug):

299 """Format category name into a standard title format

300

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 )

310

311

312def get_categories(categories_json):

313 """Retrieve and flatten the nested array from the legacy API response.

314

315 :param categories_json: The returned json

316 :returns: A list of categories

317 """

318

319 categories = []

320

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"])

326

327 for category in categories_list:

328 categories.append(

329 {"slug": category, "name": format_category_name(category)}

330 )

331

332 return categories

333

334

335def get_snap_categories(snap_categories):

336 """Retrieve list of categories with names for a snap.

337

338 :param snap_categories: List of snap categories from snap info API

339 :returns: A list of categories with names

340 """

341 categories = []

342

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 )

351

352 return categories

353

354

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

360

361 :param channel_map: Channel map list

362

363 :returns: A tuple of default/stable, track/risk channel map objects

364 """

365 ordered_versions = get_last_updated_versions(channel_maps)

366

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

382

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

390

391

392def get_revisions(channel_maps: list) -> list:

393 """Gets a sorted list of unique revisions

394

395 :param channel_map: Channel map list

396

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)))

401

402

403def get_last_updated_versions(channel_maps):

404 """Get all channels in order of updates

405

406 :param channel_map: Channel map list

407

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"])

413

414 return list(reversed(sorted(releases, key=lambda c: c["released-at"])))

415

416

417def get_last_updated_version(channel_maps):

418 """Get the oldest channel that was created

419

420 :param channel_map: Channel map list

421

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

431

432 if newest_channel["channel"]["risk"] == "stable":

433 break

434

435 return newest_channel

436

437

438def has_stable(channel_maps_list):

439 """Use the channel map to find out if the snap has a stable release

440

441 :param channel_maps_list: Channel map list

442

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

451

452 return False

453

454

455def get_lowest_available_risk(channel_map, track):

456 """Get the lowest available risk for the default track

457

458 :param channel_map: Channel map list

459 :param track: The track of the channel

460

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"]

476

477 return lowest_available_risk

478

479

480def extract_info_channel_map(channel_map, track, risk):

481 """Get the confinement and version for a channel

482

483 :param channel_map: Channel map list

484 :param track: The track of the channel

485 :param risk: The risk of the channel

486

487 :returns: Dict containing confinement and version

488 """

489 context = {

490 "confinement": None,

491 "version": None,

492 }

493

494 for arch in channel_map:

495 if track in channel_map[arch]:

496 releases = channel_map[arch][track]

497 for release in releases:

498 if release["risk"] == risk:

499 context["confinement"] = release.get("confinement")

500 context["version"] = release.get("version")

502 return context

504 return context

507def get_video_embed_code(url):

508 """Get the embed code for videos

509

510 :param url: The url of the video

511

512 :returns: Embed code

513 """

514 if "youtube" in url:

515 return {

516 "type": "youtube",

517 "url": url.replace("watch?v=", "embed/"),

518 "id": url.rsplit("?v=", 1)[-1],

519 }

520 if "youtu.be" in url:

521 return {

522 "type": "youtube",

523 "url": url.replace("youtu.be/", "youtube.com/embed/"),

524 "id": url.rsplit("/", 1)[-1],

525 }

526 if "vimeo" in url:

527 return {

528 "type": "vimeo",

529 "url": url.replace("vimeo.com/", "player.vimeo.com/video/"),

530 "id": url.rsplit("/", 1)[-1],

531 }

532 if "asciinema" in url:

533 return {

534 "type": "asciinema",

535 "url": url + ".js",

536 "id": url.rsplit("/", 1)[-1],

537 }

538

539

540def filter_screenshots(media):

541 banner_regex = r"/banner(\-icon)?(_.*)?\.(png|jpg)"

542

543 return [

544 m

545 for m in media

546 if m["type"] == "screenshot" and not re.search(banner_regex, m["url"])

547 ][:5]

548

549

550def get_video(media):

551 video = None

552 for m in media:

553 if m["type"] == "video":

554 video = get_video_embed_code(m["url"])

555 break

556 return video

557

558

559def promote_snap_with_icon(snaps):

560 """Move the first snap with an icon to the front of the list

561

562 :param snaps: The list of snaps

563

564 :returns: A list of snaps

565 """

566 try:

567 snap_with_icon = next(snap for snap in snaps if snap["icon_url"] != "")

568

569 if snap_with_icon:

570 snap_with_icon_index = snaps.index(snap_with_icon)

571

572 snaps.insert(0, snaps.pop(snap_with_icon_index))

573 except StopIteration:

574 pass

575

576 return snaps

577

578

579def get_snap_developer(snap_name):

580 """Is this a special snap published by Canonical?

581 Show some developer information

582

583 :param snap_name: The name of a snap

584

585 :returns: a list of [display_name, url]

586

587 """

588 filename = "store/content/developers/snaps.yaml"

589 snaps = helpers.get_yaml(filename, typ="rt")

590

591 if snaps and snap_name in snaps:

592 return snaps[snap_name]

593

594 return None