1from settings import REASON_CATEGORIES 

2from flask import current_app 

3from flask_jwt_extended import get_jwt_claims 

4from sqlalchemy import func, text, and_, select 

5from sqlalchemy.sql.expression import null 

6from datetime import datetime 

7 

8from src import db, settings 

9from src.utils import pagination_resp, internal_err_resp, message, Paginator, err_resp 

10from src.model import TrackModel, GenreModel, ContentType, UserModel, ContentModel, RecommendedContentModel, RecommendedContentForGroupModel, MetaUserContentModel, BadRecommendationContentModel, TrackAdditionalModel 

11from src.schemas import TrackBase, TrackObject, GenreBase, TrackExtra, MetaUserContentBase, TrackAdditionalBase 

12 

13 

14class TrackService: 

15 @staticmethod 

16 def search_track_data(search_term, page, connected_user_uuid): 

17 """ Search track data by title """ 

18 if not (UserModel.query.filter_by(uuid=connected_user_uuid).first()): 

19 return err_resp("User not found!", 404) 

20 tracks, total_pages = Paginator.get_from( 

21 TrackModel.query.filter(TrackModel.title.ilike(search_term+"%")).union( 

22 TrackModel.query.filter(TrackModel.title.ilike("%"+search_term+"%"))), 

23 page, 

24 ) 

25 

26 try: 

27 track_data = TrackObject.loads(tracks) 

28 

29 return pagination_resp( 

30 message="Track data sent", 

31 content=track_data, 

32 page=page, 

33 total_pages=total_pages 

34 ) 

35 

36 except Exception as error: 

37 current_app.logger.error(error) 

38 return internal_err_resp() 

39 

40 @staticmethod 

41 def get_popular_tracks(page, connected_user_uuid): 

42 if not (user := UserModel.query.filter_by(uuid=connected_user_uuid).first()): 

43 return err_resp("User not found!", 404) 

44 

45 # NOTE IMDB measure of popularity does not seem to be relevant for this media. 

46 tracks, total_pages = Paginator.get_from( 

47 TrackModel.query.join(TrackModel.content, aliased=True).order_by( 

48 ContentModel.rating_count.desc().nullslast(), 

49 ContentModel.rating.desc().nullslast(), 

50 ), 

51 page, 

52 ) 

53 

54 try: 

55 track_data = TrackObject.loads(tracks) 

56 

57 return pagination_resp( 

58 message="Most popular track data sent", 

59 content=track_data, 

60 page=page, 

61 total_pages=total_pages 

62 ) 

63 

64 except Exception as error: 

65 current_app.logger.error(error) 

66 return internal_err_resp() 

67 

68 @staticmethod 

69 def get_recommended_tracks_for_user(page, connected_user_uuid, reco_engine): 

70 if not (user := UserModel.query.filter_by(uuid=connected_user_uuid).first()): 

71 return err_resp("User not found!", 404) 

72 

73 filters = [RecommendedContentModel.user_id == user.user_id] 

74 if reco_engine is not None: 

75 filters.append(RecommendedContentModel.engine == reco_engine) 

76 

77 tracks, total_pages = Paginator.get_from( 

78 db.session.query(RecommendedContentModel, TrackModel) 

79 .join(TrackModel.content) 

80 .join(RecommendedContentModel, RecommendedContentModel.content_id == ContentModel.content_id) 

81 .filter( 

82 and_(*filters) 

83 ) 

84 .order_by( 

85 RecommendedContentModel.score.desc().nullslast(), 

86 ContentModel.rating_count.desc().nullslast(), 

87 ContentModel.rating.desc().nullslast(), 

88 ), 

89 page, 

90 ) 

91 

92 try: 

93 def c_load(row): 

94 track = TrackExtra.load(row[1]) 

95 track["reco_engine"] = row[0].engine 

96 track["reco_score"] = row[0].score 

97 return track 

98 

99 track_data = list(map(c_load, tracks)) 

100 

101 return pagination_resp( 

102 message="Most popular track data sent", 

103 content=track_data, 

104 page=page, 

105 total_pages=total_pages 

106 ) 

107 

108 except Exception as error: 

109 current_app.logger.error(error) 

110 return internal_err_resp() 

111 

112 @staticmethod 

113 def get_recommended_tracks_for_group(page, connected_user_uuid, reco_engine): 

114 if not (user := UserModel.query.filter_by(uuid=connected_user_uuid).first()): 

115 return err_resp("User not found!", 404) 

116 

117 # Query for recommendation from group 

118 groups_ids = [ 

119 *list(map(lambda x: x.group_id, user.groups)), 

120 *list(map(lambda x: x.group_id, user.owned_groups)) 

121 ] 

