mirror of
https://git.hiabuto.net/C3MA/CCMA.git
synced 2026-08-25 06:55:17 +02:00
Marcel PeterkauandClaude Sonnet 5
d4dfd0066a
The dialog's base class sets resizable(False, False) before this subclass's content (including the preset-populated items table) exists, so its initial size stayed locked to a too-small guess and cut off the bottom. It now explicitly sizes to its actual content after everything, including the selected preset's items, has been built. The items table only supported add/remove -- there was no way to change an already-added row's amount (e.g. after picking the Rücklastschrift preset, its prefilled fee couldn't be adjusted). Selecting a row now loads it into the description/amount fields, and a new "Aktualisieren" button applies edits back to that row. The Rücklastschrift preset label was missing the "Stufe N:" prefix the other presets have, inconsistent for no reason. Also fixed a real gap: repositories created before standard_fee_items existed had no such key in repository.json at all, so Optionen showed an empty Standardpositionen table instead of the built-in defaults. get_reminder_policy() now backfills the defaults when the key is missing entirely, while still respecting a list the board intentionally emptied and saved. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2832 lines
120 KiB
Python
2832 lines
120 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import unicodedata
|
|
from collections.abc import Iterable
|
|
from contextlib import contextmanager
|
|
from datetime import date, datetime, timedelta
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
from string import Formatter
|
|
from uuid import uuid4
|
|
|
|
from ccma.domain.contributions import (
|
|
CONTRIBUTION_OVERRIDE_KIND_LABELS,
|
|
allocated_total,
|
|
claim_balance,
|
|
claim_total,
|
|
contribution_override_ranges_overlap,
|
|
credit_allocated_total,
|
|
decimal_value,
|
|
donation_balance,
|
|
materialize_claim_items,
|
|
money_text,
|
|
payment_allocated_total,
|
|
reminder_items_total,
|
|
)
|
|
from ccma.domain.dates import (
|
|
DateValidationError,
|
|
normalize_date_input,
|
|
parse_month_input,
|
|
validate_member_dates,
|
|
)
|
|
from ccma.domain.models import (
|
|
ASSET_CUSTODY_TYPE_LABELS,
|
|
ASSET_OWNER_TYPE_LABELS,
|
|
ASSET_STATUS_LABELS,
|
|
MEMBERSHIP_STATUS_LABELS,
|
|
PAYMENT_FREQUENCY_LABELS,
|
|
Asset,
|
|
ContributionData,
|
|
Event,
|
|
Member,
|
|
)
|
|
from ccma.storage.atomic import json_content_hash_matches, read_json, write_json_atomic
|
|
|
|
|
|
class RepositoryError(RuntimeError):
|
|
pass
|
|
|
|
|
|
DEFAULT_MEMBER_NUMBER_PATTERN = "CCMA-{number:04d}"
|
|
|
|
DEFAULT_CONTRIBUTION_RULES = [
|
|
{
|
|
"rule_id": "standard-before-2022",
|
|
"name": "Regulärer Beitrag bis 2021",
|
|
"valid_from": "1900-01-01",
|
|
"valid_until": "2021-12-31",
|
|
"annual_amount": "120.00",
|
|
"admission_fee": "10.00",
|
|
"annual_due": "01-31",
|
|
"semiannual_due": ["01-31", "07-31"],
|
|
"entry_proration": {"mode": "monthly", "started_month": "included"},
|
|
"first_payment_due_days_after_acceptance": 28,
|
|
"issue_days_before_due": 30,
|
|
},
|
|
{
|
|
"rule_id": "standard-2022",
|
|
"name": "Regulärer Beitrag ab 2022",
|
|
"valid_from": "2022-01-01",
|
|
"valid_until": None,
|
|
"annual_amount": "150.00",
|
|
"admission_fee": "10.00",
|
|
"annual_due": "01-31",
|
|
"semiannual_due": ["01-31", "07-31"],
|
|
"entry_proration": {"mode": "monthly", "started_month": "included"},
|
|
"first_payment_due_days_after_acceptance": 28,
|
|
"issue_days_before_due": 30,
|
|
},
|
|
]
|
|
|
|
|
|
DEFAULT_CONFIGURATION = {
|
|
"schema_version": 1,
|
|
"organization": {
|
|
"name": "Chaos Computer Club Mannheim e.V.",
|
|
"street": "",
|
|
"postal_code": "",
|
|
"city": "Mannheim",
|
|
"country": "Deutschland",
|
|
"email": "",
|
|
"phone": "",
|
|
"website": "",
|
|
"iban": "",
|
|
"bic": "",
|
|
"creditor_id": "",
|
|
},
|
|
"member_number_policy": {
|
|
"mode": "automatic",
|
|
"pattern": DEFAULT_MEMBER_NUMBER_PATTERN,
|
|
"allocation_strategy": "next_free",
|
|
},
|
|
"member_number_sequences": {},
|
|
"reminder_policy": {
|
|
"grace_days_after_due": 7,
|
|
"levels": [
|
|
{
|
|
"level": 1,
|
|
"name": "Zahlungserinnerung",
|
|
"fee": "0.00",
|
|
"payment_deadline_days": 14,
|
|
},
|
|
{
|
|
"level": 2,
|
|
"name": "Erste Mahnung",
|
|
"fee": "5.00",
|
|
"payment_deadline_days": 14,
|
|
},
|
|
{
|
|
"level": 3,
|
|
"name": "Zweite Mahnung",
|
|
"fee": "5.00",
|
|
"payment_deadline_days": 14,
|
|
},
|
|
],
|
|
# Reusable description/amount presets offered as rows when recording a
|
|
# reminder or a failed direct debit; boards can add/edit/remove freely.
|
|
"standard_fee_items": [
|
|
{"description": "Mahngebühr", "default_amount": "5.00"},
|
|
{"description": "Rücklastschriftgebühr", "default_amount": "5.00"},
|
|
{"description": "Bankgebühr", "default_amount": "0.00"},
|
|
{"description": "Porto", "default_amount": "0.00"},
|
|
],
|
|
},
|
|
"contribution_rules": DEFAULT_CONTRIBUTION_RULES,
|
|
}
|
|
|
|
|
|
class MemberRepository:
|
|
def __init__(self, root: Path | str):
|
|
self.root = Path(root).expanduser().resolve()
|
|
self.members_root = self.root / "members"
|
|
self.assets_root = self.root / "assets"
|
|
|
|
def initialize(self) -> None:
|
|
self.members_root.mkdir(parents=True, exist_ok=True)
|
|
self.assets_root.mkdir(parents=True, exist_ok=True)
|
|
(self.root / "rules").mkdir(parents=True, exist_ok=True)
|
|
templates_root = self.root / "templates"
|
|
templates_root.mkdir(parents=True, exist_ok=True)
|
|
builtin_templates = Path(__file__).resolve().parent.parent / "assets" / "templates"
|
|
if builtin_templates.is_dir():
|
|
for source in builtin_templates.iterdir():
|
|
destination_name = (
|
|
"Forderung mit Positionen.fodt" if source.name == "Forderung.fodt" else source.name
|
|
)
|
|
destination = templates_root / destination_name
|
|
if source.is_file() and not destination.exists():
|
|
shutil.copyfile(source, destination)
|
|
config_path = self.root / "repository.json"
|
|
if not config_path.exists():
|
|
write_json_atomic(config_path, DEFAULT_CONFIGURATION)
|
|
|
|
def validate(self) -> list[str]:
|
|
errors: list[str] = []
|
|
try:
|
|
config = read_json(self.root / "repository.json")
|
|
if not json_content_hash_matches(config):
|
|
errors.append(
|
|
"repository.json: Hash fehlt oder stimmt nicht; Datei wurde vermutlich extern geändert."
|
|
)
|
|
if int(config.get("schema_version", 0)) != 1:
|
|
errors.append("repository.json: nicht unterstützte schema_version")
|
|
policy = config.get("member_number_policy") or {}
|
|
if str(policy.get("mode", "automatic")) not in {"automatic", "manual"}:
|
|
errors.append("repository.json: ungültiger Mitgliedsnummernmodus")
|
|
if str(policy.get("allocation_strategy", "next_free")) not in {"next_free", "max_plus_one"}:
|
|
errors.append("repository.json: ungültige Vergabestrategie für Mitgliedsnummern")
|
|
validate_member_number_pattern(str(policy.get("pattern", DEFAULT_MEMBER_NUMBER_PATTERN)))
|
|
except (OSError, ValueError, TypeError, json.JSONDecodeError, RepositoryError) as exc:
|
|
errors.append(f"repository.json: {exc}")
|
|
|
|
seen_numbers: dict[str, str] = {}
|
|
for member_dir in self._member_directories():
|
|
try:
|
|
member, _contributions = self.preflight_member_record(member_dir.name)
|
|
errors.extend(
|
|
f"{member_dir.name}/{warning}" for warning in self.member_hash_warnings(member_dir.name)
|
|
)
|
|
validate_member_dates(
|
|
birth_date=member.birth_date,
|
|
accepted_at=member.accepted_at,
|
|
membership_started_at=member.membership_started_at,
|
|
membership_ended_at=member.membership_ended_at,
|
|
)
|
|
if member.member_id != member_dir.name:
|
|
errors.append(f"{member_dir.name}/member.json: member_id stimmt nicht mit Ordner überein")
|
|
normalized_number = member.member_number.casefold().strip()
|
|
if normalized_number and normalized_number in seen_numbers:
|
|
errors.append(
|
|
f"{member_dir.name}/member.json: Mitgliedsnummer {member.member_number} ist doppelt"
|
|
)
|
|
elif normalized_number:
|
|
seen_numbers[normalized_number] = member.member_id
|
|
except RepositoryError as exc:
|
|
errors.append(str(exc))
|
|
except (
|
|
OSError,
|
|
ValueError,
|
|
TypeError,
|
|
KeyError,
|
|
json.JSONDecodeError,
|
|
DateValidationError,
|
|
) as exc:
|
|
errors.append(f"{member_dir.name}/member.json: {exc}")
|
|
for asset_dir in self._asset_directories():
|
|
try:
|
|
asset = self.get_asset(asset_dir.name)
|
|
errors.extend(
|
|
f"{asset_dir.name}/{warning}" for warning in self.asset_hash_warnings(asset_dir.name)
|
|
)
|
|
if asset.asset_id != asset_dir.name:
|
|
errors.append(f"{asset_dir.name}/asset.json: asset_id stimmt nicht mit Ordner überein")
|
|
if asset.schema_version != 1:
|
|
errors.append(
|
|
f"{asset_dir.name}/asset.json: "
|
|
f"nicht unterstützte schema_version {asset.schema_version}"
|
|
)
|
|
if asset.status not in ASSET_STATUS_LABELS:
|
|
errors.append(f"{asset_dir.name}/asset.json: ungültiger Asset-Status")
|
|
if asset.current_holder_member_id:
|
|
self.get_member(asset.current_holder_member_id)
|
|
if asset.status != "issued":
|
|
errors.append(
|
|
f"{asset_dir.name}/asset.json: zugeordnetes Asset muss Status issued haben"
|
|
)
|
|
elif asset.status == "issued":
|
|
errors.append(f"{asset_dir.name}/asset.json: issued benötigt current_holder_member_id")
|
|
if asset.owner_type not in ASSET_OWNER_TYPE_LABELS:
|
|
errors.append(f"{asset_dir.name}/asset.json: ungültiger Eigentümertyp")
|
|
if asset.custody_type not in ASSET_CUSTODY_TYPE_LABELS:
|
|
errors.append(f"{asset_dir.name}/asset.json: ungültiger Gewahrsamstyp")
|
|
if asset.owner_type == "member":
|
|
if asset.owner_member_id:
|
|
self.get_member(asset.owner_member_id)
|
|
else:
|
|
errors.append(
|
|
f"{asset_dir.name}/asset.json: Mitgliedseigentum benötigt owner_member_id"
|
|
)
|
|
self.get_asset_events(asset.asset_id)
|
|
except RepositoryError as exc:
|
|
errors.append(str(exc))
|
|
except (OSError, ValueError, TypeError, KeyError, json.JSONDecodeError) as exc:
|
|
errors.append(f"{asset_dir.name}/asset.json: {exc}")
|
|
return errors
|
|
|
|
def list_members(self) -> list[Member]:
|
|
members: list[Member] = []
|
|
for member_id in self.list_member_ids():
|
|
try:
|
|
members.append(self.get_member(member_id))
|
|
except RepositoryError:
|
|
continue
|
|
return sorted(members, key=lambda item: (item.last_name.casefold(), item.first_name.casefold()))
|
|
|
|
def list_member_ids(self) -> list[str]:
|
|
return sorted(directory.name for directory in self._member_directories())
|
|
|
|
def get_member(self, member_id: str) -> Member:
|
|
path = self._member_path(member_id) / "member.json"
|
|
if not path.is_file():
|
|
raise RepositoryError(f"Mitglied nicht gefunden: {member_id}")
|
|
try:
|
|
raw = read_json(path)
|
|
if not isinstance(raw, dict):
|
|
raise TypeError("Wurzelelement muss ein JSON-Objekt sein")
|
|
for section_name in ("person", "address", "banking", "membership", "contribution_profile"):
|
|
if section_name in raw and not isinstance(raw[section_name], dict):
|
|
raise TypeError(f"{section_name} muss ein JSON-Objekt sein")
|
|
return Member.from_dict(raw)
|
|
except (OSError, ValueError, TypeError, KeyError, json.JSONDecodeError) as exc:
|
|
raise RepositoryError(f"{member_id}/member.json konnte nicht gelesen werden: {exc}") from exc
|
|
|
|
def preflight_member_record(self, member_id: str) -> tuple[Member, ContributionData]:
|
|
member = self.get_member(member_id)
|
|
if member.member_id != member_id:
|
|
raise RepositoryError(f"{member_id}/member.json: member_id stimmt nicht mit Ordner überein")
|
|
if member.schema_version != 1:
|
|
raise RepositoryError(
|
|
f"{member_id}/member.json: nicht unterstützte schema_version {member.schema_version}"
|
|
)
|
|
contributions = self.get_contributions(member_id)
|
|
try:
|
|
self.get_events(member_id)
|
|
except (OSError, UnicodeError) as exc:
|
|
raise RepositoryError(f"{member_id}/events.jsonl konnte nicht gelesen werden: {exc}") from exc
|
|
return member, contributions
|
|
|
|
def create_member(
|
|
self,
|
|
*,
|
|
first_name: str,
|
|
last_name: str,
|
|
nickname: str = "",
|
|
email: str = "",
|
|
phone: str = "",
|
|
birth_date: str = "",
|
|
application_date: str = "",
|
|
member_number: str = "",
|
|
) -> Member:
|
|
with self._member_number_lock():
|
|
return self._create_member_locked(
|
|
first_name=first_name,
|
|
last_name=last_name,
|
|
nickname=nickname,
|
|
email=email,
|
|
phone=phone,
|
|
birth_date=birth_date,
|
|
application_date=application_date,
|
|
member_number=member_number,
|
|
)
|
|
|
|
def _create_member_locked(
|
|
self,
|
|
*,
|
|
first_name: str,
|
|
last_name: str,
|
|
nickname: str,
|
|
email: str,
|
|
phone: str,
|
|
birth_date: str,
|
|
application_date: str,
|
|
member_number: str,
|
|
) -> Member:
|
|
if not first_name.strip() or not last_name.strip():
|
|
raise RepositoryError("Vorname und Nachname sind erforderlich.")
|
|
try:
|
|
birth_date = normalize_date_input(birth_date, "Geburtsdatum")
|
|
application_date = normalize_date_input(application_date, "Antragsdatum")
|
|
validate_member_dates(birth_date=birth_date)
|
|
except DateValidationError as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
selected_number = member_number.strip()
|
|
policy = self.get_member_number_policy()
|
|
if selected_number:
|
|
self._assert_member_number_available(selected_number)
|
|
elif policy["mode"] == "manual":
|
|
raise RepositoryError("Eine Mitgliedsnummer ist erforderlich.")
|
|
else:
|
|
selected_number = self._allocate_member_number(
|
|
policy["pattern"], policy["allocation_strategy"]
|
|
)
|
|
member_id = str(uuid4())
|
|
directory = self._member_path(member_id)
|
|
directory.mkdir(parents=True, exist_ok=False)
|
|
(directory / "files").mkdir()
|
|
member = Member(
|
|
member_id=member_id,
|
|
member_number=selected_number,
|
|
first_name=first_name.strip(),
|
|
last_name=last_name.strip(),
|
|
nickname=nickname.strip(),
|
|
email=email.strip(),
|
|
phone=phone.strip(),
|
|
birth_date=birth_date,
|
|
application_date=application_date,
|
|
)
|
|
write_json_atomic(directory / "member.json", member.to_dict())
|
|
write_json_atomic(directory / "contributions.json", ContributionData().to_dict())
|
|
self.append_event(
|
|
member_id,
|
|
event_type="member_created",
|
|
summary="Mitgliederakte angelegt",
|
|
actor_type="user",
|
|
actor_name="Vorstand",
|
|
)
|
|
return member
|
|
|
|
def save_member(self, member: Member, *, actor_name: str = "Vorstand") -> None:
|
|
existing = self.get_member(member.member_id)
|
|
try:
|
|
member.birth_date = normalize_date_input(member.birth_date, "Geburtsdatum")
|
|
member.application_date = normalize_date_input(member.application_date, "Antragsdatum")
|
|
member.accepted_at = normalize_date_input(member.accepted_at, "Aufnahmebeschluss")
|
|
member.membership_started_at = normalize_date_input(member.membership_started_at, "Mitglied seit")
|
|
member.membership_ended_at = normalize_date_input(member.membership_ended_at, "Austrittsdatum")
|
|
member.mandate_signed_at = normalize_date_input(member.mandate_signed_at, "Mandat erteilt am")
|
|
member.mandate_revoked_at = normalize_date_input(
|
|
member.mandate_revoked_at, "Mandat widerrufen am"
|
|
)
|
|
validate_member_dates(
|
|
birth_date=member.birth_date,
|
|
accepted_at=member.accepted_at,
|
|
membership_started_at=member.membership_started_at,
|
|
membership_ended_at=member.membership_ended_at,
|
|
)
|
|
if member.membership_ended_at and date.fromisoformat(member.membership_ended_at) <= date.today():
|
|
member.status = "ended"
|
|
except DateValidationError as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
member.iban = normalize_iban(member.iban)
|
|
member.bic = member.bic.replace(" ", "").upper()
|
|
validate_iban(member.iban)
|
|
validate_bic(member.bic)
|
|
if member.mandate_active and not (
|
|
member.iban and member.mandate_reference.strip() and member.mandate_signed_at
|
|
):
|
|
raise RepositoryError(
|
|
"Ein aktives Lastschriftmandat benötigt IBAN, Mandatsreferenz und Erteilungsdatum."
|
|
)
|
|
if member.payment_frequency not in PAYMENT_FREQUENCY_LABELS:
|
|
raise RepositoryError("Ungültige Zahlweise.")
|
|
if member.member_number != existing.member_number:
|
|
self._assert_member_number_available(member.member_number, exclude_member_id=member.member_id)
|
|
changes = self._summarize_changes(existing, member)
|
|
member.updated_at = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
write_json_atomic(self._member_path(member.member_id) / "member.json", member.to_dict())
|
|
if changes:
|
|
self.append_event(
|
|
member.member_id,
|
|
event_type="member_data_changed",
|
|
summary=f"Mitgliedsdaten geändert: {', '.join(changes)}",
|
|
actor_type="user",
|
|
actor_name=actor_name,
|
|
)
|
|
|
|
def list_assets(self) -> list[Asset]:
|
|
assets: list[Asset] = []
|
|
for asset_id in self.list_asset_ids():
|
|
try:
|
|
assets.append(self.get_asset(asset_id))
|
|
except RepositoryError:
|
|
continue
|
|
return sorted(assets, key=lambda item: (item.label.casefold(), item.inventory_number.casefold()))
|
|
|
|
def list_asset_ids(self) -> list[str]:
|
|
return sorted(directory.name for directory in self._asset_directories())
|
|
|
|
def get_asset(self, asset_id: str) -> Asset:
|
|
path = self._asset_path(asset_id) / "asset.json"
|
|
if not path.is_file():
|
|
raise RepositoryError(f"Asset nicht gefunden: {asset_id}")
|
|
try:
|
|
raw = read_json(path)
|
|
if not isinstance(raw, dict):
|
|
raise TypeError("Wurzelelement muss ein JSON-Objekt sein")
|
|
return Asset.from_dict(raw)
|
|
except (OSError, ValueError, TypeError, KeyError, json.JSONDecodeError) as exc:
|
|
raise RepositoryError(f"{asset_id}/asset.json konnte nicht gelesen werden: {exc}") from exc
|
|
|
|
def create_asset(
|
|
self,
|
|
*,
|
|
label: str,
|
|
category: str = "",
|
|
inventory_number: str = "",
|
|
serial_number: str = "",
|
|
deposit_amount_default: str = "0",
|
|
notes: str = "",
|
|
owner_type: str = "club",
|
|
owner_member_id: str = "",
|
|
owner_name: str = "",
|
|
custody_type: str = "club",
|
|
location: str = "",
|
|
handed_over_at: str = "",
|
|
condition: str = "",
|
|
estimated_value: str = "",
|
|
) -> Asset:
|
|
if not label.strip():
|
|
raise RepositoryError("Eine Bezeichnung für das Asset ist erforderlich.")
|
|
try:
|
|
deposit_amount = decimal_value(deposit_amount_default or "0", "Kaution")
|
|
except ValueError as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if deposit_amount < 0:
|
|
raise RepositoryError("Die Kaution darf nicht negativ sein.")
|
|
self._validate_asset_relationships(owner_type, owner_member_id, owner_name, custody_type, "")
|
|
asset_id = str(uuid4())
|
|
directory = self._asset_path(asset_id)
|
|
directory.mkdir(parents=True, exist_ok=False)
|
|
(directory / "files").mkdir()
|
|
asset = Asset(
|
|
asset_id=asset_id,
|
|
label=label.strip(),
|
|
category=category.strip(),
|
|
inventory_number=inventory_number.strip(),
|
|
serial_number=serial_number.strip(),
|
|
deposit_amount_default=money_text(deposit_amount),
|
|
notes=notes.strip(),
|
|
owner_type=owner_type,
|
|
owner_member_id=owner_member_id,
|
|
owner_name=owner_name.strip(),
|
|
custody_type=custody_type,
|
|
location=location.strip(),
|
|
handed_over_at=handed_over_at.strip(),
|
|
condition=condition.strip(),
|
|
estimated_value=estimated_value.strip(),
|
|
)
|
|
write_json_atomic(directory / "asset.json", asset.to_dict())
|
|
self.append_asset_event(
|
|
asset.asset_id,
|
|
event_type="asset_created",
|
|
summary="Asset angelegt",
|
|
actor_type="user",
|
|
actor_name="Vorstand",
|
|
)
|
|
if asset.owner_type == "member":
|
|
self.append_event(
|
|
asset.owner_member_id,
|
|
event_type="member_asset_registered",
|
|
summary=f"Privateigentum im Inventar erfasst: {asset.label}",
|
|
actor_type="user",
|
|
actor_name="Vorstand",
|
|
references={"asset_id": asset.asset_id},
|
|
)
|
|
return asset
|
|
|
|
def save_asset(self, asset: Asset, *, actor_name: str = "Vorstand") -> None:
|
|
existing = self.get_asset(asset.asset_id)
|
|
if not asset.label.strip():
|
|
raise RepositoryError("Eine Bezeichnung für das Asset ist erforderlich.")
|
|
if asset.status not in ASSET_STATUS_LABELS:
|
|
raise RepositoryError("Ungültiger Asset-Status.")
|
|
self._validate_asset_relationships(
|
|
asset.owner_type,
|
|
asset.owner_member_id,
|
|
asset.owner_name,
|
|
asset.custody_type,
|
|
asset.current_holder_member_id,
|
|
)
|
|
try:
|
|
deposit_amount = decimal_value(asset.deposit_amount_default or "0", "Kaution")
|
|
except ValueError as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if deposit_amount < 0:
|
|
raise RepositoryError("Die Kaution darf nicht negativ sein.")
|
|
if existing.current_holder_member_id and money_text(deposit_amount) != str(
|
|
existing.deposit_amount_default
|
|
):
|
|
raise RepositoryError(
|
|
"Die Kaution kann nur geändert werden, wenn das Asset nicht ausgegeben ist."
|
|
)
|
|
asset.label = asset.label.strip()
|
|
asset.category = asset.category.strip()
|
|
asset.inventory_number = asset.inventory_number.strip()
|
|
asset.serial_number = asset.serial_number.strip()
|
|
asset.deposit_amount_default = money_text(deposit_amount)
|
|
asset.notes = asset.notes.strip()
|
|
asset.owner_name = asset.owner_name.strip()
|
|
asset.location = asset.location.strip()
|
|
asset.handed_over_at = asset.handed_over_at.strip()
|
|
asset.condition = asset.condition.strip()
|
|
asset.estimated_value = asset.estimated_value.strip()
|
|
if asset.current_holder_member_id:
|
|
self.get_member(asset.current_holder_member_id)
|
|
if asset.status != "issued":
|
|
raise RepositoryError("Ein zugeordnetes Asset muss den Status issued haben.")
|
|
elif asset.status == "issued":
|
|
raise RepositoryError("Status issued benötigt ein zugeordnetes Mitglied.")
|
|
changes = self._summarize_asset_changes(existing, asset)
|
|
asset.updated_at = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
write_json_atomic(self._asset_path(asset.asset_id) / "asset.json", asset.to_dict())
|
|
if changes:
|
|
self.append_asset_event(
|
|
asset.asset_id,
|
|
event_type="asset_data_changed",
|
|
summary=f"Assetdaten geändert: {', '.join(changes)}",
|
|
actor_type="user",
|
|
actor_name=actor_name,
|
|
)
|
|
if existing.owner_member_id != asset.owner_member_id:
|
|
if existing.owner_type == "member" and existing.owner_member_id:
|
|
self.append_event(
|
|
existing.owner_member_id,
|
|
event_type="member_asset_ownership_ended",
|
|
summary=f"Nicht mehr als Privateigentum geführt: {asset.label}",
|
|
actor_type="user",
|
|
actor_name=actor_name,
|
|
references={"asset_id": asset.asset_id},
|
|
)
|
|
if asset.owner_type == "member" and asset.owner_member_id:
|
|
self.append_event(
|
|
asset.owner_member_id,
|
|
event_type="member_asset_registered",
|
|
summary=f"Privateigentum im Inventar erfasst: {asset.label}",
|
|
actor_type="user",
|
|
actor_name=actor_name,
|
|
references={"asset_id": asset.asset_id},
|
|
)
|
|
|
|
def assign_asset(self, asset_id: str, member_id: str, *, actor_name: str = "Vorstand") -> Asset:
|
|
asset = self.get_asset(asset_id)
|
|
member = self.get_member(member_id)
|
|
if asset.current_holder_member_id:
|
|
raise RepositoryError("Das Asset ist bereits einem Mitglied zugeordnet.")
|
|
if asset.status in {"lost", "retired"}:
|
|
raise RepositoryError("Verlorene oder ausgemusterte Assets können nicht ausgegeben werden.")
|
|
asset.current_holder_member_id = member.member_id
|
|
asset.custody_type = "member"
|
|
asset.status = "issued"
|
|
asset.updated_at = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
write_json_atomic(self._asset_path(asset.asset_id) / "asset.json", asset.to_dict())
|
|
self.append_asset_event(
|
|
asset.asset_id,
|
|
event_type="asset_issued",
|
|
summary=f"Asset ausgegeben an {member.member_number or member.member_id}",
|
|
actor_type="user",
|
|
actor_name=actor_name,
|
|
references={"member_id": member.member_id},
|
|
)
|
|
self.append_event(
|
|
member.member_id,
|
|
event_type="asset_assigned",
|
|
summary=f"Asset ausgegeben: {asset.label}",
|
|
actor_type="user",
|
|
actor_name=actor_name,
|
|
references={"asset_id": asset.asset_id},
|
|
)
|
|
return asset
|
|
|
|
def return_asset(self, asset_id: str, *, actor_name: str = "Vorstand") -> Asset:
|
|
asset = self.get_asset(asset_id)
|
|
member_id = asset.current_holder_member_id
|
|
if not member_id:
|
|
raise RepositoryError("Das Asset ist aktuell keinem Mitglied zugeordnet.")
|
|
asset.current_holder_member_id = ""
|
|
asset.custody_type = "club"
|
|
asset.status = "available"
|
|
asset.updated_at = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
write_json_atomic(self._asset_path(asset.asset_id) / "asset.json", asset.to_dict())
|
|
self.append_asset_event(
|
|
asset.asset_id,
|
|
event_type="asset_returned",
|
|
summary="Asset zurückgenommen",
|
|
actor_type="user",
|
|
actor_name=actor_name,
|
|
references={"member_id": member_id},
|
|
)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="asset_returned",
|
|
summary=f"Asset zurückgegeben: {asset.label}",
|
|
actor_type="user",
|
|
actor_name=actor_name,
|
|
references={"asset_id": asset.asset_id},
|
|
)
|
|
return asset
|
|
|
|
def list_member_assets(self, member_id: str) -> list[Asset]:
|
|
self.get_member(member_id)
|
|
return [asset for asset in self.list_assets() if asset.current_holder_member_id == member_id]
|
|
|
|
def list_member_owned_assets(self, member_id: str) -> list[Asset]:
|
|
self.get_member(member_id)
|
|
return [
|
|
asset
|
|
for asset in self.list_assets()
|
|
if asset.owner_type == "member" and asset.owner_member_id == member_id
|
|
]
|
|
|
|
def _validate_asset_relationships(
|
|
self,
|
|
owner_type: str,
|
|
owner_member_id: str,
|
|
owner_name: str,
|
|
custody_type: str,
|
|
holder_member_id: str,
|
|
) -> None:
|
|
if owner_type not in ASSET_OWNER_TYPE_LABELS:
|
|
raise RepositoryError("Ungültiger Eigentümertyp.")
|
|
if custody_type not in ASSET_CUSTODY_TYPE_LABELS:
|
|
raise RepositoryError("Ungültiger Gewahrsamstyp.")
|
|
if owner_type == "member":
|
|
if not owner_member_id:
|
|
raise RepositoryError("Bei Mitgliedseigentum muss ein Mitglied ausgewählt werden.")
|
|
self.get_member(owner_member_id)
|
|
elif owner_member_id:
|
|
raise RepositoryError("Eine Eigentümer-Mitglieds-ID ist nur bei Mitgliedseigentum erlaubt.")
|
|
if owner_type == "external" and not owner_name.strip():
|
|
raise RepositoryError("Bei externem Eigentum ist ein Eigentümername erforderlich.")
|
|
if custody_type == "member":
|
|
if not holder_member_id:
|
|
raise RepositoryError("Gewahrsam beim Mitglied benötigt eine Mitgliedszuordnung.")
|
|
self.get_member(holder_member_id)
|
|
elif holder_member_id:
|
|
raise RepositoryError("Eine Halter-Mitglieds-ID ist nur bei Gewahrsam eines Mitglieds erlaubt.")
|
|
|
|
def create_manual_claim(
|
|
self,
|
|
member_id: str,
|
|
*,
|
|
title: str,
|
|
amount: str,
|
|
due_date: str,
|
|
description: str = "",
|
|
claim_type: str = "asset_charge",
|
|
references: dict[str, str] | None = None,
|
|
actor_name: str = "Vorstand",
|
|
) -> dict:
|
|
self.get_member(member_id)
|
|
if not title.strip():
|
|
raise RepositoryError("Ein Forderungstitel ist erforderlich.")
|
|
try:
|
|
normalized_due_date = normalize_date_input(due_date, "Fälligkeitsdatum")
|
|
amount_value = decimal_value(amount)
|
|
except (DateValidationError, ValueError) as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if not normalized_due_date:
|
|
raise RepositoryError("Ein Fälligkeitsdatum ist erforderlich.")
|
|
if amount_value == 0:
|
|
raise RepositoryError("Der Betrag darf nicht null sein.")
|
|
now = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
claim_id = str(uuid4())
|
|
item_type = "credit" if amount_value < 0 else "base"
|
|
claim = {
|
|
"claim_id": claim_id,
|
|
"claim_key": f"manual:{claim_type}:{claim_id}",
|
|
"title": title.strip(),
|
|
"amount": money_text(amount_value),
|
|
"due_date": normalized_due_date,
|
|
"status": "open",
|
|
"created_at": now,
|
|
"items": [
|
|
{
|
|
"item_id": str(uuid4()),
|
|
"type": item_type,
|
|
"description": description.strip() or title.strip(),
|
|
"quantity": "1.00",
|
|
"unit_price": money_text(amount_value),
|
|
"amount": money_text(amount_value),
|
|
"created_at": now,
|
|
}
|
|
],
|
|
"origin": {
|
|
"type": "manual",
|
|
"subtype": claim_type.strip() or "asset_charge",
|
|
"actor": actor_name,
|
|
"asset_id": (references or {}).get("asset_id", ""),
|
|
},
|
|
}
|
|
data = self.get_contributions(member_id)
|
|
data.claims.append(claim)
|
|
self.save_contributions(member_id, data)
|
|
event = self.append_event(
|
|
member_id,
|
|
event_type="claim_created",
|
|
summary=f"Forderung angelegt: {claim['title']}",
|
|
actor_type="user",
|
|
actor_name=actor_name,
|
|
references={"claim_id": claim_id, **(references or {})},
|
|
data={"amount": claim["amount"], "due_date": claim["due_date"], "claim_type": claim_type},
|
|
)
|
|
if references and references.get("asset_id"):
|
|
self.append_asset_event(
|
|
references["asset_id"],
|
|
event_type="asset_claim_created",
|
|
summary=f"Forderung für Asset angelegt: {claim['title']}",
|
|
actor_type="user",
|
|
actor_name=actor_name,
|
|
references={"member_id": member_id, "claim_id": claim_id},
|
|
data={"amount": claim["amount"], "due_date": claim["due_date"], "claim_type": claim_type},
|
|
)
|
|
return {"claim": claim, "event": event}
|
|
|
|
def append_asset_event(
|
|
self,
|
|
asset_id: str,
|
|
*,
|
|
event_type: str,
|
|
summary: str,
|
|
actor_type: str = "system",
|
|
actor_name: str = "CCMA",
|
|
references: dict[str, str] | None = None,
|
|
data: dict[str, object] | None = None,
|
|
) -> Event:
|
|
directory = self._asset_path(asset_id)
|
|
if not (directory / "asset.json").is_file():
|
|
raise RepositoryError(f"Asset nicht gefunden: {asset_id}")
|
|
event = Event(
|
|
event_id=str(uuid4()),
|
|
timestamp=datetime.now().astimezone().isoformat(timespec="seconds"),
|
|
event_type=event_type,
|
|
summary=summary.strip(),
|
|
actor_type=actor_type,
|
|
actor_name=actor_name,
|
|
references=references or {},
|
|
data=data or {},
|
|
)
|
|
path = directory / "events.jsonl"
|
|
line = json.dumps(event.to_dict(), ensure_ascii=False, separators=(",", ":")) + "\n"
|
|
with path.open("a", encoding="utf-8", newline="\n") as handle:
|
|
handle.write(line)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
return event
|
|
|
|
def get_asset_events(self, asset_id: str) -> list[Event]:
|
|
path = self._asset_path(asset_id) / "events.jsonl"
|
|
if not path.exists():
|
|
return []
|
|
events: list[Event] = []
|
|
with path.open("r", encoding="utf-8") as handle:
|
|
for line_number, line in enumerate(handle, start=1):
|
|
if not line.strip():
|
|
continue
|
|
try:
|
|
raw = json.loads(line)
|
|
if not isinstance(raw, dict):
|
|
raise TypeError("Event muss ein JSON-Objekt sein")
|
|
events.append(Event.from_dict(raw))
|
|
except (AttributeError, ValueError, TypeError, KeyError, json.JSONDecodeError) as exc:
|
|
raise RepositoryError(f"Ungültiges Event in Zeile {line_number}: {exc}") from exc
|
|
return events
|
|
|
|
def get_contributions(self, member_id: str) -> ContributionData:
|
|
path = self._member_path(member_id) / "contributions.json"
|
|
if not path.exists():
|
|
return ContributionData()
|
|
try:
|
|
raw = read_json(path)
|
|
if not isinstance(raw, dict):
|
|
raise TypeError("Wurzelelement muss ein JSON-Objekt sein")
|
|
for field_name in (
|
|
"claims",
|
|
"payments",
|
|
"credits",
|
|
"allocations",
|
|
"reminders",
|
|
"donations",
|
|
"contribution_overrides",
|
|
):
|
|
if field_name in raw and not isinstance(raw[field_name], list):
|
|
raise TypeError(f"{field_name} muss eine JSON-Liste sein")
|
|
if field_name in raw and any(not isinstance(item, dict) for item in raw[field_name]):
|
|
raise TypeError(f"Alle Einträge in {field_name} müssen JSON-Objekte sein")
|
|
return ContributionData.from_dict(raw)
|
|
except (OSError, ValueError, TypeError, json.JSONDecodeError) as exc:
|
|
raise RepositoryError(
|
|
f"{member_id}/contributions.json konnte nicht gelesen werden: {exc}"
|
|
) from exc
|
|
|
|
def save_contributions(self, member_id: str, data: ContributionData) -> None:
|
|
self.get_member(member_id)
|
|
write_json_atomic(self._member_path(member_id) / "contributions.json", data.to_dict())
|
|
|
|
def get_claim(self, member_id: str, claim_id: str) -> tuple[ContributionData, dict]:
|
|
data = self.get_contributions(member_id)
|
|
claim = next(
|
|
(item for item in data.claims if str(item.get("claim_id", "")) == claim_id),
|
|
None,
|
|
)
|
|
if claim is None:
|
|
raise RepositoryError(f"Forderung nicht gefunden: {claim_id}")
|
|
return data, claim
|
|
|
|
def add_claim_item(
|
|
self,
|
|
member_id: str,
|
|
claim_id: str,
|
|
*,
|
|
description: str,
|
|
quantity: str,
|
|
unit_price: str,
|
|
item_type: str = "correction",
|
|
) -> dict:
|
|
if not description.strip():
|
|
raise RepositoryError("Eine Beschreibung ist erforderlich.")
|
|
try:
|
|
selected_quantity = decimal_value(quantity, "Menge")
|
|
selected_unit_price = decimal_value(unit_price, "Einzelpreis")
|
|
except ValueError as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if selected_quantity == 0:
|
|
raise RepositoryError("Die Menge darf nicht null sein.")
|
|
data, claim = self.get_claim(member_id, claim_id)
|
|
if str(claim.get("status", "")) == "cancelled":
|
|
raise RepositoryError("Eine stornierte Forderung kann nicht verändert werden.")
|
|
amount = selected_quantity * selected_unit_price
|
|
item = {
|
|
"item_id": str(uuid4()),
|
|
"type": item_type.strip() or "correction",
|
|
"description": description.strip(),
|
|
"quantity": money_text(selected_quantity),
|
|
"unit_price": money_text(selected_unit_price),
|
|
"amount": money_text(amount),
|
|
"created_at": datetime.now().astimezone().isoformat(timespec="seconds"),
|
|
}
|
|
materialize_claim_items(claim).append(item)
|
|
claim["amount"] = money_text(claim_total(claim))
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="claim_item_added",
|
|
summary=f"Forderungsposition ergänzt: {item['description']} ({item['amount']} EUR)",
|
|
references={"claim_id": claim_id, "item_id": str(item["item_id"])},
|
|
)
|
|
return item
|
|
|
|
def update_claim(
|
|
self,
|
|
member_id: str,
|
|
claim_id: str,
|
|
*,
|
|
title: str,
|
|
due_date: str,
|
|
base_amount: str,
|
|
description: str,
|
|
actor_name: str = "Vorstand",
|
|
) -> dict:
|
|
if not title.strip():
|
|
raise RepositoryError("Ein Forderungstitel ist erforderlich.")
|
|
try:
|
|
normalized_due_date = normalize_date_input(due_date, "Fälligkeitsdatum")
|
|
selected_base_amount = decimal_value(base_amount, "Grundbetrag")
|
|
except (DateValidationError, ValueError) as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if not normalized_due_date:
|
|
raise RepositoryError("Ein Fälligkeitsdatum ist erforderlich.")
|
|
if selected_base_amount == 0:
|
|
raise RepositoryError("Der Grundbetrag darf nicht null sein.")
|
|
|
|
data, claim = self.get_claim(member_id, claim_id)
|
|
if str(claim.get("status", "")) == "cancelled":
|
|
raise RepositoryError("Eine stornierte Forderung kann nicht bearbeitet werden.")
|
|
items = materialize_claim_items(claim)
|
|
base_item = next((item for item in items if str(item.get("type", "")) == "base"), None)
|
|
if base_item is None:
|
|
raise RepositoryError("Die Forderung hat keine bearbeitbare Grundposition.")
|
|
|
|
other_total = sum(
|
|
(decimal_value(item.get("amount", "0")) for item in items if item is not base_item),
|
|
Decimal("0"),
|
|
)
|
|
new_total = selected_base_amount + other_total
|
|
settled = allocated_total(data, claim_id)
|
|
if new_total >= 0 and settled > new_total:
|
|
raise RepositoryError(
|
|
f"Der neue Gesamtbetrag darf nicht unter dem bereits zugeordneten Betrag "
|
|
f"von {money_text(settled)} EUR liegen."
|
|
)
|
|
|
|
old_values = {
|
|
"title": str(claim.get("title", "")),
|
|
"due_date": str(claim.get("due_date", "")),
|
|
"base_amount": str(base_item.get("amount", "")),
|
|
"description": str(base_item.get("description", "")),
|
|
}
|
|
now = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
claim["title"] = title.strip()
|
|
claim["due_date"] = normalized_due_date
|
|
claim["amount"] = money_text(new_total)
|
|
base_item["description"] = description.strip() or title.strip()
|
|
base_item["quantity"] = "1.00"
|
|
base_item["unit_price"] = money_text(selected_base_amount)
|
|
base_item["amount"] = money_text(selected_base_amount)
|
|
calculation = claim.get("calculation")
|
|
if not isinstance(calculation, dict):
|
|
calculation = {}
|
|
claim["calculation"] = calculation
|
|
calculation["manual_override"] = {"at": now, "actor": actor_name}
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="claim_changed",
|
|
summary=f"Forderung geändert: {claim['title']}",
|
|
actor_type="user",
|
|
actor_name=actor_name,
|
|
references={"claim_id": claim_id},
|
|
data={
|
|
"old": old_values,
|
|
"new": {
|
|
"title": claim["title"],
|
|
"due_date": claim["due_date"],
|
|
"base_amount": base_item["amount"],
|
|
"description": base_item["description"],
|
|
},
|
|
},
|
|
)
|
|
return claim
|
|
|
|
def record_payment(
|
|
self,
|
|
member_id: str,
|
|
claim_id: str,
|
|
*,
|
|
payment_date: str,
|
|
amount: str,
|
|
allocation_amount: str,
|
|
gnucash_transaction_id: str = "",
|
|
reference: str = "",
|
|
) -> dict:
|
|
try:
|
|
normalized_date = normalize_date_input(payment_date, "Zahlungsdatum")
|
|
selected_amount = decimal_value(amount)
|
|
selected_allocation = decimal_value(allocation_amount, "Zuordnung")
|
|
except (DateValidationError, ValueError) as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if not normalized_date:
|
|
raise RepositoryError("Zahlungsdatum ist erforderlich.")
|
|
if selected_amount <= 0:
|
|
raise RepositoryError("Der Zahlungsbetrag muss größer als null sein.")
|
|
if selected_allocation <= 0 or selected_allocation > selected_amount:
|
|
raise RepositoryError(
|
|
"Die Zuordnung muss größer als null und höchstens so hoch wie die Zahlung sein."
|
|
)
|
|
gnucash_id = gnucash_transaction_id.strip()
|
|
if gnucash_id:
|
|
self._assert_gnucash_id_available(gnucash_id)
|
|
data, claim = self.get_claim(member_id, claim_id)
|
|
available_claim_balance = max(claim_balance(data, claim), Decimal("0"))
|
|
if selected_allocation > available_claim_balance:
|
|
raise RepositoryError(
|
|
f"Die Forderung hat nur noch {money_text(available_claim_balance)} EUR offen."
|
|
)
|
|
payment = {
|
|
"payment_id": str(uuid4()),
|
|
"date": normalized_date,
|
|
"amount": money_text(selected_amount),
|
|
"method": "bank_transfer",
|
|
"gnucash_transaction_id": gnucash_id,
|
|
"reference": reference.strip(),
|
|
"created_at": datetime.now().astimezone().isoformat(timespec="seconds"),
|
|
}
|
|
allocation = {
|
|
"allocation_id": str(uuid4()),
|
|
"payment_id": payment["payment_id"],
|
|
"claim_id": claim_id,
|
|
"amount": money_text(selected_allocation),
|
|
}
|
|
data.payments.append(payment)
|
|
data.allocations.append(allocation)
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="payment_recorded",
|
|
summary=f"Zahlung eingegangen: {payment['amount']} EUR",
|
|
references={"claim_id": claim_id, "payment_id": str(payment["payment_id"])},
|
|
data={"allocation_amount": allocation["amount"]},
|
|
)
|
|
return payment
|
|
|
|
def record_combined_payment(
|
|
self,
|
|
member_id: str,
|
|
*,
|
|
payment_date: str,
|
|
allocations: dict[str, str],
|
|
reference: str = "",
|
|
method: str = "bank_transfer",
|
|
actor_name: str = "CCMA",
|
|
) -> dict:
|
|
"""Record one payment and atomically allocate it to multiple claims."""
|
|
try:
|
|
normalized_date = normalize_date_input(payment_date, "Zahlungsdatum")
|
|
except DateValidationError as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if not normalized_date:
|
|
raise RepositoryError("Zahlungsdatum ist erforderlich.")
|
|
if not allocations:
|
|
raise RepositoryError("Mindestens eine Zuordnung ist erforderlich.")
|
|
|
|
data = self.get_contributions(member_id)
|
|
claims_by_id = {str(claim.get("claim_id", "")): claim for claim in data.claims}
|
|
selected_allocations: dict[str, Decimal] = {}
|
|
for claim_id, raw_amount in allocations.items():
|
|
claim = claims_by_id.get(claim_id)
|
|
if claim is None:
|
|
raise RepositoryError(f"Forderung nicht gefunden: {claim_id}")
|
|
if str(claim.get("status", "")) == "cancelled":
|
|
raise RepositoryError("Eine stornierte Forderung kann nicht bezahlt werden.")
|
|
try:
|
|
selected_amount = decimal_value(raw_amount, "Zuordnung")
|
|
except ValueError as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
available = max(claim_balance(data, claim), Decimal("0"))
|
|
if selected_amount <= 0 or selected_amount > available:
|
|
raise RepositoryError(
|
|
f"{claim.get('title', 'Forderung')} hat nur "
|
|
f"{money_text(available)} EUR offen."
|
|
)
|
|
selected_allocations[claim_id] = selected_amount
|
|
|
|
total = sum(selected_allocations.values(), Decimal("0"))
|
|
now = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
payment = {
|
|
"payment_id": str(uuid4()),
|
|
"date": normalized_date,
|
|
"amount": money_text(total),
|
|
"method": method.strip() or "bank_transfer",
|
|
"gnucash_transaction_id": "",
|
|
"reference": reference.strip(),
|
|
"created_at": now,
|
|
}
|
|
for claim_id, amount in selected_allocations.items():
|
|
data.allocations.append(
|
|
{
|
|
"allocation_id": str(uuid4()),
|
|
"payment_id": payment["payment_id"],
|
|
"claim_id": claim_id,
|
|
"amount": money_text(amount),
|
|
}
|
|
)
|
|
data.payments.append(payment)
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="payment_recorded",
|
|
summary=f"Zahlung eingegangen: {payment['amount']} EUR",
|
|
actor_type="system" if method == "dummy" else "user",
|
|
actor_name=actor_name,
|
|
references={"payment_id": str(payment["payment_id"])},
|
|
data={
|
|
"allocated_amount": payment["amount"],
|
|
"claim_ids": list(selected_allocations),
|
|
"method": payment["method"],
|
|
},
|
|
)
|
|
return payment
|
|
|
|
def create_payment(
|
|
self,
|
|
member_id: str,
|
|
*,
|
|
payment_date: str,
|
|
amount: str,
|
|
claim_allocations: dict[str, str] | None = None,
|
|
donation_allocations: dict[str, str] | None = None,
|
|
gnucash_transaction_id: str = "",
|
|
reference: str = "",
|
|
method: str = "bank_transfer",
|
|
actor_name: str = "Vorstand",
|
|
) -> dict:
|
|
"""Record an incoming payment, optionally allocating parts of it immediately to
|
|
open claims and/or donations. Called with no allocations, this just logs a bank
|
|
transfer as soon as it arrives, to be assigned to claims or donations later."""
|
|
self.get_member(member_id)
|
|
try:
|
|
normalized_date = normalize_date_input(payment_date, "Zahlungsdatum")
|
|
selected_amount = decimal_value(amount)
|
|
except (DateValidationError, ValueError) as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if not normalized_date:
|
|
raise RepositoryError("Zahlungsdatum ist erforderlich.")
|
|
if selected_amount <= 0:
|
|
raise RepositoryError("Der Zahlungsbetrag muss größer als null sein.")
|
|
gnucash_id = gnucash_transaction_id.strip()
|
|
if gnucash_id:
|
|
self._assert_gnucash_id_available(gnucash_id)
|
|
|
|
data = self.get_contributions(member_id)
|
|
claims_by_id = {str(claim.get("claim_id", "")): claim for claim in data.claims}
|
|
donations_by_id = {str(item.get("donation_id", "")): item for item in data.donations}
|
|
|
|
selected_claim_allocations: dict[str, Decimal] = {}
|
|
for claim_id, raw_amount in (claim_allocations or {}).items():
|
|
claim = claims_by_id.get(claim_id)
|
|
if claim is None:
|
|
raise RepositoryError(f"Forderung nicht gefunden: {claim_id}")
|
|
if str(claim.get("status", "")) == "cancelled":
|
|
raise RepositoryError("Eine stornierte Forderung kann nicht bezahlt werden.")
|
|
try:
|
|
selected = decimal_value(raw_amount, "Zuordnung")
|
|
except ValueError as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if selected <= 0:
|
|
continue
|
|
available = max(claim_balance(data, claim), Decimal("0"))
|
|
if selected > available:
|
|
raise RepositoryError(
|
|
f"{claim.get('title', 'Forderung')} hat nur {money_text(available)} EUR offen."
|
|
)
|
|
selected_claim_allocations[claim_id] = selected
|
|
|
|
selected_donation_allocations: dict[str, Decimal] = {}
|
|
for donation_id, raw_amount in (donation_allocations or {}).items():
|
|
donation = donations_by_id.get(donation_id)
|
|
if donation is None:
|
|
raise RepositoryError(f"Spende nicht gefunden: {donation_id}")
|
|
try:
|
|
selected = decimal_value(raw_amount, "Zuordnung")
|
|
except ValueError as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if selected <= 0:
|
|
continue
|
|
available = max(donation_balance(data, donation), Decimal("0"))
|
|
if selected > available:
|
|
raise RepositoryError(f"Die Spende hat nur noch {money_text(available)} EUR offen.")
|
|
selected_donation_allocations[donation_id] = selected
|
|
|
|
allocated_total_amount = sum(selected_claim_allocations.values(), Decimal("0")) + sum(
|
|
selected_donation_allocations.values(), Decimal("0")
|
|
)
|
|
if allocated_total_amount > selected_amount:
|
|
raise RepositoryError(
|
|
f"Die Zuordnungen ({money_text(allocated_total_amount)} EUR) übersteigen den "
|
|
f"Zahlungsbetrag ({money_text(selected_amount)} EUR)."
|
|
)
|
|
|
|
payment = {
|
|
"payment_id": str(uuid4()),
|
|
"date": normalized_date,
|
|
"amount": money_text(selected_amount),
|
|
"method": method.strip() or "bank_transfer",
|
|
"gnucash_transaction_id": gnucash_id,
|
|
"reference": reference.strip(),
|
|
"created_at": datetime.now().astimezone().isoformat(timespec="seconds"),
|
|
}
|
|
for claim_id, claim_amount in selected_claim_allocations.items():
|
|
data.allocations.append(
|
|
{
|
|
"allocation_id": str(uuid4()),
|
|
"payment_id": payment["payment_id"],
|
|
"claim_id": claim_id,
|
|
"amount": money_text(claim_amount),
|
|
}
|
|
)
|
|
for donation_id, donation_amount in selected_donation_allocations.items():
|
|
data.allocations.append(
|
|
{
|
|
"allocation_id": str(uuid4()),
|
|
"payment_id": payment["payment_id"],
|
|
"donation_id": donation_id,
|
|
"amount": money_text(donation_amount),
|
|
}
|
|
)
|
|
data.payments.append(payment)
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="payment_recorded",
|
|
summary=f"Zahlung erfasst: {payment['amount']} EUR",
|
|
actor_type="user",
|
|
actor_name=actor_name,
|
|
references={"payment_id": str(payment["payment_id"])},
|
|
data={
|
|
"allocated_amount": money_text(allocated_total_amount),
|
|
"claim_ids": list(selected_claim_allocations),
|
|
"donation_ids": list(selected_donation_allocations),
|
|
},
|
|
)
|
|
return payment
|
|
|
|
def allocate_payment(self, member_id: str, claim_id: str, *, payment_id: str, amount: str) -> dict:
|
|
data, claim = self.get_claim(member_id, claim_id)
|
|
payment = next(
|
|
(item for item in data.payments if str(item.get("payment_id", "")) == payment_id),
|
|
None,
|
|
)
|
|
if payment is None:
|
|
raise RepositoryError("Zahlung nicht gefunden.")
|
|
try:
|
|
selected_amount = decimal_value(amount, "Zuordnung")
|
|
available = decimal_value(payment.get("amount", "0")) - payment_allocated_total(data, payment_id)
|
|
except ValueError as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if selected_amount <= 0 or selected_amount > available:
|
|
raise RepositoryError(f"Es sind nur {money_text(available)} EUR dieser Zahlung verfügbar.")
|
|
available_claim_balance = max(claim_balance(data, claim), Decimal("0"))
|
|
if selected_amount > available_claim_balance:
|
|
raise RepositoryError(
|
|
f"Die Forderung hat nur noch {money_text(available_claim_balance)} EUR offen."
|
|
)
|
|
allocation = {
|
|
"allocation_id": str(uuid4()),
|
|
"payment_id": payment_id,
|
|
"claim_id": claim_id,
|
|
"amount": money_text(selected_amount),
|
|
}
|
|
data.allocations.append(allocation)
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="payment_allocated",
|
|
summary=f"Zahlung zugeordnet: {allocation['amount']} EUR",
|
|
references={"claim_id": claim_id, "payment_id": payment_id},
|
|
)
|
|
return allocation
|
|
|
|
def update_payment(
|
|
self,
|
|
member_id: str,
|
|
payment_id: str,
|
|
*,
|
|
payment_date: str,
|
|
amount: str,
|
|
allocations: dict[str, str],
|
|
donation_allocations: dict[str, str] | None = None,
|
|
gnucash_transaction_id: str = "",
|
|
reference: str = "",
|
|
) -> dict:
|
|
"""`allocations` fully replaces the claim allocations of this payment, and
|
|
`donation_allocations` (if given) fully replaces its donation allocations. Omit
|
|
`donation_allocations` to leave any existing donation allocations untouched."""
|
|
data = self.get_contributions(member_id)
|
|
payment = next(
|
|
(item for item in data.payments if str(item.get("payment_id", "")) == payment_id),
|
|
None,
|
|
)
|
|
if payment is None:
|
|
raise RepositoryError("Zahlung nicht gefunden.")
|
|
try:
|
|
normalized_date = normalize_date_input(payment_date, "Zahlungsdatum")
|
|
selected_amount = decimal_value(amount)
|
|
except (DateValidationError, ValueError) as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if not normalized_date:
|
|
raise RepositoryError("Zahlungsdatum ist erforderlich.")
|
|
if selected_amount <= 0:
|
|
raise RepositoryError("Der Zahlungsbetrag muss größer als null sein.")
|
|
|
|
claims_by_id = {str(claim.get("claim_id", "")): claim for claim in data.claims}
|
|
donations_by_id = {str(item.get("donation_id", "")): item for item in data.donations}
|
|
old_allocations = [item for item in data.allocations if str(item.get("payment_id", "")) == payment_id]
|
|
old_by_claim: dict[str, list[dict]] = {}
|
|
old_by_donation: dict[str, list[dict]] = {}
|
|
for allocation in old_allocations:
|
|
if str(allocation.get("donation_id", "")):
|
|
old_by_donation.setdefault(str(allocation.get("donation_id", "")), []).append(allocation)
|
|
else:
|
|
old_by_claim.setdefault(str(allocation.get("claim_id", "")), []).append(allocation)
|
|
|
|
selected_allocations: dict[str, Decimal] = {}
|
|
for claim_id, raw_amount in allocations.items():
|
|
if claim_id not in claims_by_id:
|
|
raise RepositoryError(f"Forderung nicht gefunden: {claim_id}")
|
|
try:
|
|
allocation_amount = decimal_value(raw_amount, "Zuordnung")
|
|
except ValueError as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if allocation_amount < 0:
|
|
raise RepositoryError("Zuordnungen dürfen nicht negativ sein.")
|
|
if allocation_amount == 0:
|
|
continue
|
|
claim = claims_by_id[claim_id]
|
|
currently_allocated = sum(
|
|
(decimal_value(item.get("amount", "0")) for item in old_by_claim.get(claim_id, [])),
|
|
Decimal("0"),
|
|
)
|
|
available_claim_balance = max(
|
|
claim_balance(data, claim) + currently_allocated,
|
|
Decimal("0"),
|
|
)
|
|
if allocation_amount > available_claim_balance:
|
|
raise RepositoryError(
|
|
f"{claim.get('title', 'Forderung')} hat nur "
|
|
f"{money_text(available_claim_balance)} EUR offen."
|
|
)
|
|
selected_allocations[claim_id] = allocation_amount
|
|
|
|
if donation_allocations is None:
|
|
selected_donation_allocations = {
|
|
donation_id: sum(
|
|
(decimal_value(item.get("amount", "0")) for item in items), Decimal("0")
|
|
)
|
|
for donation_id, items in old_by_donation.items()
|
|
}
|
|
else:
|
|
selected_donation_allocations = {}
|
|
for donation_id, raw_amount in donation_allocations.items():
|
|
if donation_id not in donations_by_id:
|
|
raise RepositoryError(f"Spende nicht gefunden: {donation_id}")
|
|
try:
|
|
donation_amount = decimal_value(raw_amount, "Zuordnung")
|
|
except ValueError as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if donation_amount < 0:
|
|
raise RepositoryError("Zuordnungen dürfen nicht negativ sein.")
|
|
if donation_amount == 0:
|
|
continue
|
|
donation = donations_by_id[donation_id]
|
|
currently_allocated = sum(
|
|
(decimal_value(item.get("amount", "0")) for item in old_by_donation.get(donation_id, [])),
|
|
Decimal("0"),
|
|
)
|
|
available_balance = max(
|
|
donation_balance(data, donation) + currently_allocated, Decimal("0")
|
|
)
|
|
if donation_amount > available_balance:
|
|
raise RepositoryError(
|
|
f"Die Spende hat nur noch {money_text(available_balance)} EUR offen."
|
|
)
|
|
selected_donation_allocations[donation_id] = donation_amount
|
|
|
|
allocated_sum = sum(selected_allocations.values(), Decimal("0")) + sum(
|
|
selected_donation_allocations.values(), Decimal("0")
|
|
)
|
|
if allocated_sum > selected_amount:
|
|
raise RepositoryError(
|
|
f"Die Zuordnungen ({money_text(allocated_sum)} EUR) übersteigen den "
|
|
f"Zahlungsbetrag ({money_text(selected_amount)} EUR)."
|
|
)
|
|
|
|
gnucash_id = gnucash_transaction_id.strip()
|
|
if gnucash_id:
|
|
self._assert_gnucash_id_available(gnucash_id, exclude_payment_id=payment_id)
|
|
payment.update(
|
|
{
|
|
"date": normalized_date,
|
|
"amount": money_text(selected_amount),
|
|
"gnucash_transaction_id": gnucash_id,
|
|
"reference": reference.strip(),
|
|
"updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
|
|
}
|
|
)
|
|
|
|
retained_allocations = [
|
|
item for item in data.allocations if str(item.get("payment_id", "")) != payment_id
|
|
]
|
|
new_allocations = []
|
|
for claim_id, allocation_amount in selected_allocations.items():
|
|
prior = old_by_claim.get(claim_id, [])
|
|
new_allocations.append(
|
|
{
|
|
"allocation_id": (str(prior[0].get("allocation_id", "")) if prior else str(uuid4())),
|
|
"payment_id": payment_id,
|
|
"claim_id": claim_id,
|
|
"amount": money_text(allocation_amount),
|
|
}
|
|
)
|
|
for donation_id, donation_amount in selected_donation_allocations.items():
|
|
prior = old_by_donation.get(donation_id, [])
|
|
new_allocations.append(
|
|
{
|
|
"allocation_id": (str(prior[0].get("allocation_id", "")) if prior else str(uuid4())),
|
|
"payment_id": payment_id,
|
|
"donation_id": donation_id,
|
|
"amount": money_text(donation_amount),
|
|
}
|
|
)
|
|
data.allocations = retained_allocations + new_allocations
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="payment_changed",
|
|
summary=f"Zahlung geändert: {payment['amount']} EUR",
|
|
actor_type="user",
|
|
actor_name="Vorstand",
|
|
references={"payment_id": payment_id},
|
|
data={
|
|
"allocated_amount": money_text(allocated_sum),
|
|
"unallocated_amount": money_text(selected_amount - allocated_sum),
|
|
},
|
|
)
|
|
return payment
|
|
|
|
def update_payment_reference(
|
|
self,
|
|
member_id: str,
|
|
payment_id: str,
|
|
*,
|
|
reference: str,
|
|
gnucash_transaction_id: str = "",
|
|
actor_name: str = "Vorstand",
|
|
) -> dict:
|
|
"""Relabel an existing payment without touching its date, amount, or
|
|
allocations -- used e.g. when a GnuCash import recognizes a booking as
|
|
matching an already-recorded payment and the board wants to adopt the
|
|
(better) description from the statement instead of re-importing it."""
|
|
data = self.get_contributions(member_id)
|
|
payment = next(
|
|
(item for item in data.payments if str(item.get("payment_id", "")) == payment_id),
|
|
None,
|
|
)
|
|
if payment is None:
|
|
raise RepositoryError("Zahlung nicht gefunden.")
|
|
gnucash_id = gnucash_transaction_id.strip()
|
|
if gnucash_id and gnucash_id != str(payment.get("gnucash_transaction_id", "")):
|
|
self._assert_gnucash_id_available(gnucash_id, exclude_payment_id=payment_id)
|
|
payment["reference"] = reference.strip()
|
|
if gnucash_id:
|
|
payment["gnucash_transaction_id"] = gnucash_id
|
|
payment["updated_at"] = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="payment_changed",
|
|
summary=f"Zahlung geändert: Referenz aktualisiert ({payment['reference']})",
|
|
actor_type="user",
|
|
actor_name=actor_name,
|
|
references={"payment_id": payment_id},
|
|
)
|
|
return payment
|
|
|
|
def delete_payment(self, member_id: str, payment_id: str) -> None:
|
|
data = self.get_contributions(member_id)
|
|
payment = next(
|
|
(item for item in data.payments if str(item.get("payment_id", "")) == payment_id),
|
|
None,
|
|
)
|
|
if payment is None:
|
|
raise RepositoryError("Zahlung nicht gefunden.")
|
|
data.payments = [item for item in data.payments if str(item.get("payment_id", "")) != payment_id]
|
|
data.allocations = [
|
|
item for item in data.allocations if str(item.get("payment_id", "")) != payment_id
|
|
]
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="payment_deleted",
|
|
summary=f"Zahlung gelöscht: {payment.get('amount', '')} EUR",
|
|
actor_type="user",
|
|
actor_name="Vorstand",
|
|
references={"payment_id": payment_id},
|
|
)
|
|
|
|
def record_credit(
|
|
self,
|
|
member_id: str,
|
|
claim_id: str,
|
|
*,
|
|
credit_date: str,
|
|
amount: str,
|
|
allocation_amount: str,
|
|
reference: str = "",
|
|
) -> dict:
|
|
try:
|
|
normalized_date = normalize_date_input(credit_date, "Gutschriftsdatum")
|
|
selected_amount = decimal_value(amount, "Gutschrift")
|
|
selected_allocation = decimal_value(allocation_amount, "Zuordnung")
|
|
except (DateValidationError, ValueError) as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if not normalized_date:
|
|
raise RepositoryError("Gutschriftsdatum ist erforderlich.")
|
|
if selected_amount <= 0:
|
|
raise RepositoryError("Der Gutschriftsbetrag muss größer als null sein.")
|
|
if selected_allocation <= 0 or selected_allocation > selected_amount:
|
|
raise RepositoryError(
|
|
"Die Zuordnung muss größer als null und höchstens so hoch wie die Gutschrift sein."
|
|
)
|
|
data, claim = self.get_claim(member_id, claim_id)
|
|
if claim_total(claim) >= 0:
|
|
raise RepositoryError("Gutschriften können nur negativen Forderungen zugeordnet werden.")
|
|
credit = {
|
|
"credit_id": str(uuid4()),
|
|
"date": normalized_date,
|
|
"amount": money_text(selected_amount),
|
|
"method": "payout",
|
|
"reference": reference.strip(),
|
|
"created_at": datetime.now().astimezone().isoformat(timespec="seconds"),
|
|
}
|
|
allocation = {
|
|
"allocation_id": str(uuid4()),
|
|
"credit_id": credit["credit_id"],
|
|
"claim_id": claim_id,
|
|
"amount": money_text(selected_allocation),
|
|
}
|
|
data.credits.append(credit)
|
|
data.allocations.append(allocation)
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="credit_recorded",
|
|
summary=f"Gutschrift / Auszahlung erfasst: {credit['amount']} EUR",
|
|
references={"claim_id": claim_id, "credit_id": str(credit["credit_id"])},
|
|
data={"allocation_amount": allocation["amount"]},
|
|
)
|
|
return credit
|
|
|
|
def allocate_credit(self, member_id: str, claim_id: str, *, credit_id: str, amount: str) -> dict:
|
|
data, claim = self.get_claim(member_id, claim_id)
|
|
if claim_total(claim) >= 0:
|
|
raise RepositoryError("Gutschriften können nur negativen Forderungen zugeordnet werden.")
|
|
credit = next(
|
|
(item for item in data.credits if str(item.get("credit_id", "")) == credit_id),
|
|
None,
|
|
)
|
|
if credit is None:
|
|
raise RepositoryError("Gutschrift nicht gefunden.")
|
|
try:
|
|
selected_amount = decimal_value(amount, "Zuordnung")
|
|
available = decimal_value(credit.get("amount", "0")) - credit_allocated_total(data, credit_id)
|
|
except ValueError as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if selected_amount <= 0 or selected_amount > available:
|
|
raise RepositoryError(f"Es sind nur {money_text(available)} EUR dieser Gutschrift verfügbar.")
|
|
allocation = {
|
|
"allocation_id": str(uuid4()),
|
|
"credit_id": credit_id,
|
|
"claim_id": claim_id,
|
|
"amount": money_text(selected_amount),
|
|
}
|
|
data.allocations.append(allocation)
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="credit_allocated",
|
|
summary=f"Gutschrift zugeordnet: {allocation['amount']} EUR",
|
|
references={"claim_id": claim_id, "credit_id": credit_id},
|
|
)
|
|
return allocation
|
|
|
|
def record_donation(
|
|
self,
|
|
member_id: str,
|
|
*,
|
|
donation_date: str,
|
|
amount: str,
|
|
reference: str = "",
|
|
purpose: str = "",
|
|
actor_name: str = "Vorstand",
|
|
) -> dict:
|
|
self.get_member(member_id)
|
|
try:
|
|
normalized_date = normalize_date_input(donation_date, "Spendendatum")
|
|
selected_amount = decimal_value(amount, "Spendenbetrag")
|
|
except (DateValidationError, ValueError) as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if not normalized_date:
|
|
raise RepositoryError("Ein Spendendatum ist erforderlich.")
|
|
if selected_amount <= 0:
|
|
raise RepositoryError("Der Spendenbetrag muss größer als null sein.")
|
|
donation = {
|
|
"donation_id": str(uuid4()),
|
|
"date": normalized_date,
|
|
"amount": money_text(selected_amount),
|
|
"reference": reference.strip(),
|
|
"purpose": purpose.strip(),
|
|
"created_at": datetime.now().astimezone().isoformat(timespec="seconds"),
|
|
}
|
|
data = self.get_contributions(member_id)
|
|
data.donations.append(donation)
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="donation_recorded",
|
|
summary=f"Spende erfasst: {donation['amount']} EUR",
|
|
actor_type="user",
|
|
actor_name=actor_name,
|
|
references={"donation_id": donation["donation_id"]},
|
|
)
|
|
return donation
|
|
|
|
def get_donation(self, member_id: str, donation_id: str) -> tuple[ContributionData, dict]:
|
|
data = self.get_contributions(member_id)
|
|
donation = next(
|
|
(item for item in data.donations if str(item.get("donation_id", "")) == donation_id),
|
|
None,
|
|
)
|
|
if donation is None:
|
|
raise RepositoryError(f"Spende nicht gefunden: {donation_id}")
|
|
return data, donation
|
|
|
|
def update_donation(
|
|
self,
|
|
member_id: str,
|
|
donation_id: str,
|
|
*,
|
|
donation_date: str,
|
|
amount: str,
|
|
reference: str = "",
|
|
purpose: str = "",
|
|
actor_name: str = "Vorstand",
|
|
) -> dict:
|
|
try:
|
|
normalized_date = normalize_date_input(donation_date, "Spendendatum")
|
|
selected_amount = decimal_value(amount, "Spendenbetrag")
|
|
except (DateValidationError, ValueError) as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if not normalized_date:
|
|
raise RepositoryError("Ein Spendendatum ist erforderlich.")
|
|
if selected_amount <= 0:
|
|
raise RepositoryError("Der Spendenbetrag muss größer als null sein.")
|
|
data, donation = self.get_donation(member_id, donation_id)
|
|
allocated = sum(
|
|
(
|
|
decimal_value(item.get("amount", "0"))
|
|
for item in data.allocations
|
|
if str(item.get("donation_id", "")) == donation_id
|
|
),
|
|
Decimal("0"),
|
|
)
|
|
if selected_amount < allocated:
|
|
raise RepositoryError(
|
|
f"Der Betrag darf nicht unter den bereits zugeordneten "
|
|
f"{money_text(allocated)} EUR liegen."
|
|
)
|
|
donation["date"] = normalized_date
|
|
donation["amount"] = money_text(selected_amount)
|
|
donation["reference"] = reference.strip()
|
|
donation["purpose"] = purpose.strip()
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="donation_changed",
|
|
summary=f"Spende geändert: {donation['amount']} EUR",
|
|
actor_type="user",
|
|
actor_name=actor_name,
|
|
references={"donation_id": donation_id},
|
|
)
|
|
return donation
|
|
|
|
def delete_donation(self, member_id: str, donation_id: str, *, actor_name: str = "Vorstand") -> None:
|
|
"""Permanently remove a donation. Payments allocated to it are released, not deleted."""
|
|
data, donation = self.get_donation(member_id, donation_id)
|
|
data.donations = [
|
|
item for item in data.donations if str(item.get("donation_id", "")) != donation_id
|
|
]
|
|
data.allocations = [
|
|
item for item in data.allocations if str(item.get("donation_id", "")) != donation_id
|
|
]
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="donation_deleted",
|
|
summary=f"Spende gelöscht: {donation.get('amount', '')} EUR",
|
|
actor_type="user",
|
|
actor_name=actor_name,
|
|
references={"donation_id": donation_id},
|
|
)
|
|
|
|
def record_donation_payment(
|
|
self,
|
|
member_id: str,
|
|
donation_id: str,
|
|
*,
|
|
payment_date: str,
|
|
amount: str,
|
|
allocation_amount: str,
|
|
gnucash_transaction_id: str = "",
|
|
reference: str = "",
|
|
) -> dict:
|
|
try:
|
|
normalized_date = normalize_date_input(payment_date, "Zahlungsdatum")
|
|
selected_amount = decimal_value(amount)
|
|
selected_allocation = decimal_value(allocation_amount, "Zuordnung")
|
|
except (DateValidationError, ValueError) as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if not normalized_date:
|
|
raise RepositoryError("Zahlungsdatum ist erforderlich.")
|
|
if selected_amount <= 0:
|
|
raise RepositoryError("Der Zahlungsbetrag muss größer als null sein.")
|
|
if selected_allocation <= 0 or selected_allocation > selected_amount:
|
|
raise RepositoryError(
|
|
"Die Zuordnung muss größer als null und höchstens so hoch wie die Zahlung sein."
|
|
)
|
|
gnucash_id = gnucash_transaction_id.strip()
|
|
if gnucash_id:
|
|
self._assert_gnucash_id_available(gnucash_id)
|
|
data, donation = self.get_donation(member_id, donation_id)
|
|
available_balance = max(donation_balance(data, donation), Decimal("0"))
|
|
if selected_allocation > available_balance:
|
|
raise RepositoryError(f"Die Spende hat nur noch {money_text(available_balance)} EUR offen.")
|
|
payment = {
|
|
"payment_id": str(uuid4()),
|
|
"date": normalized_date,
|
|
"amount": money_text(selected_amount),
|
|
"method": "bank_transfer",
|
|
"gnucash_transaction_id": gnucash_id,
|
|
"reference": reference.strip(),
|
|
"created_at": datetime.now().astimezone().isoformat(timespec="seconds"),
|
|
}
|
|
allocation = {
|
|
"allocation_id": str(uuid4()),
|
|
"payment_id": payment["payment_id"],
|
|
"donation_id": donation_id,
|
|
"amount": money_text(selected_allocation),
|
|
}
|
|
data.payments.append(payment)
|
|
data.allocations.append(allocation)
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="payment_recorded",
|
|
summary=f"Zahlung für Spende eingegangen: {payment['amount']} EUR",
|
|
references={"donation_id": donation_id, "payment_id": str(payment["payment_id"])},
|
|
data={"allocation_amount": allocation["amount"]},
|
|
)
|
|
return payment
|
|
|
|
def allocate_payment_to_donation(
|
|
self, member_id: str, donation_id: str, *, payment_id: str, amount: str
|
|
) -> dict:
|
|
data, donation = self.get_donation(member_id, donation_id)
|
|
payment = next(
|
|
(item for item in data.payments if str(item.get("payment_id", "")) == payment_id),
|
|
None,
|
|
)
|
|
if payment is None:
|
|
raise RepositoryError("Zahlung nicht gefunden.")
|
|
try:
|
|
selected_amount = decimal_value(amount, "Zuordnung")
|
|
available = decimal_value(payment.get("amount", "0")) - payment_allocated_total(data, payment_id)
|
|
except ValueError as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if selected_amount <= 0 or selected_amount > available:
|
|
raise RepositoryError(f"Es sind nur {money_text(available)} EUR dieser Zahlung verfügbar.")
|
|
available_balance = max(donation_balance(data, donation), Decimal("0"))
|
|
if selected_amount > available_balance:
|
|
raise RepositoryError(f"Die Spende hat nur noch {money_text(available_balance)} EUR offen.")
|
|
allocation = {
|
|
"allocation_id": str(uuid4()),
|
|
"payment_id": payment_id,
|
|
"donation_id": donation_id,
|
|
"amount": money_text(selected_amount),
|
|
}
|
|
data.allocations.append(allocation)
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="payment_allocated",
|
|
summary=f"Zahlung Spende zugeordnet: {allocation['amount']} EUR",
|
|
references={"donation_id": donation_id, "payment_id": payment_id},
|
|
)
|
|
return allocation
|
|
|
|
def get_contribution_override(self, member_id: str, override_id: str) -> tuple[ContributionData, dict]:
|
|
data = self.get_contributions(member_id)
|
|
override = next(
|
|
(
|
|
item
|
|
for item in data.contribution_overrides
|
|
if str(item.get("override_id", "")) == override_id
|
|
),
|
|
None,
|
|
)
|
|
if override is None:
|
|
raise RepositoryError(f"Abweichender Beitrag nicht gefunden: {override_id}")
|
|
return data, override
|
|
|
|
def record_contribution_override(
|
|
self,
|
|
member_id: str,
|
|
*,
|
|
valid_from: str,
|
|
valid_until: str = "",
|
|
kind: str,
|
|
value: str,
|
|
reason: str,
|
|
actor_name: str = "Vorstand",
|
|
) -> dict:
|
|
self.get_member(member_id)
|
|
normalized = self._normalize_contribution_override_input(
|
|
valid_from=valid_from, valid_until=valid_until, kind=kind, value=value, reason=reason
|
|
)
|
|
normalized_from, normalized_until, normalized_kind, normalized_value, normalized_reason = normalized
|
|
data = self.get_contributions(member_id)
|
|
self._assert_contribution_override_range_free(data, normalized_from, normalized_until)
|
|
override = {
|
|
"override_id": str(uuid4()),
|
|
"valid_from": normalized_from,
|
|
"valid_until": normalized_until,
|
|
"kind": normalized_kind,
|
|
"value": normalized_value,
|
|
"reason": normalized_reason,
|
|
"created_at": datetime.now().astimezone().isoformat(timespec="seconds"),
|
|
"created_by": actor_name,
|
|
}
|
|
data.contribution_overrides.append(override)
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="contribution_override_recorded",
|
|
summary=f"Abweichender Mitgliedsbeitrag erfasst: {_contribution_override_summary(override)}",
|
|
actor_type="user",
|
|
actor_name=actor_name,
|
|
references={"override_id": override["override_id"]},
|
|
)
|
|
return override
|
|
|
|
def update_contribution_override(
|
|
self,
|
|
member_id: str,
|
|
override_id: str,
|
|
*,
|
|
valid_from: str,
|
|
valid_until: str = "",
|
|
kind: str,
|
|
value: str,
|
|
reason: str,
|
|
actor_name: str = "Vorstand",
|
|
) -> dict:
|
|
normalized = self._normalize_contribution_override_input(
|
|
valid_from=valid_from, valid_until=valid_until, kind=kind, value=value, reason=reason
|
|
)
|
|
normalized_from, normalized_until, normalized_kind, normalized_value, normalized_reason = normalized
|
|
data, override = self.get_contribution_override(member_id, override_id)
|
|
self._assert_contribution_override_range_free(
|
|
data, normalized_from, normalized_until, exclude_override_id=override_id
|
|
)
|
|
override.update(
|
|
{
|
|
"valid_from": normalized_from,
|
|
"valid_until": normalized_until,
|
|
"kind": normalized_kind,
|
|
"value": normalized_value,
|
|
"reason": normalized_reason,
|
|
"updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
|
|
}
|
|
)
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="contribution_override_changed",
|
|
summary=f"Abweichender Mitgliedsbeitrag geändert: {_contribution_override_summary(override)}",
|
|
actor_type="user",
|
|
actor_name=actor_name,
|
|
references={"override_id": override_id},
|
|
)
|
|
return override
|
|
|
|
def delete_contribution_override(
|
|
self, member_id: str, override_id: str, *, actor_name: str = "Vorstand"
|
|
) -> None:
|
|
data, override = self.get_contribution_override(member_id, override_id)
|
|
data.contribution_overrides = [
|
|
item
|
|
for item in data.contribution_overrides
|
|
if str(item.get("override_id", "")) != override_id
|
|
]
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="contribution_override_deleted",
|
|
summary=f"Abweichender Mitgliedsbeitrag gelöscht: {_contribution_override_summary(override)}",
|
|
actor_type="user",
|
|
actor_name=actor_name,
|
|
references={"override_id": override_id},
|
|
)
|
|
|
|
@staticmethod
|
|
def _normalize_contribution_override_input(
|
|
*, valid_from: str, valid_until: str, kind: str, value: str, reason: str
|
|
) -> tuple[str, str, str, str, str]:
|
|
try:
|
|
normalized_from = parse_month_input(valid_from, "Ab", allow_empty=False)
|
|
normalized_until = parse_month_input(valid_until, "Bis")
|
|
except DateValidationError as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if normalized_until and normalized_until < normalized_from:
|
|
raise RepositoryError("Der Zeitraum 'Bis' darf nicht vor dem Zeitraum 'Ab' liegen.")
|
|
normalized_kind = kind.strip().lower()
|
|
if normalized_kind not in CONTRIBUTION_OVERRIDE_KIND_LABELS:
|
|
raise RepositoryError("Ungültige Art des abweichenden Beitrags.")
|
|
try:
|
|
parsed_value = decimal_value(value, "Wert")
|
|
except ValueError as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if normalized_kind == "percent":
|
|
if parsed_value <= 0 or parsed_value > 100:
|
|
raise RepositoryError("Die Ermäßigung muss zwischen 0 und 100 Prozent liegen.")
|
|
elif parsed_value < 0:
|
|
raise RepositoryError("Der Jahresbetrag darf nicht negativ sein.")
|
|
normalized_reason = reason.strip()
|
|
if not normalized_reason:
|
|
raise RepositoryError("Eine Begründung ist erforderlich.")
|
|
return normalized_from, normalized_until, normalized_kind, money_text(parsed_value), normalized_reason
|
|
|
|
@staticmethod
|
|
def _assert_contribution_override_range_free(
|
|
data: ContributionData,
|
|
valid_from: str,
|
|
valid_until: str,
|
|
*,
|
|
exclude_override_id: str | None = None,
|
|
) -> None:
|
|
for existing in data.contribution_overrides:
|
|
if exclude_override_id and str(existing.get("override_id", "")) == exclude_override_id:
|
|
continue
|
|
if contribution_override_ranges_overlap(
|
|
valid_from,
|
|
valid_until,
|
|
str(existing.get("valid_from", "")),
|
|
str(existing.get("valid_until", "") or ""),
|
|
):
|
|
existing_until = str(existing.get("valid_until", "") or "") or "unbefristet"
|
|
raise RepositoryError(
|
|
"Der Zeitraum überschneidet sich mit einem bereits vorhandenen abweichenden "
|
|
f"Beitrag ({existing.get('valid_from', '')} bis {existing_until})."
|
|
)
|
|
|
|
def create_reminder_draft(
|
|
self,
|
|
member_id: str,
|
|
claim_id: str,
|
|
*,
|
|
level: int,
|
|
name: str,
|
|
payment_deadline_days: int,
|
|
detail: str = "",
|
|
items: list[dict[str, str]] | None = None,
|
|
channel: str = "email",
|
|
) -> dict:
|
|
if level < 1:
|
|
raise RepositoryError("Die Mahnstufe muss mindestens 1 sein.")
|
|
try:
|
|
normalized_items = _normalize_reminder_items(items)
|
|
except ValueError as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if payment_deadline_days < 1 or payment_deadline_days > 365:
|
|
raise RepositoryError("Die Zahlungsfrist muss zwischen 1 und 365 Tagen liegen.")
|
|
data, claim = self.get_claim(member_id, claim_id)
|
|
if str(claim.get("status", "")) == "cancelled":
|
|
raise RepositoryError("Eine stornierte Forderung kann nicht gemahnt werden.")
|
|
if _dunning_hold_is_active(claim):
|
|
raise RepositoryError("Für diese Forderung ist eine Mahnsperre aktiv.")
|
|
if claim_balance(data, claim) <= 0:
|
|
raise RepositoryError("Die Forderung hat keinen offenen Betrag.")
|
|
if any(
|
|
str(item.get("claim_id", "")) == claim_id
|
|
and int(item.get("level", 0)) == level
|
|
and str(item.get("status", "draft")) in {"draft", "generated", "sent"}
|
|
for item in data.reminders
|
|
):
|
|
raise RepositoryError(f"Mahnstufe {level} existiert bereits.")
|
|
if level > 1 and not any(
|
|
str(item.get("claim_id", "")) == claim_id
|
|
and int(item.get("level", 0)) == level - 1
|
|
and str(item.get("status", "")) == "sent"
|
|
for item in data.reminders
|
|
):
|
|
raise RepositoryError(f"Mahnstufe {level - 1} wurde noch nicht versandt.")
|
|
created_at = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
reminder = {
|
|
"reminder_id": str(uuid4()),
|
|
"reminder_key": f"{claim_id}:level-{level}",
|
|
"claim_id": claim_id,
|
|
"level": level,
|
|
"name": name.strip() or f"Mahnung Stufe {level}",
|
|
"status": "draft",
|
|
"detail": detail.strip(),
|
|
"channel": channel.strip() or "email",
|
|
"created_at": created_at,
|
|
"generated_at": None,
|
|
"sent_at": None,
|
|
"payment_deadline_days": payment_deadline_days,
|
|
"payment_deadline": None,
|
|
"balance_snapshot": money_text(claim_balance(data, claim)),
|
|
"items": normalized_items,
|
|
"fee_item_ids": [],
|
|
"document": {"path": "", "sha256": ""},
|
|
}
|
|
data.reminders.append(reminder)
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="reminder_draft_created",
|
|
summary=f"Mahnungsentwurf erstellt: {reminder['name']}",
|
|
references={"claim_id": claim_id, "reminder_id": str(reminder["reminder_id"])},
|
|
data={
|
|
"fee": money_text(reminder_items_total(reminder)),
|
|
"balance_snapshot": reminder["balance_snapshot"],
|
|
},
|
|
)
|
|
return reminder
|
|
|
|
def mark_reminder_sent(self, member_id: str, claim_id: str, reminder_id: str) -> dict:
|
|
data, claim = self.get_claim(member_id, claim_id)
|
|
reminder = self._find_reminder(data, claim_id, reminder_id)
|
|
if str(reminder.get("status", "")) not in {"draft", "generated"}:
|
|
raise RepositoryError("Nur ein Entwurf kann als versandt markiert werden.")
|
|
if claim_balance(data, claim) <= 0:
|
|
raise RepositoryError("Die Forderung hat keinen offenen Betrag mehr.")
|
|
now = datetime.now().astimezone()
|
|
if reminder.get("items") and not reminder.get("fee_item_ids"):
|
|
fee_item_ids = []
|
|
for entry in reminder["items"]:
|
|
amount = decimal_value(entry.get("amount", "0"))
|
|
if amount <= 0:
|
|
continue
|
|
description = str(entry.get("description")) or f"Mahngebühr Stufe {reminder.get('level', '')}"
|
|
item = {
|
|
"item_id": str(uuid4()),
|
|
"type": "fee",
|
|
"description": description,
|
|
"quantity": "1.00",
|
|
"unit_price": money_text(amount),
|
|
"amount": money_text(amount),
|
|
"created_at": now.isoformat(timespec="seconds"),
|
|
}
|
|
materialize_claim_items(claim).append(item)
|
|
fee_item_ids.append(item["item_id"])
|
|
if fee_item_ids:
|
|
claim["amount"] = money_text(claim_total(claim))
|
|
reminder["fee_item_ids"] = fee_item_ids
|
|
reminder["status"] = "sent"
|
|
reminder["sent_at"] = now.isoformat(timespec="seconds")
|
|
reminder["payment_deadline"] = (
|
|
now.date() + timedelta(days=int(reminder.get("payment_deadline_days", 14)))
|
|
).isoformat()
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="reminder_sent",
|
|
summary=f"Mahnung als versandt markiert: {reminder.get('name', '')}",
|
|
references={"claim_id": claim_id, "reminder_id": str(reminder["reminder_id"])},
|
|
data={
|
|
"fee": money_text(reminder_items_total(reminder)),
|
|
"payment_deadline": reminder["payment_deadline"],
|
|
},
|
|
)
|
|
return reminder
|
|
|
|
def cancel_reminder(self, member_id: str, claim_id: str, reminder_id: str) -> None:
|
|
data, _claim = self.get_claim(member_id, claim_id)
|
|
reminder = self._find_reminder(data, claim_id, reminder_id)
|
|
if str(reminder.get("status", "")) == "sent":
|
|
raise RepositoryError("Eine bereits versandte Mahnung kann nicht verworfen werden.")
|
|
reminder["status"] = "cancelled"
|
|
reminder["cancelled_at"] = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="reminder_cancelled",
|
|
summary=f"Mahnungsentwurf verworfen: {reminder.get('name', '')}",
|
|
references={"claim_id": claim_id, "reminder_id": reminder_id},
|
|
)
|
|
|
|
def register_reminder_document(
|
|
self,
|
|
member_id: str,
|
|
claim_id: str,
|
|
reminder_id: str,
|
|
*,
|
|
relative_path: str,
|
|
sha256: str,
|
|
template: str,
|
|
) -> None:
|
|
document_path = Path(relative_path)
|
|
if document_path.is_absolute() or ".." in document_path.parts:
|
|
raise RepositoryError("Der Dokumentpfad muss innerhalb des Mitglieder-Dateiordners liegen.")
|
|
data, _claim = self.get_claim(member_id, claim_id)
|
|
reminder = self._find_reminder(data, claim_id, reminder_id)
|
|
reminder["document"] = {
|
|
"path": relative_path,
|
|
"sha256": sha256,
|
|
"template": template,
|
|
"generated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
|
|
}
|
|
if str(reminder.get("status", "draft")) == "draft":
|
|
reminder["status"] = "generated"
|
|
reminder["generated_at"] = reminder["document"]["generated_at"]
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="reminder_document_generated",
|
|
summary=f"Mahndokument erzeugt: {Path(relative_path).name}",
|
|
actor_type="user",
|
|
actor_name="Vorstand",
|
|
references={
|
|
"claim_id": claim_id,
|
|
"reminder_id": reminder_id,
|
|
"document": relative_path,
|
|
},
|
|
data={"template": template, "sha256": sha256},
|
|
)
|
|
|
|
def set_dunning_hold(
|
|
self,
|
|
member_id: str,
|
|
claim_id: str,
|
|
*,
|
|
active: bool,
|
|
reason: str = "",
|
|
until: str = "",
|
|
) -> None:
|
|
data, claim = self.get_claim(member_id, claim_id)
|
|
try:
|
|
normalized_until = normalize_date_input(until, "Mahnsperre bis")
|
|
except DateValidationError as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
claim["dunning_hold"] = {
|
|
"active": active,
|
|
"reason": reason.strip(),
|
|
"until": normalized_until or None,
|
|
"updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
|
|
}
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="dunning_hold_changed",
|
|
summary="Mahnsperre gesetzt" if active else "Mahnsperre aufgehoben",
|
|
references={"claim_id": claim_id},
|
|
data={"reason": reason.strip(), "until": normalized_until},
|
|
)
|
|
|
|
@staticmethod
|
|
def _find_reminder(data: ContributionData, claim_id: str, reminder_id: str) -> dict:
|
|
reminder = next(
|
|
(
|
|
item
|
|
for item in data.reminders
|
|
if str(item.get("claim_id", "")) == claim_id
|
|
and str(item.get("reminder_id", "")) == reminder_id
|
|
),
|
|
None,
|
|
)
|
|
if reminder is None:
|
|
raise RepositoryError("Mahnung nicht gefunden.")
|
|
return reminder
|
|
|
|
def cancel_claim(self, member_id: str, claim_id: str) -> None:
|
|
data, claim = self.get_claim(member_id, claim_id)
|
|
if claim_balance(data, claim) != claim_total(claim):
|
|
raise RepositoryError("Eine Forderung mit Zahlungszuordnungen kann nicht storniert werden.")
|
|
claim["status"] = "cancelled"
|
|
claim["cancelled_at"] = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="claim_cancelled",
|
|
summary=f"Forderung storniert: {claim.get('title', claim_id)}",
|
|
references={"claim_id": claim_id},
|
|
)
|
|
|
|
def delete_claim(self, member_id: str, claim_id: str, *, actor_name: str = "Vorstand") -> None:
|
|
"""Permanently remove a claim. Any payments/credits allocated to it are released
|
|
(kept intact, just unlinked) rather than deleted, so they can be reallocated."""
|
|
data, claim = self.get_claim(member_id, claim_id)
|
|
released_allocations = [
|
|
allocation for allocation in data.allocations if str(allocation.get("claim_id", "")) == claim_id
|
|
]
|
|
released_total = sum(
|
|
(decimal_value(item.get("amount", "0")) for item in released_allocations), Decimal("0")
|
|
)
|
|
data.claims = [item for item in data.claims if str(item.get("claim_id", "")) != claim_id]
|
|
data.allocations = [
|
|
item for item in data.allocations if str(item.get("claim_id", "")) != claim_id
|
|
]
|
|
data.reminders = [
|
|
item for item in data.reminders if str(item.get("claim_id", "")) != claim_id
|
|
]
|
|
self.save_contributions(member_id, data)
|
|
self.append_event(
|
|
member_id,
|
|
event_type="claim_deleted",
|
|
summary=f"Forderung gelöscht: {claim.get('title', claim_id)}",
|
|
actor_type="user",
|
|
actor_name=actor_name,
|
|
references={"claim_id": claim_id},
|
|
data={
|
|
"amount": money_text(claim_total(claim)),
|
|
"released_allocations": money_text(released_total),
|
|
},
|
|
)
|
|
|
|
def _assert_gnucash_id_available(
|
|
self, transaction_id: str, *, exclude_payment_id: str | None = None
|
|
) -> None:
|
|
selected = transaction_id.casefold()
|
|
for member in self.list_members():
|
|
try:
|
|
payments = self.get_contributions(member.member_id).payments
|
|
except RepositoryError:
|
|
continue
|
|
if any(
|
|
str(payment.get("payment_id", "")) != exclude_payment_id
|
|
and str(payment.get("gnucash_transaction_id", "")).casefold() == selected
|
|
for payment in payments
|
|
):
|
|
raise RepositoryError(f"GnuCash-ID bereits verwendet: {transaction_id}")
|
|
|
|
def append_event(
|
|
self,
|
|
member_id: str,
|
|
*,
|
|
event_type: str,
|
|
summary: str,
|
|
actor_type: str = "system",
|
|
actor_name: str = "CCMA",
|
|
references: dict[str, str] | None = None,
|
|
data: dict[str, object] | None = None,
|
|
) -> Event:
|
|
directory = self._member_path(member_id)
|
|
if not (directory / "member.json").is_file():
|
|
raise RepositoryError(f"Mitglied nicht gefunden: {member_id}")
|
|
event = Event(
|
|
event_id=str(uuid4()),
|
|
timestamp=datetime.now().astimezone().isoformat(timespec="seconds"),
|
|
event_type=event_type,
|
|
summary=summary.strip(),
|
|
actor_type=actor_type,
|
|
actor_name=actor_name,
|
|
references=references or {},
|
|
data=data or {},
|
|
)
|
|
path = directory / "events.jsonl"
|
|
line = json.dumps(event.to_dict(), ensure_ascii=False, separators=(",", ":")) + "\n"
|
|
with path.open("a", encoding="utf-8", newline="\n") as handle:
|
|
handle.write(line)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
return event
|
|
|
|
def get_events(self, member_id: str) -> list[Event]:
|
|
path = self._member_path(member_id) / "events.jsonl"
|
|
if not path.exists():
|
|
return []
|
|
events: list[Event] = []
|
|
with path.open("r", encoding="utf-8") as handle:
|
|
for line_number, line in enumerate(handle, start=1):
|
|
if not line.strip():
|
|
continue
|
|
try:
|
|
raw = json.loads(line)
|
|
if not isinstance(raw, dict):
|
|
raise TypeError("Event muss ein JSON-Objekt sein")
|
|
events.append(Event.from_dict(raw))
|
|
except (AttributeError, ValueError, TypeError, KeyError, json.JSONDecodeError) as exc:
|
|
raise RepositoryError(f"Ungültiges Event in Zeile {line_number}: {exc}") from exc
|
|
return events
|
|
|
|
def search(self, query: str) -> list[Member]:
|
|
tokens = [_normalize(token) for token in query.split() if token.strip()]
|
|
if not tokens:
|
|
return self.list_members()
|
|
scored: list[tuple[int, Member]] = []
|
|
for member in self.list_members():
|
|
fields = [
|
|
member.member_number,
|
|
member.first_name,
|
|
member.last_name,
|
|
member.nickname,
|
|
member.display_name,
|
|
member.email,
|
|
member.phone,
|
|
member.street,
|
|
member.postal_code,
|
|
member.city,
|
|
member.birth_date,
|
|
_german_date(member.birth_date),
|
|
]
|
|
normalized = [_normalize(value) for value in fields if value]
|
|
if not all(any(token in value for value in normalized) for token in tokens):
|
|
continue
|
|
exact = sum(token == value for token in tokens for value in normalized)
|
|
prefix = sum(value.startswith(token) for token in tokens for value in normalized)
|
|
scored.append((exact * 100 + prefix * 10, member))
|
|
scored.sort(key=lambda item: (-item[0], item[1].last_name.casefold(), item[1].first_name.casefold()))
|
|
return [member for _, member in scored]
|
|
|
|
def member_count(self) -> int:
|
|
return sum(1 for _ in self._member_directories())
|
|
|
|
def asset_count(self) -> int:
|
|
return sum(1 for _ in self._asset_directories())
|
|
|
|
def get_configuration(self) -> dict:
|
|
try:
|
|
configuration = read_json(self.root / "repository.json")
|
|
except (OSError, ValueError, TypeError, json.JSONDecodeError) as exc:
|
|
raise RepositoryError(f"repository.json konnte nicht gelesen werden: {exc}") from exc
|
|
if not isinstance(configuration, dict):
|
|
raise RepositoryError("repository.json enthält keine gültige Konfiguration.")
|
|
return configuration
|
|
|
|
def member_hash_warnings(self, member_id: str) -> list[str]:
|
|
warnings: list[str] = []
|
|
try:
|
|
member_raw = read_json(self._member_path(member_id) / "member.json")
|
|
if not json_content_hash_matches(member_raw):
|
|
warnings.append(
|
|
"member.json: Hash fehlt oder stimmt nicht; Datei wurde vermutlich extern geändert."
|
|
)
|
|
except (OSError, ValueError, TypeError, json.JSONDecodeError):
|
|
pass
|
|
try:
|
|
contributions_raw = read_json(self._member_path(member_id) / "contributions.json")
|
|
if not json_content_hash_matches(contributions_raw):
|
|
warnings.append(
|
|
"contributions.json: Hash fehlt oder stimmt nicht; "
|
|
"Datei wurde vermutlich extern geändert."
|
|
)
|
|
except (OSError, ValueError, TypeError, json.JSONDecodeError):
|
|
pass
|
|
return warnings
|
|
|
|
def asset_hash_warnings(self, asset_id: str) -> list[str]:
|
|
warnings: list[str] = []
|
|
try:
|
|
asset_raw = read_json(self._asset_path(asset_id) / "asset.json")
|
|
if not json_content_hash_matches(asset_raw):
|
|
warnings.append(
|
|
"asset.json: Hash fehlt oder stimmt nicht; Datei wurde vermutlich extern geändert."
|
|
)
|
|
except (OSError, ValueError, TypeError, json.JSONDecodeError):
|
|
pass
|
|
return warnings
|
|
|
|
def refresh_member_record_hashes(self, member_id: str) -> None:
|
|
member = self.get_member(member_id)
|
|
contributions = self.get_contributions(member_id)
|
|
write_json_atomic(self._member_path(member_id) / "member.json", member.to_dict())
|
|
write_json_atomic(self._member_path(member_id) / "contributions.json", contributions.to_dict())
|
|
|
|
def refresh_asset_record_hashes(self, asset_id: str) -> None:
|
|
asset = self.get_asset(asset_id)
|
|
write_json_atomic(self._asset_path(asset_id) / "asset.json", asset.to_dict())
|
|
|
|
def get_member_number_policy(self) -> dict[str, str]:
|
|
try:
|
|
config = read_json(self.root / "repository.json")
|
|
except (OSError, ValueError, TypeError, json.JSONDecodeError) as exc:
|
|
raise RepositoryError(f"Mitgliedsnummernregel konnte nicht gelesen werden: {exc}") from exc
|
|
policy = config.get("member_number_policy") or {}
|
|
mode = str(policy.get("mode", "automatic"))
|
|
if mode not in {"automatic", "manual"}:
|
|
mode = "automatic"
|
|
pattern = str(policy.get("pattern", DEFAULT_MEMBER_NUMBER_PATTERN))
|
|
allocation_strategy = str(policy.get("allocation_strategy", "next_free"))
|
|
if allocation_strategy not in {"next_free", "max_plus_one"}:
|
|
allocation_strategy = "next_free"
|
|
validate_member_number_pattern(pattern)
|
|
return {
|
|
"mode": mode,
|
|
"pattern": pattern,
|
|
"allocation_strategy": allocation_strategy,
|
|
}
|
|
|
|
def save_member_number_policy(
|
|
self, *, mode: str, pattern: str, allocation_strategy: str = "next_free"
|
|
) -> None:
|
|
if mode not in {"automatic", "manual"}:
|
|
raise RepositoryError("Ungültiger Mitgliedsnummernmodus.")
|
|
if allocation_strategy not in {"next_free", "max_plus_one"}:
|
|
raise RepositoryError("Ungültige Vergabestrategie für Mitgliedsnummern.")
|
|
validate_member_number_pattern(pattern)
|
|
config = read_json(self.root / "repository.json")
|
|
config["member_number_policy"] = {
|
|
"mode": mode,
|
|
"pattern": pattern.strip(),
|
|
"allocation_strategy": allocation_strategy,
|
|
}
|
|
config.setdefault("member_number_sequences", {})
|
|
write_json_atomic(self.root / "repository.json", config)
|
|
|
|
def get_reminder_policy(self) -> dict:
|
|
config = self.get_configuration()
|
|
policy = config.get("reminder_policy") or {}
|
|
levels = sorted(
|
|
(
|
|
{
|
|
"level": int(entry.get("level", 0)),
|
|
"name": str(entry.get("name", "")),
|
|
"fee": money_text(entry.get("fee", "0")),
|
|
"payment_deadline_days": int(entry.get("payment_deadline_days", 14)),
|
|
}
|
|
for entry in policy.get("levels") or []
|
|
),
|
|
key=lambda entry: entry["level"],
|
|
)
|
|
# Repositories created before this setting existed have no "standard_fee_items"
|
|
# key at all -- fall back to the built-in defaults rather than an empty list, but
|
|
# once the board has actually saved (and possibly emptied) the list, respect that.
|
|
if "standard_fee_items" in policy:
|
|
standard_items_raw = policy.get("standard_fee_items") or []
|
|
else:
|
|
standard_items_raw = DEFAULT_CONFIGURATION["reminder_policy"]["standard_fee_items"]
|
|
standard_fee_items = [
|
|
{
|
|
"description": str(entry.get("description", "")),
|
|
"default_amount": money_text(entry.get("default_amount", "0")),
|
|
}
|
|
for entry in standard_items_raw
|
|
]
|
|
return {
|
|
"grace_days_after_due": int(policy.get("grace_days_after_due", 7)),
|
|
"levels": levels,
|
|
"standard_fee_items": standard_fee_items,
|
|
}
|
|
|
|
def save_reminder_policy(
|
|
self,
|
|
*,
|
|
grace_days_after_due: int,
|
|
levels: list[dict[str, str]],
|
|
standard_fee_items: list[dict[str, str]],
|
|
) -> None:
|
|
if grace_days_after_due < 0 or grace_days_after_due > 365:
|
|
raise RepositoryError("Die Kulanzfrist muss zwischen 0 und 365 Tagen liegen.")
|
|
normalized_levels = []
|
|
for index, entry in enumerate(levels, start=1):
|
|
name = str(entry.get("name", "")).strip()
|
|
if not name:
|
|
raise RepositoryError(f"Mahnstufe {index} benötigt einen Namen.")
|
|
try:
|
|
fee = decimal_value(entry.get("fee", "0"), "Gebühr")
|
|
except ValueError as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if fee < 0:
|
|
raise RepositoryError("Die Gebühr darf nicht negativ sein.")
|
|
try:
|
|
deadline_days = int(entry.get("payment_deadline_days", 14))
|
|
except (TypeError, ValueError) as exc:
|
|
raise RepositoryError("Die Zahlungsfrist muss eine ganze Zahl sein.") from exc
|
|
if deadline_days < 1 or deadline_days > 365:
|
|
raise RepositoryError("Die Zahlungsfrist muss zwischen 1 und 365 Tagen liegen.")
|
|
# Levels are always renumbered 1..N by row order -- picking arbitrary level
|
|
# numbers would break the sequencing checks in create_reminder_draft.
|
|
normalized_levels.append(
|
|
{"level": index, "name": name, "fee": money_text(fee), "payment_deadline_days": deadline_days}
|
|
)
|
|
if not normalized_levels:
|
|
raise RepositoryError("Mindestens eine Mahnstufe ist erforderlich.")
|
|
normalized_items = []
|
|
for entry in standard_fee_items:
|
|
description = str(entry.get("description", "")).strip()
|
|
if not description:
|
|
raise RepositoryError("Jede Standardposition benötigt eine Beschreibung.")
|
|
try:
|
|
amount = decimal_value(entry.get("default_amount", "0"), "Standardbetrag")
|
|
except ValueError as exc:
|
|
raise RepositoryError(str(exc)) from exc
|
|
if amount < 0:
|
|
raise RepositoryError("Der Standardbetrag darf nicht negativ sein.")
|
|
normalized_items.append({"description": description, "default_amount": money_text(amount)})
|
|
config = self.get_configuration()
|
|
config["reminder_policy"] = {
|
|
"grace_days_after_due": grace_days_after_due,
|
|
"levels": normalized_levels,
|
|
"standard_fee_items": normalized_items,
|
|
}
|
|
write_json_atomic(self.root / "repository.json", config)
|
|
|
|
def save_organization(self, values: dict[str, str]) -> None:
|
|
organization = {key: str(value).strip() for key, value in values.items()}
|
|
organization["iban"] = normalize_iban(organization.get("iban", ""))
|
|
organization["bic"] = organization.get("bic", "").replace(" ", "").upper()
|
|
validate_iban(organization["iban"])
|
|
validate_bic(organization["bic"])
|
|
if not organization.get("name"):
|
|
raise RepositoryError("Der Vereinsname ist erforderlich.")
|
|
config = self.get_configuration()
|
|
config["organization"] = organization
|
|
write_json_atomic(self.root / "repository.json", config)
|
|
|
|
def preview_member_number(
|
|
self, pattern: str | None = None, allocation_strategy: str | None = None
|
|
) -> str:
|
|
policy = self.get_member_number_policy()
|
|
selected_pattern = pattern or policy["pattern"]
|
|
selected_strategy = allocation_strategy or policy["allocation_strategy"]
|
|
validate_member_number_pattern(selected_pattern)
|
|
config = read_json(self.root / "repository.json")
|
|
return self._next_available_member_number(config, selected_pattern, selected_strategy)[0]
|
|
|
|
def _member_directories(self) -> Iterable[Path]:
|
|
if not self.members_root.exists():
|
|
return []
|
|
return (
|
|
path for path in self.members_root.iterdir() if path.is_dir() and not path.name.startswith(".")
|
|
)
|
|
|
|
def _asset_directories(self) -> Iterable[Path]:
|
|
if not self.assets_root.exists():
|
|
return []
|
|
return (
|
|
path for path in self.assets_root.iterdir() if path.is_dir() and not path.name.startswith(".")
|
|
)
|
|
|
|
def _member_path(self, member_id: str) -> Path:
|
|
if not member_id or Path(member_id).name != member_id or member_id in {".", ".."}:
|
|
raise RepositoryError("Ungültige Mitglieds-ID.")
|
|
return self.members_root / member_id
|
|
|
|
def _asset_path(self, asset_id: str) -> Path:
|
|
if not asset_id or Path(asset_id).name != asset_id or asset_id in {".", ".."}:
|
|
raise RepositoryError("Ungültige Asset-ID.")
|
|
return self.assets_root / asset_id
|
|
|
|
def _allocate_member_number(self, pattern: str, allocation_strategy: str) -> str:
|
|
config = read_json(self.root / "repository.json")
|
|
member_number, next_value = self._next_available_member_number(
|
|
config, pattern, allocation_strategy
|
|
)
|
|
sequences = config.get("member_number_sequences")
|
|
if not isinstance(sequences, dict):
|
|
sequences = {}
|
|
config["member_number_sequences"] = sequences
|
|
sequences[pattern] = next_value
|
|
write_json_atomic(self.root / "repository.json", config)
|
|
return member_number
|
|
|
|
def _next_available_member_number(
|
|
self, config: dict, pattern: str, allocation_strategy: str
|
|
) -> tuple[str, int]:
|
|
if allocation_strategy not in {"next_free", "max_plus_one"}:
|
|
raise RepositoryError("Ungültige Vergabestrategie für Mitgliedsnummern.")
|
|
existing = {member.member_number.casefold() for member in self.list_members() if member.member_number}
|
|
number = 1
|
|
if allocation_strategy == "max_plus_one":
|
|
used_values = (
|
|
value
|
|
for member_number in existing
|
|
if (value := _number_from_formatted_member_number(pattern, member_number)) is not None
|
|
)
|
|
number = max(used_values, default=0) + 1
|
|
for _attempt in range(1_000_000):
|
|
candidate = format_member_number(pattern, number)
|
|
number += 1
|
|
if candidate.casefold() not in existing:
|
|
return candidate, number
|
|
raise RepositoryError("Keine freie Mitgliedsnummer im konfigurierten Nummernbereich gefunden.")
|
|
|
|
@contextmanager
|
|
def _member_number_lock(self):
|
|
lock_path = self.root / ".member-number.lock"
|
|
lock_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with lock_path.open("a+", encoding="utf-8") as handle:
|
|
if os.name == "nt":
|
|
import msvcrt
|
|
|
|
handle.seek(0)
|
|
if not handle.read(1):
|
|
handle.write("\0")
|
|
handle.flush()
|
|
handle.seek(0)
|
|
msvcrt.locking(handle.fileno(), msvcrt.LK_LOCK, 1)
|
|
else:
|
|
import fcntl
|
|
|
|
fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
|
|
try:
|
|
yield
|
|
finally:
|
|
if os.name == "nt":
|
|
handle.seek(0)
|
|
msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
|
|
else:
|
|
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
|
|
|
|
def _assert_member_number_available(
|
|
self,
|
|
member_number: str,
|
|
*,
|
|
exclude_member_id: str | None = None,
|
|
) -> None:
|
|
normalized = member_number.casefold().strip()
|
|
if not normalized:
|
|
raise RepositoryError("Eine Mitgliedsnummer ist erforderlich.")
|
|
for member in self.list_members():
|
|
if member.member_id != exclude_member_id and member.member_number.casefold() == normalized:
|
|
raise RepositoryError(f"Die Mitgliedsnummer {member_number} ist bereits vergeben.")
|
|
|
|
@staticmethod
|
|
def _summarize_changes(before: Member, after: Member) -> list[str]:
|
|
labels = {
|
|
"member_number": "Mitgliedsnummer",
|
|
"first_name": "Vorname",
|
|
"last_name": "Nachname",
|
|
"nickname": "Nickname",
|
|
"email": "E-Mail-Adresse",
|
|
"phone": "Telefonnummer",
|
|
"birth_date": "Geburtsdatum",
|
|
"application_date": "Antragsdatum",
|
|
"membership_ended_at": "Austrittsdatum",
|
|
"status": "Status",
|
|
"payment_frequency": "Zahlungsweise",
|
|
"contribution_rule_id": "Beitragsregel",
|
|
"honorary": "Ehrenmitgliedschaft",
|
|
"notes": "interne Notiz",
|
|
"street": "Anschrift",
|
|
"address_addition": "Adresszusatz",
|
|
"postal_code": "Postleitzahl",
|
|
"city": "Ort",
|
|
"country": "Land",
|
|
"account_holder": "Kontoinhaber",
|
|
"iban": "IBAN",
|
|
"bic": "BIC",
|
|
"mandate_reference": "SEPA-Mandatsreferenz",
|
|
"mandate_signed_at": "Mandat erteilt am",
|
|
"mandate_active": "Lastschriftmandat aktiv",
|
|
"mandate_revoked_at": "Mandat widerrufen am",
|
|
}
|
|
changes: list[str] = []
|
|
for field, label in labels.items():
|
|
old_value = getattr(before, field)
|
|
new_value = getattr(after, field)
|
|
if old_value == new_value:
|
|
continue
|
|
if field == "status":
|
|
old_label = MEMBERSHIP_STATUS_LABELS.get(str(old_value), str(old_value))
|
|
new_label = MEMBERSHIP_STATUS_LABELS.get(str(new_value), str(new_value))
|
|
changes.append(f"Status von {old_label} zu {new_label}")
|
|
else:
|
|
changes.append(label)
|
|
return changes
|
|
|
|
@staticmethod
|
|
def _summarize_asset_changes(before: Asset, after: Asset) -> list[str]:
|
|
labels = {
|
|
"label": "Bezeichnung",
|
|
"category": "Kategorie",
|
|
"inventory_number": "Inventarnummer",
|
|
"serial_number": "Seriennummer",
|
|
"status": "Status",
|
|
"current_holder_member_id": "Zuordnung",
|
|
"owner_type": "Eigentümertyp",
|
|
"owner_member_id": "Eigentümer",
|
|
"owner_name": "Externer Eigentümer",
|
|
"custody_type": "Gewahrsam",
|
|
"location": "Standort",
|
|
"handed_over_at": "Übergabedatum",
|
|
"condition": "Zustand",
|
|
"estimated_value": "Wert",
|
|
"deposit_amount_default": "Kaution",
|
|
"notes": "Notiz",
|
|
}
|
|
changes: list[str] = []
|
|
for field, label in labels.items():
|
|
if getattr(before, field) != getattr(after, field):
|
|
changes.append(label)
|
|
return changes
|
|
|
|
|
|
def normalize_iban(value: str) -> str:
|
|
return "".join(value.split()).upper()
|
|
|
|
|
|
def validate_iban(value: str) -> None:
|
|
if not value:
|
|
return
|
|
if not 15 <= len(value) <= 34 or not value[:2].isalpha() or not value[2:].isalnum():
|
|
raise RepositoryError("Die IBAN hat kein gültiges Format.")
|
|
rearranged = value[4:] + value[:4]
|
|
numeric = "".join(
|
|
str(ord(character) - 55) if character.isalpha() else character for character in rearranged
|
|
)
|
|
if int(numeric) % 97 != 1:
|
|
raise RepositoryError("Die IBAN-Prüfsumme ist ungültig.")
|
|
|
|
|
|
def validate_bic(value: str) -> None:
|
|
if value and (len(value) not in {8, 11} or not value.isalnum()):
|
|
raise RepositoryError("Der BIC muss 8 oder 11 Zeichen enthalten.")
|
|
|
|
|
|
def _normalize(value: str) -> str:
|
|
normalized = unicodedata.normalize("NFKD", value.casefold().strip())
|
|
return "".join(character for character in normalized if not unicodedata.combining(character))
|
|
|
|
|
|
def _german_date(value: str) -> str:
|
|
try:
|
|
return date.fromisoformat(value).strftime("%d.%m.%Y")
|
|
except ValueError:
|
|
return ""
|
|
|
|
|
|
def _contribution_override_summary(override: dict) -> str:
|
|
value = str(override.get("value", ""))
|
|
value_text = f"{value}%" if str(override.get("kind", "")) == "percent" else f"{value} EUR"
|
|
valid_from = str(override.get("valid_from", ""))
|
|
valid_until = str(override.get("valid_until", "") or "")
|
|
period = f"{valid_from} bis {valid_until}" if valid_until else f"ab {valid_from}"
|
|
return f"{value_text} ({period})"
|
|
|
|
|
|
def _normalize_reminder_items(items: list[dict[str, str]] | None) -> list[dict[str, str]]:
|
|
normalized = []
|
|
for entry in items or []:
|
|
description = str(entry.get("description", "")).strip()
|
|
if not description:
|
|
raise ValueError("Jede Position benötigt eine Beschreibung.")
|
|
amount = decimal_value(entry.get("amount", "0"), "Betrag")
|
|
if amount < 0:
|
|
raise ValueError("Der Betrag darf nicht negativ sein.")
|
|
normalized.append({"description": description, "amount": money_text(amount)})
|
|
return normalized
|
|
|
|
|
|
def _dunning_hold_is_active(claim: dict) -> bool:
|
|
hold = claim.get("dunning_hold") or {}
|
|
if not hold.get("active"):
|
|
return False
|
|
if not hold.get("until"):
|
|
return True
|
|
try:
|
|
return date.fromisoformat(str(hold["until"])) >= date.today()
|
|
except ValueError:
|
|
return True
|
|
|
|
|
|
def validate_member_number_pattern(pattern: str) -> None:
|
|
selected = pattern.strip()
|
|
if not selected:
|
|
raise RepositoryError("Das Mitgliedsnummern-Pattern darf nicht leer sein.")
|
|
has_number = False
|
|
try:
|
|
for _literal, field_name, format_spec, conversion in Formatter().parse(selected):
|
|
if field_name is None:
|
|
continue
|
|
if field_name not in {"number", "year"}:
|
|
raise RepositoryError(f"Unbekannter Platzhalter: {{{field_name}}}")
|
|
if conversion:
|
|
raise RepositoryError("Konvertierungen wie !r sind im Pattern nicht erlaubt.")
|
|
if "{" in format_spec or "}" in format_spec:
|
|
raise RepositoryError("Verschachtelte Formatierungen sind nicht erlaubt.")
|
|
has_number = has_number or field_name == "number"
|
|
if not has_number:
|
|
raise RepositoryError("Das Pattern muss den Platzhalter {number} enthalten.")
|
|
first = format_member_number(selected, 1)
|
|
second = format_member_number(selected, 2)
|
|
if first == second:
|
|
raise RepositoryError("Das Pattern erzeugt keine eindeutigen Mitgliedsnummern.")
|
|
except RepositoryError:
|
|
raise
|
|
except (KeyError, ValueError) as exc:
|
|
raise RepositoryError(f"Ungültiges Mitgliedsnummern-Pattern: {exc}") from exc
|
|
|
|
|
|
def _number_from_formatted_member_number(pattern: str, member_number: str) -> int | None:
|
|
parts: list[str] = []
|
|
number_groups = 0
|
|
for literal, field_name, _format_spec, _conversion in Formatter().parse(pattern.strip()):
|
|
parts.append(re.escape(literal))
|
|
if field_name == "year":
|
|
parts.append(re.escape(str(date.today().year)))
|
|
elif field_name == "number":
|
|
parts.append(r"(\d+)")
|
|
number_groups += 1
|
|
match = re.fullmatch("".join(parts), member_number, flags=re.IGNORECASE)
|
|
if not match or not number_groups:
|
|
return None
|
|
values = [int(value) for value in match.groups()]
|
|
if any(value != values[0] for value in values[1:]):
|
|
return None
|
|
value = values[0]
|
|
return value if format_member_number(pattern, value).casefold() == member_number.casefold() else None
|
|
|
|
|
|
def format_member_number(pattern: str, number: int, *, year: int | None = None) -> str:
|
|
try:
|
|
value = pattern.strip().format(number=number, year=year or date.today().year)
|
|
except (KeyError, ValueError) as exc:
|
|
raise RepositoryError(f"Mitgliedsnummer konnte nicht formatiert werden: {exc}") from exc
|
|
if not value or len(value) > 80 or any(character in value for character in "\r\n\t"):
|
|
raise RepositoryError("Das Pattern erzeugt eine ungültige Mitgliedsnummer.")
|
|
return value
|