<commit_before># -*- coding: utf-8 -*-
"""Application views."""

import json
import os
import secrets
from collections import namedtuple
from datetime import datetime
from io import BytesIO

from flask import abort, flash, jsonify, redirect, render_template, request, send_file, send_from_directory, url_for
from flask_admin.actions import action
from flask_admin.contrib.peewee import ModelView
from flask_admin.form import SecureForm
from flask_admin.model import typefmt
from flask_login import current_user, login_required
from jinja2 import Markup
from playhouse.shortcuts import model_to_dict
from werkzeug.utils import secure_filename
from wtforms.fields import BooleanField

from orcid_api.rest import ApiException

from . import admin, app, models, orcid_client, utils
from .config import ORCID_BASE_URL, SCOPE_ACTIVITIES_UPDATE, SCOPE_READ_LIMITED
from .forms import (ApplicationFrom, BitmapMultipleValueField, CredentialForm, EmailTemplateForm,
                    FileUploadForm, JsonOrYamlFileUploadForm, LogoForm, OrgRegistrationForm,
                    PartialDateField, RecordForm, UserInvitationForm)
from .login_provider import roles_required
from .models import (Affiliation, AffiliationRecord, CharField, Client, File, FundingContributor,
                     FundingRecord, Grant, ModelException, OrcidApiCall, OrcidToken, Organisation,
                     OrgInfo, OrgInvitation, PartialDate, Role, Task, TextField, Token, Url, User,
                     UserInvitation, UserOrg, UserOrgAffiliation, db)
# NB! Should be disabled in production
from .pyinfo import info
from .utils import generate_confirmation_token, send_user_invitation

HEADERS = {"Accept": "application/vnd.orcid+json", "Content-type": "application/vnd.orcid+json"}


@app.route("/favicon.ico")
def favicon():
    """Support for the "favicon" legacy: faveicon location in the root directory."""
    return send_from_directory(
        os.path.join(app.root_path, "static", "images"),
        "favicon.ico",
        mimetype="image/vnd.microsoft.icon")


@app.route("/pyinfo")
@roles_required(Role.SUPERUSER)
def pyinfo():
    """Show Python and runtime environment and settings."""
    return render_template("pyinfo.html", **info)


@app.route("/u/<short_id>")
def short_url(short_id):
    """Redirect to the full URL."""
    try:
        u = Url.get(short_id=short_id)
        if request.args:
            return redirect(utils.append_qs(u.url, **request.args))
        return redirect(u.url)
    except Url.DoesNotExist:
        abort(404)


def read_uploaded_file(form):
    """Read up the whole content and deconde it and return the whole content."""
    raw = request.files[form.file_.name].read()
    for encoding in "utf-8", "utf-8-sig", "utf-16":
        try:
            return raw.decode(encoding)
        except UnicodeDecodeError:
            continue
    return raw.decode("latin-1")


