import json import logging import os from abc import abstractmethod from datetime import datetime from flask import Markup, flash from flask_admin import AdminIndexView, expose from flask_admin.babel import gettext from flask_admin.contrib import rediscli from flask_admin.contrib.sqla import ModelView from flask_admin.contrib.sqla.fields import QuerySelectMultipleField from sqlalchemy.exc import IntegrityError from sqlalchemy.orm.instrumentation import instance_state from werkzeug.utils import redirect, secure_filename from wtforms import fields from wtforms.fields import SelectField, SelectMultipleField, TextAreaField from wtforms.validators import DataRequired, Length, NumberRange, ValidationError from wtforms.widgets import CheckboxInput, ListWidget from apollo_main_db.apollo.models import ApolloVersionCarousel from src.config import S3_BUCKET_DOMAIN, S3_BUCKET_NAME from src.constants import DELETE_ACTION, OS_CHOICES, OS_LIST from src.keys import APOLLO_COPIES_PATTERN, APOLLO_LATEST_VERSION_PATTERN, APOLLO_USER_CAROUSEL_PATTERN, \ APOLLO_VERSION_CAROUSEL_PATTERN from src.redis_db import invalidate_cache_by_pattern, redis_clients from src.s3_utils import S3ImageUploadField, delete_image_by_url from src.utils import get_s3_image_key_from_url, has_basic_access, has_full_access from src.validators import OsValidator, VersionValidator class HomeView(AdminIndexView): @expose() def index(self): return self.render(self._template, is_access_granted=has_basic_access()) class InvalidateCacheMixin(ModelView): @abstractmethod def invalidate_cache(self): return NotImplemented def after_model_change(self, form, model, is_created): super().after_model_change(form, model, is_created) self.invalidate_cache() def after_model_delete(self, model): super().after_model_delete(model) self.invalidate_cache() class CreateModelOverrideMixin(ModelView): def _create_model(self, form, field_override_data: dict or None = None): if field_override_data is None: field_override_data = {} try: model = self._manager.new_instance() state = instance_state(model) self._manager.dispatch.init(state, [], {}) for name, field in form._fields.items(): if name in field_override_data: setattr(model, name, field_override_data[name]) continue field.populate_obj(model, name) self.session.add(model) self._on_model_change(form, model, True) self.session.commit() except Exception as ex: if not self.handle_view_exception(ex): flash(gettext("Failed to create record. %(error)s", error=str(ex)), "error") log = logging.getLogger("flask-admin.sqla") log.exception("Failed to create record.") self.session.rollback() return False else: self.after_model_change(form, model, True) return model class AuthModelView(ModelView): def is_accessible(self) -> bool: return has_basic_access() def inaccessible_callback(self, name, **kwargs): return redirect("/logout") def on_model_change(self, form, model, is_created): now = datetime.now() if is_created and hasattr(model, "created_at"): model.created_at = now if hasattr(model, "updated_at"): model.updated_at = now class ApolloCopiesView(AuthModelView, CreateModelOverrideMixin, InvalidateCacheMixin): class ChangeLogField(fields.TextAreaField): def _value(self): return json.loads(self.data, strict=False).get("changelog", "") if self.data else "" def process_formdata(self, valuelist): if valuelist: value = valuelist[0] if not value: self.data = None return self.data = '{"changelog": "%s"}' % (valuelist[0],) column_formatters = {"data": lambda v, c, m, p: json.loads(m.data, strict=False).get("changelog", "")} form_overrides = {"data": ChangeLogField} column_labels = {"os": "OS", "data": "Splash Screen Copy"} form_choices = {"os": OS_CHOICES} form_excluded_columns = ("created_at", "updated_at") column_default_sort = ("created_at", True) form_args = { "os": {"label": "OS", "validators": [DataRequired(), OsValidator()]}, "version": { "label": "Version", "validators": [DataRequired(), VersionValidator()], "render_kw": {"placeholder": "e.g. 1.2.3"}, }, "data": {"label": "Splash Screen Copy", "validators": [DataRequired()]}, } def invalidate_cache(self): for redis_client in redis_clients.values(): invalidate_cache_by_pattern(redis_client, APOLLO_COPIES_PATTERN) invalidate_cache_by_pattern(redis_client, APOLLO_LATEST_VERSION_PATTERN) def get_create_form(self): form = super().get_create_form() form.os = SelectMultipleField( label="OS", choices=OS_CHOICES, validators=[DataRequired(), OsValidator()], default=OS_LIST, widget=ListWidget(prefix_label=False), option_widget=CheckboxInput(), ) return form def changelog_validation(self, form): changelog_len = len(form.data.raw_data[0].replace("\r\n", "")) if changelog_len > 120: raise ValidationError(f"Max Сhangelog length is 120 characters, current is {changelog_len}") def on_model_change(self, form, model, is_created): self.changelog_validation(form) return super().on_model_change(form, model, is_created) def create_model(self, form): try: self.changelog_validation(form) except ValidationError as e: flash(str(e), "error") return False do_redirect = True selected_os = form.os.data for _os in selected_os: created_model = None try: created_model = self._create_model(form, {"os": _os}) except IntegrityError: self.session.rollback() log = logging.getLogger("flask-admin.sqla") log.exception("Failed to create record.") flash(f"Duplicate entry {_os.upper()} - {form.version.data}.", "error") if not created_model: do_redirect = False else: flash(f"Version for {_os.upper()} is created.") return do_redirect class ApolloGoVersionView(AuthModelView, CreateModelOverrideMixin, InvalidateCacheMixin): column_labels = {"os": "OS", "created_at": "Created at"} form_choices = {"os": OS_CHOICES} form_excluded_columns = ("created_at",) column_default_sort = ("created_at", True) form_args = { "os": {"label": "OS", "validators": [DataRequired(), OsValidator()]}, "version": { "label": "Version", "validators": [DataRequired(), VersionValidator()], "render_kw": {"placeholder": "e.g. 1.2.3"}, }, } column_searchable_list = ("os", "version") def invalidate_cache(self): for redis_client in redis_clients.values(): invalidate_cache_by_pattern(redis_client, APOLLO_COPIES_PATTERN) invalidate_cache_by_pattern(redis_client, APOLLO_LATEST_VERSION_PATTERN) def get_create_form(self): form = super().get_create_form() form.os = SelectMultipleField( label="OS", choices=OS_CHOICES, validators=[DataRequired(), OsValidator()], default=[], widget=ListWidget(prefix_label=False), option_widget=CheckboxInput(), ) return form def create_model(self, form): selected_os = form.data.get("os", []) version = form.data.get("version") model = None for _os in selected_os: model = self._create_model(form, {"os": _os}) if not model: continue flash(gettext(f"Changelog for version {version} and OS {_os.upper()} was successfully created.")) return model class RedisCliView(rediscli.RedisCli): def is_visible(self): return has_full_access() def is_accessible(self) -> bool: return has_full_access() # TODO: https://data-analytics.atlassian.net/browse/AG-6690 fix while waiting for release Flask-Admin==1.5.8 @expose("/run/", methods=("POST",)) def execute_view(self): from flask import request from flask_admin.contrib.rediscli import CommandError log = logging.getLogger("flask-admin.redis") """ AJAX API. """ try: cmd = request.form.get("cmd") if not cmd: return self._error("Cli: Empty command.") parts = self._parse_cmd(cmd) if not parts: return self._error("Cli: Failed to parse command.") return self._execute_command(parts[0], parts[1:]) except CommandError as err: return self._error("Cli: %s" % err) except Exception as ex: log.exception(ex) return self._error("Cli: %s" % ex) class ApolloVersionCarouselView(AuthModelView, CreateModelOverrideMixin, InvalidateCacheMixin): form_columns = ("os", "version") form_excluded_columns = ("created_at", "updated_at") column_labels = {"os": "OS", "version": "Version", "created_at": "Created at", "updated_at": "Updated at"} column_searchable_list = ("os", "version") form_args = { "os": {"label": "OS", "validators": [DataRequired(), OsValidator()]}, "version": { "label": "Version", "validators": [DataRequired(), VersionValidator()], "render_kw": {"placeholder": "e.g. 1.2.3"}, }, } column_default_sort = [("created_at", True), ("os", False)] def invalidate_cache(self): for redis_client in redis_clients.values(): invalidate_cache_by_pattern(redis_client, APOLLO_VERSION_CAROUSEL_PATTERN) def get_create_form(self): form = super().get_create_form() form.os = SelectMultipleField( label="OS", choices=OS_CHOICES, validators=[DataRequired(), OsValidator()], default=[], widget=ListWidget(prefix_label=False), option_widget=CheckboxInput(), ) return form def get_edit_form(self): form = super().get_create_form() form.os = SelectField(label="OS", choices=OS_CHOICES, validators=[DataRequired(), OsValidator()]) return form def validate_form(self, form): action = form.data.get("action") if action == DELETE_ACTION: return super().validate_form(form) os, version = form.data.get("os"), form.data.get("version") if form._obj: return self.edit_form_validation(form, os, version) return self.create_form_validation(form, os, version) def create_form_validation(self, form, os, version): if not os or not version: return super().validate_form(form) os_list = os if isinstance(os, list) else [os] existed_versions = self.get_query().filter(self.model.version == version, self.model.os.in_(os_list)) for _version in existed_versions: flash(f"Version <{_version.version}> already exists for <{_version.os.upper()}>!", "error") return False return super().validate_form(form) def edit_form_validation(self, form, os, version): if not os or not version: return super().validate_form(form) if form._obj.version == version and form._obj.os == os: return super().validate_form(form) os_list = os if isinstance(os, list) else [os] existed_versions = self.get_query().filter(self.model.version == version, self.model.os.in_(os_list)) for _version in existed_versions: flash(f"Version <{_version.version}> already exists for <{_version.os.upper()}>!", "error") return False return super().validate_form(form) def create_model(self, form): selected_os = form.data.get("os", []) version = form.data.get("version") model = None for _os in selected_os: model = self._create_model(form, {"os": _os}) if not model: continue flash(gettext(f"Version {version} was created for OS: <{_os.upper()}>.")) return model class ApolloVersionCarouselSlideView(AuthModelView, CreateModelOverrideMixin, InvalidateCacheMixin): def _add_timestamp_to_name(obj, file_data): parts = os.path.splitext(file_data.filename) return secure_filename(f'{parts[0]}-dt-{datetime.now().strftime("%Y%m%d-%H%M%S")}{parts[1]}') def _image_preview(view, context, model, name): if not model.img_url: return "" return Markup(f'No such image') def _thumb_name(filename): thumb_url = get_s3_image_key_from_url(filename) return thumb_url form_excluded_columns = ("created_at", "updated_at") column_formatters = {"img_url": _image_preview} column_labels = { "slide_num": "Slide Number", "title_1": "Title #1", "img_url": "Image", "title_2": "Title #2", "description": "Description", "version_carousel": "Version", "created_at": "Created at", "updated_at": "Updated at", } form_args = { "slide_num": {"validators": [DataRequired(), NumberRange(min=1, max=10)]}, "title_1": {"default": "What's new?", "validators": [DataRequired(), Length(max=15)]}, "img_url": dict( validators=[DataRequired()], base_path="/", relative_path="carousel/slides/", url_relative_path="", namegen=_add_timestamp_to_name, endpoint="", allowed_extensions=["png", "gif", "jpg"], ), "title_2": {"validators": [DataRequired(), Length(max=20)]}, "description": {"validators": [DataRequired(), Length(max=180)]}, } column_searchable_list = ("title_1", "title_2", ApolloVersionCarousel.os, ApolloVersionCarousel.version) column_select_related_list = ("version_carousel",) column_filters = ("slide_num", "title_1", "title_2", ApolloVersionCarousel.os, ApolloVersionCarousel.version) column_sortable_list = ("slide_num", "created_at", "updated_at") column_default_sort = [ (ApolloVersionCarousel.created_at, True), (ApolloVersionCarousel.os, False), ("slide_num", False), ] form_overrides = {"img_url": S3ImageUploadField, "description": TextAreaField} def on_model_change(self, form, model, is_created): super().on_model_change(form, model, is_created) if is_created: model.img_url = f"https://{S3_BUCKET_DOMAIN}/{model.img_url}" return # On editing - if new img url not equal previous one - save new img full path. if model.img_url not in form.img_url.object_data: delete_image_by_url(form.img_url.object_data, S3_BUCKET_NAME) model.img_url = f"https://{S3_BUCKET_DOMAIN}/{model.img_url}" def after_model_delete(self, model): super().after_model_delete(model) delete_image_by_url(model.img_url, S3_BUCKET_NAME) def invalidate_cache(self): for redis_client in redis_clients.values(): invalidate_cache_by_pattern(redis_client, APOLLO_VERSION_CAROUSEL_PATTERN) def get_create_form(self): form = super().get_create_form() form.version_carousel = QuerySelectMultipleField( label="Version", query_factory=lambda: self.session.query(ApolloVersionCarousel).all() ) return form def validate_form(self, form): action = form.data.get("action") if action == DELETE_ACTION: return super().validate_form(form) if form._obj: return self.edit_form_validation(form) return self.create_form_validation(form) def create_form_validation(self, form): if hasattr(form, "version_carousel") is False: return super().validate_form(form) carousel_list = form.version_carousel.data or [] for carousel_data in carousel_list: if not carousel_data: continue slide_num = form.slide_num.data slide_exists = ( self.get_query() .filter(self.model.slide_num == slide_num, self.model.carousel_id == carousel_data.id) .scalar() ) if slide_exists: flash(f"Slide #{slide_num} already exists for <{carousel_data}> version!", "error") return False return super().validate_form(form) def edit_form_validation(self, form): carousel_data = form.version_carousel.data if carousel_data: slide_num = form.slide_num.data obj = form._obj if obj.slide_num == slide_num and obj.carousel_id == carousel_data.id: return super().validate_form(form) slide_exists = ( self.get_query() .filter(self.model.slide_num == slide_num, self.model.carousel_id == carousel_data.id) .scalar() ) if slide_exists: flash("Such record already exists!", "error") return False return super().validate_form(form) def scaffold_form(self): form_class = super().scaffold_form() static_root_parent = "static/" form_class.img_url.kwargs["bucket_name"] = S3_BUCKET_NAME form_class.img_url.kwargs["static_root_parent"] = static_root_parent return form_class def create_model(self, form): selected_versions = form.data.get("version_carousel", []) or [None] slide_num = form.data.get("slide_num") model = None for _version in selected_versions: model = self._create_model(form, {"version_carousel": _version}) if model and _version is not None: flash(gettext(f"Slide #{slide_num} was created for version: <{_version}>.")) return model class ApolloUserCarouselVersionView(AuthModelView, CreateModelOverrideMixin, InvalidateCacheMixin): can_edit = False can_create = False form_excluded_columns = ("created_at", "updated_at") form_args = {"user_id": {"validators": [DataRequired()]}} column_labels = { "user_id": "User ID", "version_carousel": "Version", "created_at": "Created at", "updated_at": "Updated at", } column_searchable_list = ("user_id", ApolloVersionCarousel.os, ApolloVersionCarousel.version) column_select_related_list = ("version_carousel",) column_filters = ("user_id", ApolloVersionCarousel.os, ApolloVersionCarousel.version) def invalidate_cache(self): for redis_client in redis_clients.values(): invalidate_cache_by_pattern(redis_client, APOLLO_USER_CAROUSEL_PATTERN) def validate_form(self, form): action = form.data.get("action") if action == DELETE_ACTION: return super().validate_form(form) if form._obj: return self.edit_form_validation(form) return self.create_form_validation(form) def create_form_validation(self, form): if hasattr(form, "version_carousel") is False: return super().validate_form(form) carousel_list = form.version_carousel.data or [] for carousel_data in carousel_list: if not carousel_data: continue user_id = form.user_id.data user_viewed = ( self.get_query() .filter(self.model.user_id == user_id, self.model.carousel_id == carousel_data.id) .scalar() ) if user_viewed: flash(f"Record -> User (ID: {user_id}) viewed <{carousel_data}> carousel has already exists!", "error") return False return super().validate_form(form) def edit_form_validation(self, form): carousel_data = form.version_carousel.data user_id = form.user_id.data obj = form._obj if obj.user_id == user_id and obj.carousel_id == carousel_data.id: return super().validate_form(form) user_viewed = ( self.get_query().filter(self.model.user_id == user_id, self.model.carousel_id == carousel_data.id).scalar() ) if user_viewed: flash("Such record already exists!", "error") return False return super().validate_form(form) def get_create_form(self): form = super().get_create_form() form.version_carousel = QuerySelectMultipleField( label="Version", validators=[DataRequired()], query_factory=lambda: self.session.query(ApolloVersionCarousel).all(), ) return form def create_model(self, form): selected_versions = form.data.get("version_carousel", []) user_id = form.data.get("user_id") model = None for _version in selected_versions: model = self._create_model(form, {"version_carousel": _version}) if model and _version is not None: flash(gettext(f"User (ID: {user_id}) has viewed <{_version}> carousel.")) return model class UserOnboardingView(AuthModelView): form_excluded_columns = ("created_at",) column_list = ("user_id", "is_mobile", "is_portal", "created_at") column_labels = {"user_id": "User ID", "created_at": "Created at"} column_sortable_list = ("user_id", "created_at") column_searchable_list = ("user_id",) column_default_sort = ("created_at", True) class ApolloNPSSurveyResultsView(AuthModelView): can_edit = False can_create = False can_delete = False can_export = True form_excluded_columns = ("created_at", "updated_at") column_list = ("survey_id", "user_id", "email", "name", "answers", "status") column_labels = { "survey_id": "Survey ID", "user_id": "User ID", "email": "Email", "name": "Name", "answers": "Answers", "status": "status", } column_sortable_list = ("survey_id", "created_at") column_searchable_list = ("survey_id",) column_default_sort = ("created_at", True) class ApolloNPSSurveyQuestionsView(AuthModelView): can_edit = False can_create = False can_delete = False form_excluded_columns = ("created_at", "updated_at") column_list = ("survey_id", "question_num", "message", "type", "labels") column_labels = { "survey_id": "Survey ID", "question_num": "Question Number", "message": "Message", "type": "Type", "labels": "Labels", } column_sortable_list = ("survey_id", "created_at") column_searchable_list = ("survey_id",) column_default_sort = [("survey_id", True), ("question_num", False)] class ApolloNPSSurveyView(AuthModelView): can_edit = False can_create = False can_delete = False form_excluded_columns = ("created_at", "updated_at") column_list = ("id", "name", "start_date", "end_date", "active") column_labels = { "id": "Survey ID", "name": "Survey Name", "start_date": "Start Date", "end_date": "End Date", "active": "Active", } column_searchable_list = ("id",) form_columns = ("name", "start_date", "end_date", "active")