1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
|
import requests
from generalresearch.config import is_debug
from generalresearch.currency import USDCent
from generalresearch.models.thl.wallet.cashout_method import CashoutMethodOut
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_EMAIL_TEMPLATE_ID = 3
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_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)
|