class AppModelView(ModelView):
    """ModelView customization."""

    roles_required = Role.SUPERUSER
    export_types = [
        "csv",
        "xls",
        "tsv",
        "yaml",
        "json",
        "xlsx",
        "ods",
        "html",
    ]
    form_base_class = SecureForm
    column_type_formatters = dict(typefmt.BASE_FORMATTERS)
    column_type_formatters.update({
        datetime:
        lambda view, value: Markup(value.strftime("%Y‑%m‑%d&nbsp;%H:%M")),
    })
    column_type_formatters_export = dict(typefmt.EXPORT_FORMATTERS)
    column_type_formatters_export.update({PartialDate: lambda view, value: str(value)})
    column_exclude_list = (
        "updated_at",
        "updated_by",
    )
    form_overrides = dict(start_date=PartialDateField, end_date=PartialDateField)
    form_widget_args = {c: {"readonly": True} for c in column_exclude_list}

    def __init__(self, model=None, *args, **kwargs):
        """Pick the model based on the ModelView class name assuming it is ModelClass + "Admin"."""
        if model is None:
            if hasattr(self, "model"):
                model = self.model
            else:
                model_class_name = self.__class__.__name__.replace("Admin", '')
                model = globals().get(model_class_name)
            if model is None:
                if model_class_name not in dir(models):
                    raise Exception(f"Model class {model_class_name} doesn't exit.")
                model = models.__dict__.get(model_class_name)
        super().__init__(model, *args, **kwargs)

    # TODO: remove whent it gets merged into the upsteem repo (it's a workaround to make
    # joins LEFT OUTERE)
    def _handle_join(self, query, field, joins):
        if field.model_class != self.model:
            model_name = field.model_class.__name__

            if model_name not in joins:
                query = query.join(field.model_class, "LEFT OUTER")
                joins.add(model_name)

        return query

    def get_pk_value(self, model):
        """Get correct value for composite keys."""
        if self.model._meta.composite_key:
            return tuple([
                model._data[field_name] for field_name in self.model._meta.primary_key.field_names
            ])
        return super().get_pk_value(model)

    def get_one(self, id):
        """Fix for composite keys."""
        if self.model._meta.composite_key:
            return self.model.get(**dict(zip(self.model._meta.primary_key.field_names, id)))
        return super().get_one(id)

    def init_search(self):
        """Include linked columns in the search if they are defined with 'liked_table.column'."""
        if self.column_searchable_list:
            for p in self.column_searchable_list:
                if "." in p:
                    m, p = p.split('.')
                    m = getattr(self.model, m).rel_model
                    p = getattr(m, p)

                elif isinstance(p, str):
                    p = getattr(self.model, p)

                # Check type
                if not isinstance(p, (
                        CharField,
                        TextField,
                )):
                    raise Exception('Can only search on text columns. ' +
                                    'Failed to setup search for "%s"' % p)

                self._search_fields.append(p)

        return bool(self._search_fields)

    def is_accessible(self):
        """Verify if the view is accessible for the current user."""
        if not current_user.is_active or not current_user.is_authenticated:
            return False

        if current_user.has_role(self.roles_required):
            return True

        return False

    def inaccessible_callback(self, name, **kwargs):
        """Handle access denial. Redirect to login page if user doesn"t have access."""
        return redirect(url_for("login", next=request.url))

    def get_query(self):
        """Add URL query to the data select for foreign key and select data that user has access to."""
        query = super().get_query()

        if current_user and not current_user.has_role(Role.SUPERUSER) and current_user.has_role(
                Role.ADMIN):
            # Show only rows realted to the curretn organisation the user is admin for.
            # Skip this part for SUPERUSER.
            db_columns = [c.db_column for c in self.model._meta.fields.values()]
            if "org_id" in db_columns or "organisation_id" in db_columns:
                if "org_id" in db_columns:
                    query = query.where(self.model.org_id == current_user.organisation.id)
                else:
                    query = query.where(self.model.organisation_id == current_user.organisation.id)

        if request.args and any(a.endswith("_id") for a in request.args):
            for f in self.model._meta.fields.values():
                if f.db_column.endswith("_id") and f.db_column in request.args:
                    query = query.where(f == int(request.args[f.db_column]))
        return query

    def _get_list_extra_args(self):
        """Workaournd for https://github.com/flask-admin/flask-admin/issues/1512."""
        view_args = super()._get_list_extra_args()
        extra_args = {
            k: v
            for k, v in request.args.items()
            if k not in (
                'page',
                'page_size',
                'sort',
                'desc',
                'search',
            ) and not k.startswith('flt')
        }
        view_args.extra_args = extra_args
        return view_args


