foerderbarometer/input/utils/mail/attachments.py

95 lines
2.3 KiB
Python
Raw Normal View History

import os
2025-10-17 15:22:52 +00:00
import posixpath
import time
import mimetypes
2025-10-17 10:06:23 +00:00
2025-10-17 15:22:52 +00:00
from os import PathLike
from pathlib import Path
2025-10-17 15:22:52 +00:00
from urllib.parse import urlparse
from urllib.request import urlretrieve
2025-10-17 10:06:23 +00:00
from django.conf import settings
from django.core.mail import EmailMultiAlternatives
2025-10-17 14:04:14 +00:00
from foerderbarometer.constants import *
2025-10-17 15:22:52 +00:00
PathList = list[Path]
2025-10-17 10:06:23 +00:00
2025-10-17 15:22:52 +00:00
def ensure_dir(directory: PathLike) -> Path:
"""
2025-10-17 15:22:52 +00:00
Ensure that the given directory exists.
Creates it recursively if it doesn't.
"""
2025-10-17 14:04:14 +00:00
2025-10-17 15:22:52 +00:00
directory = Path(directory)
2025-10-17 15:22:52 +00:00
directory.mkdir(parents=True, exist_ok=True)
2025-10-17 15:22:52 +00:00
return directory
2025-10-17 15:22:52 +00:00
def is_fresh(path: Path, ttl_seconds: int) -> bool:
"""
Check if the cached file exists and is still fresh within TTL.
"""
2025-10-17 14:04:14 +00:00
try:
2025-10-17 14:04:14 +00:00
mtime = path.stat().st_mtime
except FileNotFoundError:
return False
2025-10-17 14:04:14 +00:00
else:
return time.time() - mtime < ttl_seconds
2025-10-17 14:04:14 +00:00
2025-10-17 15:22:52 +00:00
def get_attachment(url: str) -> Path:
filepath = urlparse(url).path
filename = posixpath.basename(filepath)
destination = ensure_dir(settings.MAIL_ATTACHMENT_CACHE_DIR) / filename
2025-10-17 14:04:14 +00:00
2025-10-17 15:22:52 +00:00
if is_fresh(destination, settings.MAIL_ATTACHMENT_TTL_SECONDS):
return destination
2025-10-17 15:22:52 +00:00
return download_attachment(url, destination)
2025-10-17 14:04:14 +00:00
2025-10-17 15:22:52 +00:00
def download_attachment(url: str, destination: Path) -> Path:
filepath = destination.with_suffix('.tmp')
2025-10-17 14:04:14 +00:00
2025-10-17 15:22:52 +00:00
try:
urlretrieve(url, filepath)
os.replace(filepath, destination)
finally:
filepath.unlink(missing_ok=True)
2025-10-17 14:04:14 +00:00
2025-10-17 15:22:52 +00:00
return destination
2025-10-17 15:22:52 +00:00
def collect_attachment_paths(recipient: str, type_code: str) -> PathList:
2025-10-17 14:04:14 +00:00
assert recipient in RECIPIENTS
assert type_code in TYPES
config = settings.MAIL_ATTACHMENT_URLS[recipient]
urls = [*config[TYPE_ALL], *config.get(type_code, [])]
2025-10-17 15:22:52 +00:00
return [get_attachment(url) for url in urls]
2025-10-17 14:04:14 +00:00
2025-10-17 15:22:52 +00:00
def get_mime_type(path: Path) -> str:
mime_type, encoding = mimetypes.guess_type(path)
2025-10-17 14:04:14 +00:00
2025-10-17 15:22:52 +00:00
return mime_type or 'application/octet-stream'
2025-10-17 15:22:52 +00:00
def attach_files(message: EmailMultiAlternatives, files: list[Path]):
"""
Attach files to the EmailMultiAlternatives message.
2025-10-17 15:22:52 +00:00
MIME type is guessed from path; falls back to application/octet-stream.
"""
2025-10-17 14:04:14 +00:00
2025-10-17 15:22:52 +00:00
for path in files:
mime_type = get_mime_type(path)
2025-10-17 15:22:52 +00:00
with open(path, 'rb') as fp:
message.attach(path.name, fp.read(), mime_type)