import logging from datetime import date, timedelta from flask_admin import AdminIndexView, BaseView, expose from flask_admin.contrib.sqla import ModelView from flask_admin.menu import MenuLink from flask_atlas_auth import current_user from sqlalchemy import delete, desc, func, insert, nulls_last from src.auth import Auth from src.constants import EXCLUDED_ARTIST_LIST, EXCLUDED_LABELS_LIST, STRICT_EXCLUDED_LABELS_LIST from src.models import Artist, Label, NonSignedArtist, NonSignedTrack from src.query_wrapper import QueryWrapper from src.settings import DEFAULT_FORMATTERS, EXPORT_FORMATTERS, Converter, LabelFilterConverter from src.utils import escape_name, get_last_data, get_user_email, get_user_id, has_basic_access, is_authenticated log = logging.getLogger(__name__) class HomeView(AdminIndexView): @expose('/') def index(self): template = self._template if has_basic_access(): artist_last_data = get_last_data(NonSignedArtist) track_last_data = get_last_data(NonSignedTrack) if artist_last_data: self._template_args['total_nsa'] = format(artist_last_data['total_nsa'], ',') self._template_args['artist_updated_at'] = artist_last_data['created_at'].strftime("%Y-%m-%d %H:%M:%S") if track_last_data: self._template_args['total_nst'] = format(track_last_data['total_nst'], ',') self._template_args['track_updated_at'] = track_last_data['created_at'].strftime("%Y-%m-%d %H:%M:%S") template = 'admin/home.html' return self.render(template, is_access_granted=has_basic_access()) class UploadView(Auth, BaseView): @expose('/') def index(self): created_at = func.now() artist_data = [{ "name": artist.replace("\\", ""), "escaped_name": artist, "user_id": current_user.id, "email": current_user.email, "created_at": created_at, "updated_at": created_at } for artist in EXCLUDED_ARTIST_LIST] partial_label_data = [{ "name": label.replace("\\", ""), "escaped_name": label, "user_id": current_user.id, "email": current_user.email, "created_at": created_at, "updated_at": created_at, "full_match": False } for label in EXCLUDED_LABELS_LIST] strict_label_data = [{ "name": label.replace("\\", ""), "escaped_name": label, "user_id": current_user.id, "email": current_user.email, "created_at": created_at, "updated_at": created_at, "full_match": True } for label in STRICT_EXCLUDED_LABELS_LIST] label_data = partial_label_data + strict_label_data inser_artist_query = insert(Artist).values(artist_data) inser_label_query = insert(Label).values(label_data) self._template = 'admin/upload.html' for query in (inser_artist_query, inser_label_query): res = QueryWrapper.insert(query) if isinstance(res, dict): self._template_args['message'] = res["error"] return self.render(self._template) message = 'basic artist/label data uploading process succesfully finished.' self._template_args['message'] = message return self.render(self._template) class ArtistView(Auth, ModelView): column_list = ['name', 'email', 'updated_at'] column_type_formatters = DEFAULT_FORMATTERS column_default_sort = [('updated_at', True), ('name', False)] column_sortable_list = ('name', 'user_id', 'email', 'comment', 'created_at', 'updated_at') column_searchable_list = ['name', 'email'] filter_converter = Converter() column_filters = ('email', 'updated_at') column_editable_list = ['name', 'comment'] column_details_list = ['name', 'user_id', 'email', 'comment', 'created_at', 'updated_at'] column_type_formatters_detail = DEFAULT_FORMATTERS column_type_formatters_export = EXPORT_FORMATTERS page_size = 100 can_view_details = True can_set_page_size = True can_create = True can_edit = True can_delete = True can_export = True form_create_rules = ( 'name', 'comment', ) form_edit_rules = ( 'name', 'comment', ) def on_model_change(self, form, model, is_created): model.user_id = get_user_id() model.email = get_user_email() model.escaped_name = escape_name(model.name) if is_created: model.updated_at = func.now() class LabelView(Auth, ModelView): column_list = ['name', 'full_match', 'email', 'updated_at'] column_type_formatters = DEFAULT_FORMATTERS column_default_sort = [('updated_at', True), ('name', False)] column_sortable_list = ('name', 'parent_label', 'full_match', 'comment', 'user_id', 'email', 'created_at', 'updated_at') column_searchable_list = ['name', 'email'] filter_converter = LabelFilterConverter() column_filters = ("full_match", 'email', 'updated_at') column_editable_list = ['name', 'parent_label', 'full_match', 'comment'] column_details_list = ['name', 'parent_label', 'full_match', 'comment', 'user_id', 'email', 'created_at', 'updated_at'] column_type_formatters_detail = DEFAULT_FORMATTERS column_type_formatters_export = EXPORT_FORMATTERS page_size = 100 can_view_details = True can_set_page_size = True can_create = True can_edit = True can_delete = True can_export = True form_create_rules = ( 'name', 'parent_label', 'full_match', 'comment', ) form_edit_rules = ( 'name', 'parent_label', 'full_match', 'comment', ) def on_model_change(self, form, model, is_created): model.user_id = get_user_id() model.email = get_user_email() model.escaped_name = escape_name(model.name) if is_created: model.updated_at = func.now() class SignedArtistReportView(Auth, ModelView): column_exclude_list = ('created_at',) column_export_exclude_list = ('created_at',) column_type_formatters = DEFAULT_FORMATTERS column_default_sort = [('drop_off_date', True), ('artist_name', False)] column_searchable_list = ['artist_name', 'label', 'album_type'] column_filters = ('artist_name', 'label', 'drop_off_date') column_type_formatters_detail = DEFAULT_FORMATTERS column_type_formatters_export = EXPORT_FORMATTERS filter_converter = Converter() page_size = 100 can_view_details = True can_set_page_size = True can_create = False can_edit = False can_delete = False can_export = True @property def seven_days_ago(self): today = date.today() return str(today - timedelta(days=7)) def get_list(self, page, sort_column, sort_desc, search, filters, execute=True, page_size=None): # aplly default filter that return return all signed artists that were dropped for the last 7 days. if not filters: filters = [(8, 'Drop Off Date', self.seven_days_ago)] return super().get_list(page, sort_column, sort_desc, search, filters, execute, page_size) def _order_by(self, query, joins, sort_joins, sort_field, sort_desc): query, joins = super()._order_by(query, joins, sort_joins, sort_field, sort_desc) # updates default query that values sortable field that equal to NULL are placing at the end of the list. if sort_desc and sort_field.key != 'drop_off_date': query = query.subquery() query = self.session.query(query).order_by(nulls_last(desc(sort_field.key))) return query, joins class SyncView(Auth, BaseView): def __init__(self, *args, **kwargs) -> None: self.cms_api = kwargs.pop('cms_api') super().__init__(*args, **kwargs) @expose('/') def index(self): artists = self.cms_api.get_artist_objects() labels = self.cms_api.get_label_objects() delete_artist_query = delete(Artist) delete_label_query = delete(Label) for query in (delete_artist_query, delete_label_query): res = QueryWrapper.delete(query) if isinstance(res, dict): self._template_args['message'] = res["error"] return self.render('admin/upload.html') log.debug('artists and labels were removed.') insert_artist_query = insert(Artist).values(artists) insert_label_query = insert(Label).values(labels) for query in (insert_artist_query, insert_label_query): res = QueryWrapper.insert(query) if "error" in res: self._template_args['message'] = res["error"] return self.render('admin/upload.html') log.debug('artists and labels were inserted.') message = "data was succesfully updated from prod environment" self._template_args['message'] = message return self.render('admin/upload.html') class LoginMenuLink(MenuLink): def is_accessible(self): return not is_authenticated() class LogoutMenuLink(MenuLink): def is_accessible(self): return is_authenticated()