class UserAdmin(AppModelView):
    """User model view."""

    edit_template = "admin/user_edit.html"
    roles = {1: "Superuser", 2: "Administrator", 4: "Researcher", 8: "Technical Contact"}

    form_extra_fields = dict(is_superuser=BooleanField("Is Superuser"))
    form_excluded_columns = (
        "roles",
        "created_at",
        "updated_at",
        "created_by",
        "updated_by",
    )
    column_exclude_list = (
        "password",
        "username",
        "first_name",
        "last_name",
    )
    column_formatters = dict(
        roles=lambda v, c, m, p: ", ".join(n for r, n in v.roles.items() if r & m.roles),
        orcid=lambda v, c, m, p: m.orcid.replace("-", "\u2011") if m.orcid else "")
    column_searchable_list = (
        "name",
        "orcid",
        "email",
        "eppn",
        "organisation.name",
    )
    form_overrides = dict(roles=BitmapMultipleValueField)
    form_args = dict(roles=dict(choices=roles.items()))

    form_ajax_refs = {
        "organisation": {
            "fields": (Organisation.name, "name")
        },
    }
    can_export = True

    def update_model(self, form, model):
        """Added prevalidation of the form."""
        if "roles" not in self.form_excluded_columns and form.roles.data != model.roles:
            if bool(form.roles.data & Role.ADMIN) != UserOrg.select().where(
                (UserOrg.user_id == model.id) & UserOrg.is_admin).exists():  # noqa: E125
                if form.roles.data & Role.ADMIN:
                    flash(f"Cannot add ADMIN role to {model} "
                          "since there is no organisation the user is an administrator for.",
                          "danger")
                else:
                    flash(f"Cannot revoke ADMIN role from {model} "
                          "since there is an organisation the user is an administrator for.",
                          "danger")
                form.roles.data = model.roles
                return False
            if bool(form.roles.data & Role.TECHNICAL) != Organisation.select().where(
                    Organisation.tech_contact_id == model.id).exists():
                if model.has_role(Role.TECHNICAL):
                    flash(f"Cannot revoke TECHNICAL role from {model} "
                          "since there is an organisation the user is the technical contact for.",
                          "danger")
                else:
                    flash(f"Cannot add TECHNICAL role to {model} "
                          "since there is no organisation the user is the technical contact for.",
                          "danger")
                form.roles.data = model.roles
                return False

        return super().update_model(form, model)


class OrganisationAdmin(AppModelView):
    """Organisation model view."""

    column_exclude_list = ("orcid_client_id", "orcid_secret", "created_at")
    column_searchable_list = (
        "name",
        "tuakiri_name",
        "city",
    )<commit_msg>addded logo support<commit_after># -*- coding: utf-8 -*-
"""Application views."""

import json
import os
import secrets
from collections import namedtuple
from datetime import datetime
from io import BytesIO

from flask import abort, flash, jsonify, redirect, render_template, request, send_file, send_from_directory, url_for
from flask_admin.actions import action
from flask_admin.contrib.peewee import ModelView
from flask_admin.form import SecureForm
from flask_admin.model import typefmt
from flask_login import current_user, login_required
from jinja2 import Markup
from playhouse.shortcuts import model_to_dict
from werkzeug.utils import secure_filename
from wtforms.fields import BooleanField

from orcid_api.rest import ApiException

from . import admin, app, models, orcid_client, utils
from .config import ORCID_BASE_URL, SCOPE_ACTIVITIES_UPDATE, SCOPE_READ_LIMITED
from .forms import (ApplicationFrom, BitmapMultipleValueField, CredentialForm, EmailTemplateForm,
                    FileUploadForm, JsonOrYamlFileUploadForm, LogoForm, OrgRegistrationForm,
                    PartialDateField, RecordForm, UserInvitationForm)
from .login_provider import roles_required
from .models import (Affiliation, AffiliationRecord, CharField, Client, File, FundingContributor,
                     FundingRecord, Grant, ModelException, OrcidApiCall, OrcidToken, Organisation,
                     OrgInfo, OrgInvitation, PartialDate, Role, Task, TextField, Token, Url, User,
                     UserInvitation, UserOrg, UserOrgAffiliation, db)
# NB! Should be disabled in production
from .pyinfo import info
from .utils import generate_confirmation_token, send_user_invitation

HEADERS = {"Accept": "application/vnd.orcid+json", "Content-type": "application/vnd.orcid+json"}


@app.route("/favicon.ico")
def favicon():
    """Support for the "favicon" legacy: faveicon location in the root directory."""
    return send_from_directory(
        os.path.join(app.root_path, "static", "images"),
        "favicon.ico",
        mimetype="image/vnd.microsoft.icon")


@app.route("/pyinfo")
@roles_required(Role.SUPERUSER)
def pyinfo():
    """Show Python and runtime environment and settings."""
    return render_template("pyinfo.html", **info)


