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