122 

123 filters = [RecommendedContentForGroupModel.group_id.in_(groups_ids)] 

124 if reco_engine is not None: 

125 filters.append(RecommendedContentModel.engine == reco_engine) 

126 

127 tracks, total_pages = Paginator.get_from( 

128 db.session.query(RecommendedContentForGroupModel, TrackModel) 

129 .join(TrackModel.content) 

130 .join(RecommendedContentForGroupModel, RecommendedContentForGroupModel.content_id == ContentModel.content_id) 

131 .filter( 

132 and_(*filters) 

133 ) 

134 .order_by( 

135 RecommendedContentModel.score.desc().nullslast(), 

136 ContentModel.rating_count.desc().nullslast(), 

137 ContentModel.rating.desc().nullslast(), 

138 ), 

139 page, 

140 ) 

141 

142 try: 

143 def c_load(row): 

144 app = TrackExtra.load(row[1]) 

145 app["reco_engine"] = row[0].engine 

146 app["reco_score"] = row[0].score 

147 return app 

148 

149 track_data = list(map(c_load, tracks)) 

150 

151 return pagination_resp( 

152 message="Most popular track data sent", 

153 content=track_data, 

154 page=page, 

155 total_pages=total_pages 

156 ) 

157 

158 except Exception as error: 

159 current_app.logger.error(error) 

160 return internal_err_resp() 

161 

162 @staticmethod 

163 def get_ordered_genre(connected_user_uuid): 

164 if not (UserModel.query.filter_by(uuid=connected_user_uuid).first()): 

165 return err_resp("User not found!", 404) 

166 genres = GenreModel.query.filter_by( 

167 content_type=ContentType.TRACK).order_by(GenreModel.count.desc()).all() 

168 

169 try: 

170 genres_data = GenreBase.loads(genres) 

171 

172 resp = message(True, "track genres data sent") 

173 resp["content"] = genres_data 

174 return resp, 200 

175 

176 except Exception as error: 

177 current_app.logger.error(error) 

178 return internal_err_resp() 

179 

180 @staticmethod 

181 def get_history(user_uuid, page): 

182 """ Get the history of listened track """ 

183 if not (user := UserModel.query.filter_by(uuid=user_uuid).first()): 

184 return err_resp("User not found!", 404) 

185 

186 datas, total_pages = Paginator.get_from( 

187 db.session.query(MetaUserContentModel, TrackModel) 

188 .join(TrackModel.content) 

189 .join(MetaUserContentModel, MetaUserContentModel.content_id == TrackModel.content_id) 

190 .filter(and_(MetaUserContentModel.user_id == user.user_id, MetaUserContentModel.last_count_increment != None)) 

191 .order_by( 

192 MetaUserContentModel.last_count_increment.desc() 

193 ), 

194 page, 

195 ) 

196 

197 try: 

198 history_data = list(map(lambda x: { 

199 "last_played_date": MetaUserContentBase.load(x[0])["last_count_increment"], 

200 "track": TrackObject.load(x[1]) 

201 }, datas)) 

202 

203 return pagination_resp( 

204 message="History of listened track data successfully updated", 

205 content=history_data, 

206 page=page, 

207 total_pages=total_pages 

208 ) 

209 except Exception as error: 

210 import traceback 

211 traceback.print_exc() 

212 current_app.logger.error(error) 

213 return internal_err_resp() 

214 

215 @staticmethod 

216 def add_bad_recommendation(user_uuid, content_id, data): 

217 """ Add bad user recommendation """ 

218 if not (user := UserModel.query.filter_by(uuid=user_uuid).first()): 

219 return err_resp("User not found!", 404) 

220 

221 if not (track := TrackModel.query.filter_by(content_id=content_id).first()): 

222 return err_resp("Track not found!", 404) 

223 

224 try: 

225 for type, value in data.items(): 

226 if type in REASON_CATEGORIES['track']: 

227 for r in value: 

228 

229 new_bad_reco = BadRecommendationContentModel( 

230 user_id=user.user_id, 

231 content_id=track.content_id, 

232 reason_categorie=type, 

233 reason=r 

234 ) 

235 

236 db.session.add(new_bad_reco) 

237 db.session.flush() 

238 db.session.commit() 

239 

240 resp = message(True, "Bad recommendation has been registered.") 

241 return resp, 201 

242 

243 except Exception as error: 

244 current_app.logger.error(error) 

245 return internal_err_resp() 

246 

247 @staticmethod 

248 def add_additional_track(user_uuid, data): 

249 """ Add additional track""" 

250 if not (user := UserModel.query.filter_by(uuid=user_uuid).first()): 

251 return err_resp("User not found!", 404) 

252 

253 # Check permissions 

