app-MAIL-temp/app/dashboard/views/alias_contact_manager.py

from dataclasses import dataclass
from operator import or_
from typing import Optional

from flask import render_template, request, redirect, flash
from flask import url_for
from flask_login import login_required, current_user
from flask_wtf import FlaskForm
from sqlalchemy import and_, func, case
from wtforms import StringField, validators, ValidationError

# Need to import directly from config to allow modification from the tests
from app import config, parallel_limiter, contact_utils
from app.alias_audit_log_utils import emit_alias_audit_log, AliasAuditLogAction
from app.contact_utils import ContactCreateError
from app.dashboard.base import dashboard_bp
from app.db import Session
from app.email_validation import is_valid_email
from app.errors import (
    CannotCreateContactForReverseAlias,
    ErrContactErrorUpgradeNeeded,
    ErrAddressInvalid,
    ErrContactAlreadyExists,
)
from app.log import LOG
from app.models import Alias, Contact, EmailLog
from app.utils import CSRFValidationForm


def email_validator():
    """validate email address. Handle both only email and email with name:
    - ab@cd.com
    - AB CD <ab@cd.com>

    """
    message = "Invalid email format. Email must be either email@example.com or *First Last <email@example.com>*"

    def _check(form, field):
        email = field.data
        email = email.strip()
        email_part = email

        if "<" in email and ">" in email:
            if email.find("<") + 1 < email.find(">"):
                email_part = email[email.find("<") + 1 : email.find(">")].strip()

        if not is_valid_email(email_part):
            raise ValidationError(message)

    return _check


def create_contact(alias: Alias, contact_address: str) -> Contact:
    """
    Create a contact for a user. Can be restricted for new free users by enabling DISABLE_CREATE_CONTACTS_FOR_FREE_USERS.
    Can throw exceptions:
     - ErrAddressInvalid
     - ErrContactAlreadyExists
     - ErrContactUpgradeNeeded - If DISABLE_CREATE_CONTACTS_FOR_FREE_USERS this exception will be raised for new free users
    """
    if not contact_address:
        raise ErrAddressInvalid("Empty address")
    output = contact_utils.create_contact(email=contact_address, alias=alias)
    if output.error == ContactCreateError.InvalidEmail:
        raise ErrAddressInvalid(contact_address)
    elif output.error == ContactCreateError.NotAllowed:
        raise ErrContactErrorUpgradeNeeded()
    elif output.error is not None:
        raise ErrAddressInvalid("Invalid address")
    elif not output.created:
        raise ErrContactAlreadyExists(output.contact)

    contact = output.contact
    LOG.d(
        "create reverse-alias for %s %s, reverse alias:%s",
        contact_address,
        alias,
        contact.reply_email,
    )

    return contact


class NewContactForm(FlaskForm):
    email = StringField(
        "Email", validators=[validators.DataRequired(), email_validator()]
    )


@dataclass
class ContactInfo(object):
    contact: Contact

    nb_forward: int
    nb_reply: int

    latest_email_log: EmailLog


def get_contact_infos(
    alias: Alias, page=0, contact_id=None, query: str = ""
) -> [ContactInfo]:
    """if contact_id is set, only return the contact info for this contact"""
    sub = (
        Session.query(
            Contact.id,
            func.sum(case([(EmailLog.is_reply, 1)], else_=0)).label("nb_reply"),
            func.sum(
                case(
                    [
                        (
                            and_(
                                EmailLog.is_reply.is_(False),
                                EmailLog.blocked.is_(False),
                            ),
                            1,
                        )
                    ],
                    else_=0,
                )
            ).label("nb_forward"),
            func.max(EmailLog.created_at).label("max_email_log_created_at"),
        )
        .join(
            EmailLog,
            EmailLog.contact_id == Contact.id,
            isouter=True,
        )
        .filter(Contact.alias_id == alias.id)
        .group_by(Contact.id)
        .subquery()
    )

    q = (
        Session.query(
            Contact,
            EmailLog,
            sub.c.nb_reply,
            sub.c.nb_forward,
        )
        .join(
            EmailLog,
            EmailLog.contact_id == Contact.id,
            isouter=True,
        )
        .filter(Contact.alias_id == alias.id)
        .filter(Contact.id == sub.c.id)
        .filter(
            or_(
                EmailLog.created_at == sub.c.max_email_log_created_at,
                # no email log yet for this contact
                sub.c.max_email_log_created_at.is_(None),
            )
        )
    )

    if query:
        q = q.filter(
            or_(
                Contact.website_email.ilike(f"%{query}%"),
                Contact.name.ilike(f"%{query}%"),
            )
        )

    if contact_id:
        q = q.filter(Contact.id == contact_id)

    latest_activity = case(
        [
            (EmailLog.created_at > Contact.created_at, EmailLog.created_at),
            (EmailLog.created_at < Contact.created_at, Contact.created_at),
        ],
        else_=Contact.created_at,
    )
    q = (
        q.order_by(latest_activity.desc())
        .limit(config.PAGE_LIMIT)
        .offset(page * config.PAGE_LIMIT)
    )

    ret = []
    for contact, latest_email_log, nb_reply, nb_forward in q:
        contact_info = ContactInfo(
            contact=contact,
            nb_forward=nb_forward,
            nb_reply=nb_reply,
            latest_email_log=latest_email_log,
        )
        ret.append(contact_info)

    return ret


