"""
Action triggered after register action (create temp user, change password etc...)
"""
import datetime
from warnings import warn
from flask import render_template, current_app, url_for
from sqlalchemy import select
from markupsafe import Markup
from pypnusershub.db.models import Application, CorRoleToken, User
from pypnusershub.db.models_register import TempUser
from sqlalchemy.sql import func
from geonature.core.gn_commons.models import TModules
from geonature.core.gn_meta.models import (
TDatasets,
TAcquisitionFramework,
CorDatasetActor,
CorAcquisitionFrameworkActor,
)
from geonature.utils.utilsmails import send_mail
from geonature.utils.env import db, DB
[docs]
def validators_emails():
"""
Returns a list of validators emails
"""
emails = current_app.config["ACCOUNT_MANAGEMENT"]["VALIDATOR_EMAIL"]
return emails if isinstance(emails, list) else [emails]
[docs]
def validate_temp_user(data):
"""
Send an email after the action of account creation.
Parameters
----------
data : dict
Must contain the token of the temp user. Example : {"token": "iamavalidtoken"}
"""
token = data.get("token", None)
user = DB.session.scalars(select(TempUser).where(TempUser.token_role == token).limit(1)).first()
if not user:
raise ValueError(f"Token {token} is not associated to a temp user")
user_dict = user.as_dict()
subject = "Demande de création de compte GeoNature"
url_validation = f"{current_app.config['URL_APPLICATION']}/#/confirm?token={token}"
if current_app.config["ACCOUNT_MANAGEMENT"]["AUTO_ACCOUNT_CREATION"]:
template = "email_self_validate_account.html"
recipients = [user.email]
else:
template = "email_admin_validate_account.html"
recipients = current_app.config["ACCOUNT_MANAGEMENT"]["VALIDATOR_EMAIL"]
url_validation += f"&asValidator=true"
additional_fields = [
{"key": key, "value": value} for key, value in (user_dict.get("champs_addi") or {}).items()
]
msg_html = render_template(
template,
url_validation=url_validation,
user=user_dict,
additional_fields=additional_fields,
)
send_mail(recipients, subject, msg_html)
return True
[docs]
def execute_actions_after_validation(data):
"""
Execute post actions after a temp user has been validated.
If AUTO_DATASET_CREATION is set to True, create a personal JDD and AF
for the user. Then, send an email to the user to inform him of the
creation of his account.
"""
if current_app.config["ACCOUNT_MANAGEMENT"]["AUTO_DATASET_CREATION"]:
create_dataset_user(data)
inform_user(data)
return True
[docs]
def create_dataset_user(user):
"""
After dataset validation, add a personal Acquisition Framework and Dataset
so the user can add new datasets.
Parameters
----------
user : dict
The user data
"""
af_desc_and_name = "Cadre d'acquisition personnel de {name} {surname}".format(
name=user["nom_role"], surname=user["prenom_role"]
)
# actor = data productor
af_productor = CorAcquisitionFrameworkActor(
id_role=user["id_role"],
id_nomenclature_actor_role=func.ref_nomenclatures.get_id_nomenclature("ROLE_ACTEUR", "6"),
)
af_contact = CorAcquisitionFrameworkActor(
id_role=user["id_role"],
id_nomenclature_actor_role=func.ref_nomenclatures.get_id_nomenclature("ROLE_ACTEUR", "1"),
)
new_af = TAcquisitionFramework(
acquisition_framework_name=af_desc_and_name,
acquisition_framework_desc=af_desc_and_name
+ " - auto-créé via la demande de création de compte",
acquisition_framework_start_date=datetime.datetime.now(),
)
new_af.cor_af_actor = [af_productor, af_contact]
db.session.add(new_af)
ds_desc_and_name = "Jeu de données personnel de {name} {surname}".format(
name=user["nom_role"], surname=user["prenom_role"]
)
ds_productor = CorDatasetActor(
id_role=user["id_role"],
id_nomenclature_actor_role=func.ref_nomenclatures.get_id_nomenclature("ROLE_ACTEUR", "6"),
)
ds_contact = CorDatasetActor(
id_role=user["id_role"],
id_nomenclature_actor_role=func.ref_nomenclatures.get_id_nomenclature("ROLE_ACTEUR", "1"),
)
# add new JDD: terrestrial and marine = True as default
new_dataset = TDatasets(
acquisition_framework=new_af,
dataset_name=ds_desc_and_name,
dataset_shortname=ds_desc_and_name + " - auto-créé via la demande de création de compte",
dataset_desc=ds_desc_and_name,
marine_domain=True,
terrestrial_domain=True,
)
new_dataset.cor_dataset_actor = [ds_productor, ds_contact]
db.session.add(new_dataset)
for module_code in current_app.config["ACCOUNT_MANAGEMENT"]["DATASET_MODULES_ASSOCIATION"]:
# module = TModules.query.filter_by(module_code=module_code).one_or_none()
module = db.session.execute(
select(TModules).filter_by(module_code=module_code)
).scalar_one_or_none()
if module is None:
warn("Module code '{}' does not exist, can not associate dataset.".format(module_code))
continue
new_dataset.modules.append(module)
db.session.commit()
[docs]
def send_email_for_recovery(user):
"""
Send an email with the login of the role and the possibility to reset
its password
Parameters
----------
user : dict
The user data
"""
if not user:
raise KeyError(f"No user was given")
if not (token := db.session.get(CorRoleToken, user.id_role).token):
raise Exception(f"No token is associated to {user.id_role}")
msg_html = render_template(
"email_login_and_new_pass.html",
identifiant=user.identifiant,
url_password=current_app.config["URL_APPLICATION"] + "/#/login/new-password?token=" + token,
)
subject = "Confirmation changement Identifiant / mot de passe"
send_mail([user.email], subject, msg_html)
return True