1 | """ |
||
2 | byceps.services.shop.order.email.service |
||
3 | ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ |
||
4 | |||
5 | Notification e-mails about shop orders |
||
6 | |||
7 | :Copyright: 2006-2021 Jochen Kupperschmidt |
||
8 | :License: Revised BSD (see `LICENSE` file for details) |
||
9 | """ |
||
10 | |||
11 | 1 | from __future__ import annotations |
|
12 | 1 | from dataclasses import dataclass |
|
13 | 1 | from typing import Iterator |
|
14 | 1 | ||
15 | from flask_babel import gettext |
||
16 | 1 | ||
17 | 1 | from .....services.email import service as email_service |
|
18 | 1 | from .....services.email.transfer.models import Message |
|
19 | from .....services.shop.order import service as order_service |
||
20 | 1 | from .....services.shop.order.transfer.models import Order, OrderID |
|
21 | 1 | from .....services.shop.shop import service as shop_service |
|
22 | 1 | from .....services.snippet import service as snippet_service |
|
23 | 1 | from .....services.snippet.service import SnippetNotFound |
|
24 | 1 | from .....services.snippet.transfer.models import Scope |
|
25 | 1 | from .....services.user import service as user_service |
|
26 | 1 | from .....services.user.transfer.models import User |
|
27 | 1 | from .....typing import BrandID |
|
28 | 1 | from .....util.datetime.timezone import utc_to_local_tz |
|
29 | 1 | from .....util.l10n import force_user_locale |
|
30 | 1 | from .....util.money import format_euro_amount |
|
31 | 1 | from .....util.templating import load_template |
|
32 | 1 | ||
33 | from ...shop.transfer.models import ShopID |
||
34 | 1 | ||
35 | |||
36 | @dataclass(frozen=True) |
||
37 | 1 | class OrderEmailData: |
|
38 | order: Order |
||
39 | 1 | brand_id: BrandID |
|
40 | 1 | orderer: User |
|
41 | 1 | orderer_screen_name: str |
|
42 | 1 | orderer_email_address: str |
|
43 | |||
44 | |||
45 | 1 | def send_email_for_incoming_order_to_orderer(order_id: OrderID) -> None: |
|
46 | 1 | data = _get_order_email_data(order_id) |
|
47 | |||
48 | 1 | with force_user_locale(data.orderer): |
|
49 | message = _assemble_email_for_incoming_order_to_orderer(data) |
||
50 | 1 | ||
51 | _send_email(message) |
||
52 | |||
53 | 1 | ||
54 | 1 | def send_email_for_canceled_order_to_orderer(order_id: OrderID) -> None: |
|
55 | data = _get_order_email_data(order_id) |
||
56 | 1 | ||
57 | with force_user_locale(data.orderer): |
||
58 | 1 | message = _assemble_email_for_canceled_order_to_orderer(data) |
|
59 | |||
60 | _send_email(message) |
||
61 | 1 | ||
62 | 1 | ||
63 | def send_email_for_paid_order_to_orderer(order_id: OrderID) -> None: |
||
64 | 1 | data = _get_order_email_data(order_id) |
|
65 | |||
66 | 1 | with force_user_locale(data.orderer): |
|
67 | message = _assemble_email_for_paid_order_to_orderer(data) |
||
68 | |||
69 | 1 | _send_email(message) |
|
70 | |||
71 | |||
72 | 1 | def _assemble_email_for_incoming_order_to_orderer( |
|
73 | data: OrderEmailData, |
||
74 | 1 | ) -> Message: |
|
75 | order = data.order |
||
76 | order_number = order.order_number |
||
77 | |||
78 | 1 | subject = gettext( |
|
79 | 1 | 'Your order (%(order_number)s) has been received.', |
|
80 | 1 | order_number=order_number, |
|
81 | 1 | ) |
|
82 | |||
83 | 1 | date_str = _get_order_date_str(order) |
|
84 | indentation = ' ' |
||
85 | line_items = [ |
||
86 | '\n'.join( |
||
87 | [ |
||
88 | indentation |
||
89 | + gettext('Description') |
||
90 | + ': ' |
||
91 | + line_item.description, |
||
92 | 1 | indentation + gettext('Quantity') + ': ' + str(line_item.quantity), |
|
93 | 1 | indentation |
|
94 | + gettext('Unit price') |
||
95 | 1 | + ': ' |
|
96 | 1 | + format_euro_amount(line_item.unit_price), |
|
97 | ] |
||
98 | ) |
||
99 | for line_item in sorted(order.line_items, key=lambda li: li.description) |
||
100 | ] |
||
101 | total_amount = ( |
||
102 | 1 | indentation |
|
103 | + gettext('Total amount') |
||
104 | + ': ' |
||
105 | 1 | + format_euro_amount(order.total_amount) |
|
106 | ) |
||
107 | payment_instructions = _get_payment_instructions(order) |
||
108 | paragraphs = [ |
||
109 | 1 | gettext( |
|
110 | 1 | 'thank you for your order %(order_number)s on %(order_date)s through our website.', |
|
111 | 1 | order_number=order_number, |
|
112 | order_date=date_str, |
||
113 | 1 | ), |
|
114 | gettext('You have ordered these items:'), |
||
115 | *line_items, |
||
116 | total_amount, |
||
117 | payment_instructions, |
||
118 | ] |
||
119 | body = _assemble_body(data, paragraphs) |
||
120 | |||
121 | recipient_address = data.orderer_email_address |
||
122 | 1 | ||
123 | 1 | return _assemble_email_to_orderer( |
|
124 | subject, body, data.brand_id, recipient_address |
||
125 | ) |
||
126 | |||
127 | 1 | ||
128 | 1 | def _get_payment_instructions(order: Order) -> str: |
|
129 | 1 | fragment = _get_snippet_body(order.shop_id, 'email_payment_instructions') |
|
130 | |||
131 | 1 | template = load_template(fragment) |
|
132 | return template.render( |
||
133 | order_id=order.id, |
||
134 | order_number=order.order_number, |
||
135 | ) |
||
136 | |||
137 | |||
138 | View Code Duplication | def _assemble_email_for_canceled_order_to_orderer( |
|
1 ignored issue
–
show
Duplication
introduced
by
Loading history...
|
|||
139 | data: OrderEmailData, |
||
140 | 1 | ) -> Message: |
|
141 | order = data.order |
||
142 | 1 | order_number = order.order_number |
|
143 | |||
144 | 1 | subject = '\u274c ' + gettext( |
|
145 | 1 | 'Your order (%(order_number)s) has been canceled.', |
|
146 | 1 | order_number=order_number, |
|
147 | 1 | ) |
|
148 | |||
149 | 1 | date_str = _get_order_date_str(order) |
|
150 | cancelation_reason = order.cancelation_reason or '' |
||
151 | paragraphs = [ |
||
152 | gettext( |
||
153 | 'your order %(order_number)s on %(order_date)s has been canceled by us for this reason:', |
||
154 | order_number=order_number, |
||
155 | order_date=date_str, |
||
156 | ), |
||
157 | 1 | cancelation_reason, |
|
158 | ] |
||
159 | 1 | body = _assemble_body(data, paragraphs) |
|
160 | |||
161 | 1 | recipient_address = data.orderer_email_address |
|
162 | |||
163 | return _assemble_email_to_orderer( |
||
164 | subject, body, data.brand_id, recipient_address |
||
165 | ) |
||
166 | |||
167 | |||
168 | 1 | View Code Duplication | def _assemble_email_for_paid_order_to_orderer(data: OrderEmailData) -> Message: |
1 ignored issue
–
show
|
|||
169 | 1 | order = data.order |
|
170 | order_number = order.order_number |
||
171 | 1 | ||
172 | 1 | subject = '\u2705 ' + gettext( |
|
173 | 'Your order (%(order_number)s) has been paid.', |
||
174 | order_number=order_number, |
||
175 | 1 | ) |
|
176 | |||
177 | date_str = _get_order_date_str(order) |
||
178 | paragraphs = [ |
||
179 | gettext( |
||
180 | 'thank you for your order %(order_number)s on %(order_date)s through our website.', |
||
181 | order_number=order_number, |
||
182 | order_date=date_str, |
||
183 | 1 | ), |
|
184 | 1 | gettext( |
|
185 | 1 | 'We have received your payment and have marked your order as paid.' |
|
186 | 1 | ), |
|
187 | ] |
||
188 | 1 | body = _assemble_body(data, paragraphs) |
|
189 | |||
190 | recipient_address = data.orderer_email_address |
||
191 | 1 | ||
192 | 1 | return _assemble_email_to_orderer( |
|
193 | subject, body, data.brand_id, recipient_address |
||
194 | 1 | ) |
|
195 | |||
196 | |||
197 | def _get_order_email_data(order_id: OrderID) -> OrderEmailData: |
||
198 | 1 | """Collect data required for an order e-mail template.""" |
|
199 | order = order_service.get_order(order_id) |
||
200 | |||
201 | 1 | shop = shop_service.get_shop(order.shop_id) |
|
202 | orderer_id = order.placed_by_id |
||
203 | orderer = user_service.get_user(orderer_id) |
||
204 | 1 | screen_name = orderer.screen_name or 'UnknownUser' |
|
205 | 1 | email_address = user_service.get_email_address(orderer_id) |
|
206 | |||
207 | return OrderEmailData( |
||
208 | order=order, |
||
209 | brand_id=shop.brand_id, |
||
210 | orderer=orderer, |
||
211 | orderer_screen_name=screen_name, |
||
212 | orderer_email_address=email_address, |
||
213 | ) |
||
214 | 1 | ||
215 | |||
216 | 1 | def _assemble_body(data: OrderEmailData, paragraphs: list[str]) -> str: |
|
217 | 1 | """Assemble the plain text part of the email.""" |
|
218 | 1 | salutation = gettext( |
|
219 | 'Hello %(screen_name)s,', screen_name=data.orderer_screen_name |
||
220 | 1 | ) |
|
221 | footer = _get_snippet_body(data.order.shop_id, 'email_footer') |
||
222 | 1 | ||
223 | return '\n\n'.join([salutation] + paragraphs + [footer]) |
||
224 | |||
225 | 1 | ||
226 | 1 | def _assemble_email_to_orderer( |
|
227 | subject: str, |
||
228 | body: str, |
||
229 | brand_id: BrandID, |
||
230 | recipient_address: str, |
||
231 | ) -> Message: |
||
232 | """Assemble an email message with the rendered template as its body.""" |
||
233 | config = email_service.get_config(brand_id) |
||
234 | sender = config.sender |
||
235 | recipients = [recipient_address] |
||
236 | |||
237 | return Message(sender, recipients, subject, body) |
||
238 | |||
239 | |||
240 | def _get_order_date_str(order: Order) -> str: |
||
241 | return utc_to_local_tz(order.created_at).strftime('%d.%m.%Y') |
||
242 | |||
243 | |||
244 | def _get_snippet_body(shop_id: ShopID, name: str) -> str: |
||
245 | scope = Scope('shop', str(shop_id)) |
||
246 | |||
247 | version = snippet_service.find_current_version_of_snippet_with_name( |
||
248 | scope, name |
||
249 | ) |
||
250 | |||
251 | if not version: |
||
252 | raise SnippetNotFound(scope, name) |
||
253 | |||
254 | return version.body.strip() |
||
255 | |||
256 | |||
257 | def _send_email(message: Message) -> None: |
||
258 | email_service.enqueue_message(message) |
||
259 |