|
5 | 5 | from contextlib import asynccontextmanager |
6 | 6 | from email.headerregistry import Address |
7 | 7 | from email.message import EmailMessage |
8 | | -from email.mime.application import MIMEApplication |
9 | | -from pathlib import Path |
10 | 8 | from typing import Final |
11 | 9 |
|
12 | 10 | import httpx |
@@ -166,6 +164,15 @@ async def _get_invoice_pdf(invoice_pdf: str) -> httpx.Response | None: |
166 | 164 | return _response |
167 | 165 |
|
168 | 166 |
|
| 167 | +def _guess_file_type(filename: str) -> tuple[str, str]: |
| 168 | + mimetype, _encoding = mimetypes.guess_type(filename) |
| 169 | + if mimetype: |
| 170 | + maintype, subtype = mimetype.split("/", maxsplit=1) |
| 171 | + else: |
| 172 | + maintype, subtype = "application", "octet-stream" |
| 173 | + return maintype, subtype |
| 174 | + |
| 175 | + |
169 | 176 | async def _create_user_email( |
170 | 177 | env: Environment, |
171 | 178 | user: _UserData, |
@@ -206,39 +213,25 @@ async def _create_user_email( |
206 | 213 | match = invoice_file_name_pattern.search( |
207 | 214 | pdf_response.headers["content-disposition"] |
208 | 215 | ) |
209 | | - if match: |
210 | | - _file_name = match.group("filename") |
| 216 | + if ( |
| 217 | + match |
| 218 | + and (file_name := match.group("filename")) |
| 219 | + and (pdf_data := pdf_response.content) |
| 220 | + ): |
| 221 | + main_type, sub_type = _guess_file_type(file_name) |
| 222 | + msg.add_attachment( |
| 223 | + pdf_data, |
| 224 | + filename=file_name, |
| 225 | + maintype=main_type, |
| 226 | + subtype=sub_type, |
| 227 | + ) |
211 | 228 |
|
212 | | - attachment = MIMEApplication(pdf_response.content, Name=_file_name) |
213 | | - attachment["Content-Disposition"] = f"attachment; filename={_file_name}" |
214 | | - msg.attach(attachment) |
215 | 229 | else: |
216 | 230 | _logger.error("No match find for email attachment. This should not happen.") |
217 | 231 |
|
218 | 232 | return msg |
219 | 233 |
|
220 | 234 |
|
221 | | -def _guess_file_type(file_path: Path) -> tuple[str, str]: |
222 | | - assert file_path.is_file() |
223 | | - mimetype, _encoding = mimetypes.guess_type(file_path) |
224 | | - if mimetype: |
225 | | - maintype, subtype = mimetype.split("/", maxsplit=1) |
226 | | - else: |
227 | | - maintype, subtype = "application", "octet-stream" |
228 | | - return maintype, subtype |
229 | | - |
230 | | - |
231 | | -def _add_attachments(msg: EmailMessage, file_paths: list[Path]): |
232 | | - for attachment_path in file_paths: |
233 | | - maintype, subtype = _guess_file_type(attachment_path) |
234 | | - msg.add_attachment( |
235 | | - attachment_path.read_bytes(), |
236 | | - filename=attachment_path.name, |
237 | | - maintype=maintype, |
238 | | - subtype=subtype, |
239 | | - ) |
240 | | - |
241 | | - |
242 | 235 | @asynccontextmanager |
243 | 236 | async def _create_email_session( |
244 | 237 | settings: SMTPSettings, |
|
0 commit comments