@app.route("/u/<short_id>")
def short_url(short_id):
    """Redirect to the full URL."""
    try:
        u = Url.get(short_id=short_id)
        if request.args:
            return redirect(utils.append_qs(u.url, **request.args))
        return redirect(u.url)
    except Url.DoesNotExist:
        abort(404)


def read_uploaded_file(form):
    """Read up the whole content and deconde it and return the whole content."""
    raw = request.files[form.file_.name].read()
    for encoding in "utf-8", "utf-8-sig", "utf-16":
        try:
            return raw.decode(encoding)
        except UnicodeDecodeError:
            continue
    return raw.decode("latin-1")


class AppModelView(ModelView):
    """ModelView customization."""

    roles_required = Role.SUPERUSER
    export_types = [
        "csv",
        "xls",
        "tsv",
        "yaml",
        "json",
        "xlsx",
        "ods",
        "html",
    ]
    form_base_class = SecureForm
    column_type_formatters = dict(typefmt.BASE_FORMATTERS)
    column_type_formatters.update({
        datetime:
        lambda view, value: Markup(value.strftime("%Y‑%m‑%d&nbsp;%H:%M")),
    })
    column_type_formatters_export = dict(typefmt.EXPORT_FORMATTERS)
    column_type_formatters_export.update({PartialDate: lambda view, value: str(value)})
    column_exclude_list = (
        "updated_at",
        "updated_by",
    )
    form_overrides = dict(start_date=PartialDateField, end_date=PartialDateField)
    form_widget_args = {c: {"readonly": True} for c in column_exclude_list}

    def __init__(self, model=None, *args, **kwargs):
        """Pick the model based on the ModelView class name assuming it is ModelClass + "Admin"."""
        if model is None:
            if hasattr(self, "model"):
                model = self.model
            else:
                model_class_name = self.__class__.__name__.replace("Admin", '')
                model = globals().get(model_class_name)
            if model is None:
                if model_class_name not in dir(models):
                    raise Exception(f"Model class {model_class_name} doesn't exit.")
                model = models.__dict__.get(model_class_name)
        super().__init__(model, *args, **kwargs)

    # TODO: remove whent it gets merged into the upsteem repo (it's a workaround to make
    # joins LEFT OUTERE)
    def _handle_join(self, query, field, joins):
        if field.model_class != self.model:
            model_name = field.model_class.__name__

            if model_name not in joins:
                query = query.join(field.model_class, "LEFT OUTER")
                joins.add(model_name)

        return query

    def get_pk_value(self, model):
        """Get correct value for composite keys."""
        if self.model._meta.composite_key:
            return tuple([
                model._data[field_name] for field_name in self.model._meta.primary_key.field_names
            ])
        return super().get_pk_value(model)

    def get_one(self, id):
        """Fix for composite keys."""
        if self.model._meta.composite_key:
            return self.model.get(**dict(zip(self.model._meta.primary_key.field_names, id)))
        return super().get_one(id)

    def init_search(self):
        """Include linked columns in the search if they are defined with 'liked_table.column'."""
        if self.column_searchable_list:
            for p in self.column_searchable_list:
                if "." in p:
                    m, p = p.split('.')
                    m = getattr(self.model, m).rel_model
                    p = getattr(m, p)

                elif isinstance(p, str):
                    p = getattr(self.model, p)

                # Check type
                if not isinstance(p, (
                        CharField,
                        TextField,
                )):
                    raise Exception('Can only search on text columns. ' +
                                    'Failed to setup search for "%s"' % p)

                self._search_fields.append(p)

        return bool(self._search_fields)

    def is_accessible(self):
        """Verify if the view is accessible for the current user."""
        if not current_user.is_active or not current_user.is_authenticated:
            return False

        if current_user.has_role(self.roles_required):
            return True

        return False

    def inaccessible_callback(self, name, **kwargs):
        """Handle access denial. Redirect to login page if user doesn"t have access."""
        return redirect(url_for("login", next=request.url))

    def get_query(self):
        """Add URL query to the data select for foreign key and select data that user has access to."""
        query = super().get_query()

        if current_user and not current_user.has_role(Role.SUPERUSER) and current_user.has_role(
                Role.ADMIN):
            # Show only rows realted to the curretn organisation the user is admin for.
            # Skip this part for SUPERUSER.
            db_columns = [c.db_column for c in self.model._meta.fields.values()]
            if "org_id" in db_columns or "organisation_id" in db_columns:
                if "org_id" in db_columns:
                    query = query.where(self.model.org_id == current_user.organisation.id)
                else:
                    query = query.where(self.model.organisation_id == current_user.organisation.id)

        if request.args and any(a.endswith("_id") for a in request.args):
            for f in self.model._meta.fields.values():
                if f.db_column.endswith("_id") and f.db_column in request.args:
                    query = query.where(f == int(request.args[f.db_column]))
        return query

    def _get_list_extra_args(self):
        """Workaournd for https://github.com/flask-admin/flask-admin/issues/1512."""
        view_args = super()._get_list_extra_args()
        extra_args = {
            k: v
            for k, v in request.args.items()
            if k not in (
                'page',
                'page_size',
                'sort',
                'desc',
                'search',
            ) and not k.startswith('flt')
        }
        view_args.extra_args = extra_args
        return view_args


