import re import markdown from jinja2 import Template from crc import app from crc.api.common import ApiError from crc.scripts.script import Script from crc.services.ldap_service import LdapService from crc.services.email_service import EmailService from flask import render_template, request class Email(Script): """This Script allows to be introduced as part of a workflow and called from there, specifying recipients and content """ def get_description(self): return """ Creates an email, using the provided arguments. The first argument is the subject of the email, all subsequent arguments should be email addresses in quotes, or variables containing an email address or a list of email addresses." The "documentation" should contain markdown that will become the body of the email message. Example: email ("My Subject", "dhf8r@virginia.edu", pi.email) """ def do_task_validate_only(self, task, study_id, workflow_id, *args, **kwargs): self.get_subject(args) self.get_email_recipients(task, args) self.get_content(task) def do_task(self, task, study_id, workflow_id, *args, **kwargs): if len(args) < 2: raise ApiError(code="missing_argument", message="Email script requires a subject and at least one email address as arguments") subject = self.get_subject(args) recipients = self.get_email_recipients(task, args) if recipients: content, content_html = self.get_content(task) EmailService.add_email( subject=subject, sender=app.config['DEFAULT_SENDER'], recipients=recipients, content=content, content_html=content_html, study_id=study_id ) def check_valid_email(self, email): # regex from https://emailregex.com/ regex = r"(^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$)" if (re.search(regex, email)): return True else: return False def get_email_recipients(self, task, args): emails = [] if len(args) < 2: raise ApiError(code="missing_argument", message="Email script requires at least one email address as an argument. " "Multiple email addresses are accepted.") # Every argument following the subject should be an email, or a list of emails. for arg in args[1:]: if isinstance(arg, str): emails_to_check = [arg] elif isinstance(arg, list): emails_to_check = arg else: raise ApiError(code="invalid_argument", message=f"Email script requires a valid email address, but received '{arg}'") for e in emails_to_check: if self.check_valid_email(e): emails.append(e) else: raise ApiError(code="invalid_argument", message="The email address you provided could not be parsed. " "The value you provided is '%s" % e) return emails @staticmethod def get_subject(args): # subject = '' if len(args[0]) < 1: raise ApiError(code="missing_argument", message="No subject was provided for the email message.") subject = args[0] if not subject or not isinstance(subject, str): raise ApiError(code="invalid_argument", message="The subject you provided could not be parsed. " "The value is \"%s\" " % subject) return subject def get_content(self, task): content = task.task_spec.documentation template = Template(content) rendered = template.render(task.data) rendered_markdown = markdown.markdown(rendered).replace('\n', '
') wrapped = self.get_cr_connect_wrapper(rendered_markdown) return rendered, wrapped @staticmethod def get_cr_connect_wrapper(email_body): return render_template('mail_content_template.html', email_body=email_body, base_url=request.base_url)