def delete_contact(alias: Alias, contact_id: int):
    contact: Optional[Contact] = Contact.get(contact_id)

    if not contact:
        flash("Unknown error. Refresh the page", "warning")
    elif contact.alias_id != alias.id:
        flash("You cannot delete reverse-alias", "warning")
    else:
        delete_contact_email = contact.website_email
        emit_alias_audit_log(
            alias=alias,
            action=AliasAuditLogAction.DeleteContact,
            message=f"Delete contact {contact_id} ({contact.email})",
        )
        Contact.delete(contact_id)
        Session.commit()

        flash(f"Reverse-alias for {delete_contact_email} has been deleted", "success")


@dashboard_bp.route("/alias_contact_manager/<int:alias_id>/", methods=["GET", "POST"])
@login_required
@parallel_limiter.lock(name="contact_creation")
def alias_contact_manager(alias_id):
    highlight_contact_id = None
    if request.args.get("highlight_contact_id"):
        try:
            highlight_contact_id = int(request.args.get("highlight_contact_id"))
        except ValueError:
            flash("Invalid contact id", "error")
            return redirect(url_for("dashboard.index"))

    alias = Alias.get(alias_id)

    page = 0
    if request.args.get("page"):
        page = int(request.args.get("page"))

    query = request.args.get("query") or ""

    # sanity check
    if not alias:
        flash("You do not have access to this page", "warning")
        return redirect(url_for("dashboard.index"))

    if alias.user_id != current_user.id:
        flash("You do not have access to this page", "warning")
        return redirect(url_for("dashboard.index"))

    new_contact_form = NewContactForm()
    csrf_form = CSRFValidationForm()

    if request.method == "POST":
        if not csrf_form.validate():
            flash("Invalid request", "warning")
            return redirect(request.url)
        if request.form.get("form-name") == "create":
            if new_contact_form.validate():
                contact_address = new_contact_form.email.data.strip()
                try:
                    contact = create_contact(alias, contact_address)
                except (
                    ErrContactErrorUpgradeNeeded,
                    ErrAddressInvalid,
                    ErrContactAlreadyExists,
                    CannotCreateContactForReverseAlias,
                ) as excp:
                    flash(excp.error_for_user(), "error")
                    return redirect(request.url)
                flash(f"Reverse alias for {contact_address} is created", "success")
                return redirect(
                    url_for(
                        "dashboard.alias_contact_manager",
                        alias_id=alias_id,
                        highlight_contact_id=contact.id,
                    )
                )
        elif request.form.get("form-name") == "delete":
            contact_id = request.form.get("contact-id")
            delete_contact(alias, contact_id)
            return redirect(
                url_for("dashboard.alias_contact_manager", alias_id=alias_id)
            )

        elif request.form.get("form-name") == "search":
            query = request.form.get("query")
            return redirect(
                url_for(
                    "dashboard.alias_contact_manager",
                    alias_id=alias_id,
                    query=query,
                    highlight_contact_id=highlight_contact_id,
                )
            )

    contact_infos = get_contact_infos(alias, page, query=query)
    last_page = len(contact_infos) < config.PAGE_LIMIT
    nb_contact = Contact.filter(Contact.alias_id == alias.id).count()

    # if highlighted contact isn't included, fetch it
    # make sure highlighted contact is at array start
    contact_ids = [contact_info.contact.id for contact_info in contact_infos]
    if highlight_contact_id and highlight_contact_id not in contact_ids:
        contact_infos = (
            get_contact_infos(alias, contact_id=highlight_contact_id, query=query)
            + contact_infos
        )

    return render_template(
        "dashboard/alias_contact_manager.html",
        contact_infos=contact_infos,
        alias=alias,
        new_contact_form=new_contact_form,
        highlight_contact_id=highlight_contact_id,
        page=page,
        last_page=last_page,
        query=query,
        nb_contact=nb_contact,
        can_create_contacts=current_user.can_create_contacts(),
        csrf_form=csrf_form,
    )