254 permissions = get_jwt_claims()['permissions'] 

255 if "add_content" not in permissions: 

256 return err_resp("Permission missing", 403) 

257 

258 try: 

259 

260 new_additional_track = TrackAdditionalModel( 

261 title=data['title'], 

262 ) 

263 

264 if 'year' in data: 

265 new_additional_track.year = data['year'] 

266 if 'artist_name' in data: 

267 new_additional_track.artist_name = data['artist_name'] 

268 if 'release' in data: 

269 new_additional_track.release = data['release'] 

270 if 'track_mmid' in data: 

271 new_additional_track.track_mmid = data['track_mmid'] 

272 if 'recording_mbid' in data: 

273 new_additional_track.recording_mbid = data['recording_mbid'] 

274 if 'spotify_id' in data: 

275 new_additional_track.spotify_id = data['spotify_id'] 

276 if 'covert_art_url' in data: 

277 new_additional_track.covert_art_url = data['covert_art_url'] 

278 

279 for genre_id in data["genres"]: 

280 if (ge := GenreModel.query.filter_by(genre_id=genre_id).first()): 

281 new_additional_track.genres.append(ge) 

282 else: 

283 return err_resp("Genre %s not found!" % genre_id, 404) 

284 

285 db.session.add(new_additional_track) 

286 db.session.commit() 

287 

288 resp = message(True, "Track have been added to validation.") 

289 return resp, 201 

290 

291 except Exception as error: 

292 current_app.logger.error(error) 

293 return internal_err_resp() 

294 

295 @staticmethod 

296 def get_additional_track(connected_user_uuid, page): 

297 if not (user := UserModel.query.filter_by(uuid=connected_user_uuid).first()): 

298 return err_resp("User not found!", 404) 

299 

300 # Check permissions 

301 permissions = get_jwt_claims()['permissions'] 

302 if "add_content" not in permissions: 

303 return err_resp("Permission missing", 403) 

304 

305 tracks, total_pages = Paginator.get_from( 

306 TrackAdditionalModel.query, 

307 page, 

308 ) 

309 

310 try: 

311 track_data = TrackAdditionalBase.loads(tracks) 

312 

313 return pagination_resp( 

314 message="Additional track data sent", 

315 content=track_data, 

316 page=page, 

317 total_pages=total_pages 

318 ) 

319 

320 except Exception as error: 

321 current_app.logger.error(error) 

322 return internal_err_resp() 

323 

324 @staticmethod 

325 def validate_additional_track(connected_user_uuid, track_id): 

326 if not (user := UserModel.query.filter_by(uuid=connected_user_uuid).first()): 

327 return err_resp("User not found!", 404) 

328 

329 # Check permissions 

330 permissions = get_jwt_claims()['permissions'] 

331 if "validate_added_content" not in permissions: 

332 return err_resp("Permission missing", 403) 

333 

334 if not (track := TrackAdditionalModel.query.filter_by(track_id=track_id).first()): 

335 return err_resp("Additional track not found!", 404) 

336 

337 try: 

338 content = ContentModel(rating=None, genres=track.genres) 

339 db.session.add(content) 

340 db.session.flush() 

341 

342 new_track = TrackModel( 

343 title=track.title, 

344 year=track.year, 

345 artist_name=track.artist_name, 

346 release=track.release, 

347 track_mmid=track.track_mmid, 

348 recording_mbid=track.recording_mbid, 

349 spotify_id=track.spotify_id, 

350 covert_art_url=track.covert_art_url, 

351 content=content 

352 ) 

353 db.session.add(new_track) 

354 db.session.delete(track) 

355 

356 db.session.commit() 

357 

358 resp = message( 

359 True, "Additional track data successfully validated") 

360 return resp, 201 

361 

362 except Exception as error: 

363 current_app.logger.error(error) 

364 return internal_err_resp() 

365 

366 @staticmethod 

367 def decline_additional_track(connected_user_uuid, track_id): 

368 if not (user := UserModel.query.filter_by(uuid=connected_user_uuid).first()): 

369 return err_resp("User not found!", 404) 

370 

371 # Check permissions 

372 permissions = get_jwt_claims()['permissions'] 

373 if "delete_content" not in permissions: 

374 return err_resp("Permission missing", 403) 

375 

376 if not (track := TrackAdditionalModel.query.filter_by(track_id=track_id).first()): 

377 return err_resp("Additional track not found!", 404) 

378 

379 try: 

380 db.session.delete(track) 

381 db.session.commit() 

382 

383 resp = message(True, "Additional track successfully deleted") 

384 return resp, 201 

385 

386 except Exception as error: 

387 current_app.logger.error(error) 

388 return internal_err_resp()