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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
|
from html import escape
from typing import Any
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 format_cashout_status_email_tokens(
cashout: CashoutRequestInfo,
) -> dict[str, str]:
tx = cashout.transaction_info or {}
assert cashout.id is not None
assert cashout.status is not None
tokens = {
"cashout_id": cashout.id,
"cashout_status": cashout.status.value,
"cashout_description": cashout.description,
# From tango
"credentials": format_tango_credentials(tx),
# Tango supplies formatted HTML instructions intended for the recipient
"redemption_instructions": tx.get("redemption_instructions") or "",
# From paypal
"transaction_id": tx.get("transaction_id") or "",
"amount": USDCent(cashout.amount).to_usd_str(),
"cashout_method_type": cashout.payout_type.value,
"subject": f"Your {cashout.payout_type.value} cashout for James Billings",
}
# Cash-in-mail details are flat values in transaction_info.
for field in (
"tracking_number",
"shipping_method",
"carrier",
"ship_date",
"estimated_delivery_date",
"delivery_status",
):
tokens[field] = str(tx.get(field) or "")
return tokens
def format_tango_credentials(transaction: dict[str, Any]) -> str:
if not transaction.get("credential_list"):
return ""
rows = []
for credential in transaction["credential_list"]:
label = escape(credential["label"])
value = escape(credential["value"], quote=True)
rendered_value = (
f'<a href="{value}">{value}</a>' if credential["type"] == "url" else value
)
rows.append(f"<p><strong>{label}:</strong> {rendered_value}</p>")
return "".join(rows)
def send_cashout_status_email(email: str, cashout: CashoutRequestInfo) -> None:
email_tokens = format_cashout_status_email_tokens(cashout)
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)
|
