import requests from generalresearch.config import is_debug from generalresearch.currency import USDCent from generalresearch.models.thl.wallet.cashout_method import ( CashoutMethodOut, CashoutRequestInfo, ) from jb.config import settings MAUTIC_BASE_URL = "https://mail.jamesbillings67.com" LOGIN_EMAIL_TEMPLATE_ID = 1 # Todo: We need a new template for the cashout confirmation CASHOUT_REQUEST_EMAIL_TEMPLATE_ID = 3 # Todo: We need a new template for the cashout status / sent CASHOUT_STATUS_EMAIL_TEMPLATE_ID = 4 assert settings.mautic_api_key auth_headers = {"Authorization": f"Basic {settings.mautic_api_key.get_secret_value()}"} def get_or_create_contact(email: str, amt_worker_id: str | None = None) -> int: body = {"email": email} if amt_worker_id: body["amt_worker_id"] = amt_worker_id res = requests.post( url=f"{MAUTIC_BASE_URL}/api/contacts/new", json=body, headers=auth_headers, ).json() return int(res["contact"]["id"]) def send_login_email_from_url(mautic_url: str, magic_link: str) -> None: if is_debug(): print("MAGIC_LINK: ", magic_link) return email_tokens = { "magic_link": magic_link, } send_email_with_tokens(email_tokens, mautic_url) def send_email_with_tokens(tokens: dict[str, str], mautic_url: str) -> None: body = {"tokens": tokens} response = requests.post(url=mautic_url, json=body, headers=auth_headers) try: response.raise_for_status() except requests.exceptions.HTTPError: print(f"Failed to send email. Status code: {response.status_code}") print(response.text) raise d = response.json() assert d.get("success"), f"Failed to send email: {d.get('failed')}" print("Email sent successfully") def send_login_email(email: str, magic_token: str) -> None: contact_id = get_or_create_contact(email=email) mautic_url = f"{MAUTIC_BASE_URL}/api/emails/{LOGIN_EMAIL_TEMPLATE_ID}/contact/{contact_id}/send" magic_link = f"{settings.base_url}auth/magic-link/?token={magic_token}" send_login_email_from_url(mautic_url, magic_link) def send_amt_link_email(email: str, magic_token: str) -> None: # Don't actually associate the email with the worker ID # until they click the link contact_id = get_or_create_contact(email=email) mautic_url = f"{MAUTIC_BASE_URL}/api/emails/{LOGIN_EMAIL_TEMPLATE_ID}/contact/{contact_id}/send" magic_link = f"{settings.base_url}auth/link-amt/?token={magic_token}" send_login_email_from_url(mautic_url, magic_link) def send_cashout_confirmation_email( email: str, token: str, cashout_method: CashoutMethodOut, amount: USDCent ) -> None: contact_id = get_or_create_contact(email=email) mautic_url = f"{MAUTIC_BASE_URL}/api/emails/{CASHOUT_REQUEST_EMAIL_TEMPLATE_ID}/contact/{contact_id}/send" magic_link = f"{settings.base_url}wallet/cashout/confirm/?token={token}" if is_debug(): print("MAGIC_LINK: ", magic_link) return email_tokens = { "magic_link": magic_link, "cashout_method_name": cashout_method.name, "cashout_method_description": cashout_method.description, "cashout_method_type": cashout_method.type.value, "amount": amount.to_usd_str(), } if cashout_method.image_url: email_tokens["image_url"] = cashout_method.image_url send_email_with_tokens(email_tokens, mautic_url) def send_cashout_status_email(email: str, cashout: CashoutRequestInfo) -> None: assert cashout.status is not None email_tokens = { "cashout_id": str(cashout.id), "cashout_status": cashout.status.value, } # todo: (if tango) format credentials and redemption instructions # "credentials": "", # "redemption_instructions": "", if is_debug(): print("CASHOUT_STATUS_EMAIL: ", email_tokens) return contact_id = get_or_create_contact(email=email) mautic_url = f"{MAUTIC_BASE_URL}/api/emails/{CASHOUT_STATUS_EMAIL_TEMPLATE_ID}/contact/{contact_id}/send" send_email_with_tokens(email_tokens, mautic_url)