Files
CCMA/src/ccma/storage/atomic.py
T
Marcel PeterkauandClaude Opus 5 d6cd58a788 Render every outgoing e-mail from an editable text template
The wording of the dunning and SEPA pre-notification mails was hard-coded in
Python, so adjusting a single sentence required a new release. Both texts now
live in plain text templates that ship as defaults, are copied into the store's
templates/mail/ directory on first start and can be edited there or under
Optionen -> E-Mail-Vorlagen; an existing file is never overwritten and a deleted
one is restored from the shipped default.

A template carries its subject in the first line and the body after a blank
line. Placeholders use the same {{ ... }} syntax as the document templates and
share their member/organization values, so a name means the same thing in a
letter and in the mail that carries it. Unknown placeholders are rejected while
editing instead of during a send run, a line holding nothing but placeholders
that render empty is dropped, and {{#claims}} ... {{/claims}} repeats per entry.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-28 20:49:08 +02:00

83 lines
2.7 KiB
Python

import hashlib
import json
import os
import tempfile
from copy import deepcopy
from pathlib import Path
from typing import Any
CONTENT_HASH_FIELD = "content_hash"
def _hashable_copy(data: Any, *, hash_field: str = CONTENT_HASH_FIELD) -> Any:
if isinstance(data, dict):
return {
key: _hashable_copy(value, hash_field=hash_field)
for key, value in data.items()
if key != hash_field
}
if isinstance(data, list):
return [_hashable_copy(item, hash_field=hash_field) for item in data]
return data
def compute_json_content_hash(data: Any, *, hash_field: str = CONTENT_HASH_FIELD) -> str:
payload = json.dumps(
_hashable_copy(data, hash_field=hash_field),
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
return hashlib.sha256(payload).hexdigest()
def attach_json_content_hash(data: Any, *, hash_field: str = CONTENT_HASH_FIELD) -> Any:
cloned = deepcopy(data)
if isinstance(cloned, dict):
cloned[hash_field] = compute_json_content_hash(cloned, hash_field=hash_field)
return cloned
def json_content_hash_matches(data: Any, *, hash_field: str = CONTENT_HASH_FIELD) -> bool:
if not isinstance(data, dict):
return True
stored = str(data.get(hash_field, "")).strip()
if not stored:
return False
return stored == compute_json_content_hash(data, hash_field=hash_field)
def write_json_atomic(path: Path, data: Any) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
descriptor, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent)
temporary = Path(temporary_name)
payload = attach_json_content_hash(data)
try:
with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as handle:
json.dump(payload, handle, ensure_ascii=False, indent=2)
handle.write("\n")
handle.flush()
os.fsync(handle.fileno())
os.replace(temporary, path)
finally:
temporary.unlink(missing_ok=True)
def write_text_atomic(path: Path, content: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
descriptor, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent)
temporary = Path(temporary_name)
try:
with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as handle:
handle.write(content)
handle.flush()
os.fsync(handle.fileno())
os.replace(temporary, path)
finally:
temporary.unlink(missing_ok=True)
def read_json(path: Path) -> Any:
with path.open("r", encoding="utf-8") as handle:
return json.load(handle)