class UserAdmin(AppModelView):
    """User model view."""

    edit_template = "admin/user_edit.html"
    roles = {1: "Superuser", 2: "Administrator", 4: "Researcher", 8: "Technical Contact"}

    form_extra_fields = dict(is_superuser=BooleanField("Is Superuser"))
    form_excluded_columns = (
        "roles",
        "created_at",
        "updated_at",
        "created_by",
        "updated_by",
    )
    column_exclude_list = (
        "password",
        "username",
        "first_name",
        "last_name",
    )
    column_formatters = dict(
        roles=lambda v, c, m, p: ", ".join(n for r, n in v.roles.items() if r & m.roles),
        orcid=lambda v, c, m, p: m.orcid.replace("-", "\u2011") if m.orcid else "")
    column_searchable_list = (
        "name",
        "orcid",
        "email",
        "eppn",
        "organisation.name",
    )
    form_overrides = dict(roles=BitmapMultipleValueField)
    form_args = dict(roles=dict(choices=roles.items()))

    form_ajax_refs = {
        "organisation": {
            "fields": (Organisation.name, "name")
        },
    }
    can_export = True

    def update_model(self, form, model):
        """Added prevalidation of the form."""
        if "roles" not in self.form_excluded_columns and form.roles.data != model.roles:
            if bool(form.roles.data & Role.ADMIN) != UserOrg.select().where(
                (UserOrg.user_id == model.id) & UserOrg.is_admin).exists():  # noqa: E125
                if form.roles.data & Role.ADMIN:
                    flash(f"Cannot add ADMIN role to {model} "
                          "since there is no organisation the user is an administrator for.",
                          "danger")
                else:
                    flash(f"Cannot revoke ADMIN role from {model} "
                          "since there is an organisation the user is an administrator for.",
                          "danger")
                form.roles.data = model.roles
                return False
            if bool(form.roles.data & Role.TECHNICAL) != Organisation.select().where(
                    Organisation.tech_contact_id == model.id).exists():
                if model.has_role(Role.TECHNICAL):
                    flash(f"Cannot revoke TECHNICAL role from {model} "
                          "since there is an organisation the user is the technical contact for.",
                          "danger")
                else:
                    flash(f"Cannot add TECHNICAL role to {model} "
                          "since there is no organisation the user is the technical contact for.",
                          "danger")
                form.roles.data = model.roles
                return False

        return super().update_model(form, model)


class OrganisationAdmin(AppModelView):
    """Organisation model view."""

    column_exclude_list = ("orcid_client_id", "orcid_secret", "created_at", "logo", )
    column_searchable_list = (
        "name",
        "tuakiri_name",
        "city",
    )
{'fraction_non_alphanumeric': 0.1029718456725756, 'fraction_numerical': 0.0021724018074383037, 'mean_word_length': 2.2097336494212803, 'pattern_counts': {'":': 12, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '67239', 'n_tokens_mistral': 6770, 'n_tokens_neox': 6455, 'n_words': 1723}