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

333 lines
10 KiB
Python
Raw Normal View History

from dataclasses import dataclass
from operator import or_
2019-12-15 16:17:37 +01:00
from flask import render_template, request, redirect, flash
from flask import url_for
2019-12-15 16:17:37 +01:00
from flask_login import login_required, current_user
from flask_wtf import FlaskForm
from sqlalchemy import and_, func, case
2019-12-15 16:17:37 +01:00
from wtforms import StringField, validators, ValidationError
2022-05-12 18:33:13 +02:00
# Need to import directly from config to allow modification from the tests
from app import config, parallel_limiter
2019-12-15 16:17:37 +01:00
from app.dashboard.base import dashboard_bp
from app.db import Session
from app.email_utils import (
generate_reply_email,
2021-09-10 17:31:29 +02:00
parse_full_address,
)
from app.email_validation import is_valid_email
from app.errors import (
CannotCreateContactForReverseAlias,
ErrContactErrorUpgradeNeeded,
ErrAddressInvalid,
ErrContactAlreadyExists,
)
2019-12-15 16:17:37 +01:00
from app.log import LOG
from app.models import Alias, Contact, EmailLog, User
from app.utils import sanitize_email, CSRFValidationForm
2019-12-15 16:17:37 +01:00
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
2020-02-18 05:54:51 +01:00
email = email.strip()
2019-12-15 16:17:37 +01:00
email_part = email
if "<" in email and ">" in email:
if email.find("<") + 1 < email.find(">"):
email_part = email[email.find("<") + 1 : email.find(">")].strip()
2020-11-03 11:11:09 +01:00
if not is_valid_email(email_part):
raise ValidationError(message)
2019-12-15 16:17:37 +01:00
return _check
2022-05-12 18:46:42 +02:00
def user_can_create_contacts(user: User) -> bool:
if user.is_premium():
return True
2022-05-12 19:02:06 +02:00
if user.flags & User.FLAG_FREE_DISABLE_CREATE_ALIAS == 0:
return True
return not config.DISABLE_CREATE_CONTACTS_FOR_FREE_USERS
2022-05-12 18:46:42 +02:00
2022-05-12 18:33:13 +02:00
def create_contact(user: User, alias: Alias, contact_address: str) -> Contact:
2022-05-12 18:36:12 +02:00
"""
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")
try:
contact_name, contact_email = parse_full_address(contact_address)
except ValueError:
raise ErrAddressInvalid(contact_address)
2022-05-12 18:42:16 +02:00
contact_email = sanitize_email(contact_email)
if not is_valid_email(contact_email):
raise ErrAddressInvalid(contact_email)
contact = Contact.get_by(alias_id=alias.id, website_email=contact_email)
if contact:
2022-05-12 18:33:13 +02:00
raise ErrContactAlreadyExists(contact)
2022-05-12 18:46:42 +02:00
if not user_can_create_contacts(user):
raise ErrContactErrorUpgradeNeeded()
contact = Contact.create(
user_id=alias.user_id,
alias_id=alias.id,
website_email=contact_email,
name=contact_name,
reply_email=generate_reply_email(contact_email, alias),
)
LOG.d(
"create reverse-alias for %s %s, reverse alias:%s",
contact_address,
alias,
contact.reply_email,
)
Session.commit()
return contact
2019-12-15 16:17:37 +01:00
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
2021-07-14 18:57:25 +02:00
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_(
2020-12-06 18:04:29 +01:00
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"),
)
2020-08-27 10:20:48 +02:00
.join(
EmailLog,
EmailLog.contact_id == Contact.id,
isouter=True,
)
.filter(Contact.alias_id == alias.id)
.group_by(Contact.id)
.subquery()
)
q = (
Session.query(
2020-08-27 10:20:48 +02:00
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
2020-12-06 18:04:29 +01:00
sub.c.max_email_log_created_at.is_(None),
)
)
)
2021-07-14 18:57:25 +02:00
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,
)
2022-05-12 18:30:46 +02:00
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 = 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
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"])
2019-12-15 16:17:37 +01:00
@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)
2019-12-15 16:17:37 +01:00
2020-05-23 22:34:46 +02:00
page = 0
if request.args.get("page"):
page = int(request.args.get("page"))
2021-07-14 18:57:25 +02:00
query = request.args.get("query") or ""
2019-12-15 16:17:37 +01:00
# sanity check
if not alias:
2019-12-15 16:17:37 +01:00
flash("You do not have access to this page", "warning")
return redirect(url_for("dashboard.index"))
if alias.user_id != current_user.id:
2019-12-15 16:17:37 +01:00
flash("You do not have access to this page", "warning")
return redirect(url_for("dashboard.index"))
new_contact_form = NewContactForm()
csrf_form = CSRFValidationForm()
2019-12-15 16:17:37 +01:00
if request.method == "POST":
if not csrf_form.validate():
flash("Invalid request", "warning")
return redirect(request.url)
2019-12-15 16:17:37 +01:00
if request.form.get("form-name") == "create":
if new_contact_form.validate():
contact_address = new_contact_form.email.data.strip()
try:
contact = create_contact(current_user, 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")
2019-12-15 16:17:37 +01:00
return redirect(
url_for(
"dashboard.alias_contact_manager",
alias_id=alias_id,
highlight_contact_id=contact.id,
2019-12-15 16:17:37 +01:00
)
)
elif request.form.get("form-name") == "delete":
2020-03-17 10:56:59 +01:00
contact_id = request.form.get("contact-id")
delete_contact(alias, contact_id)
return redirect(
url_for("dashboard.alias_contact_manager", alias_id=alias_id)
)
2019-12-15 16:17:37 +01:00
2021-07-14 18:57:25 +02:00
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)
2022-05-12 18:30:46 +02:00
last_page = len(contact_infos) < config.PAGE_LIMIT
nb_contact = Contact.filter(Contact.alias_id == alias.id).count()
2019-12-15 16:17:37 +01:00
# 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]
2020-09-14 11:38:01 +02:00
if highlight_contact_id and highlight_contact_id not in contact_ids:
contact_infos = (
2021-07-14 18:57:25 +02:00
get_contact_infos(alias, contact_id=highlight_contact_id, query=query)
+ contact_infos
)
2019-12-15 16:17:37 +01:00
return render_template(
"dashboard/alias_contact_manager.html",
contact_infos=contact_infos,
alias=alias,
2019-12-15 16:17:37 +01:00
new_contact_form=new_contact_form,
highlight_contact_id=highlight_contact_id,
2020-05-23 22:34:46 +02:00
page=page,
last_page=last_page,
2021-07-14 18:57:25 +02:00
query=query,
nb_contact=nb_contact,
2022-05-12 18:46:42 +02:00
can_create_contacts=user_can_create_contacts(current_user),
csrf_form=csrf_form,
2019-12-15 16:17:37 +01:00
)