Souko/CCMA
1
0
Fork 0
mirror of https://git.hiabuto.net/C3MA/CCMA.git synced 2026-08-25 06:55:17 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat: configure member number allocation strategy

Browse Source
This commit is contained in:
Marcel Peterkau
2026-07-30 19:07:18 +02:00
parent 460dcac77c
commit cfd5de42f2
3 changed files with 190 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ccma/storage/repository.py
+119 -17
View File
@@ -2,9 +2,11 @@ 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
@@ -93,6 +95,7 @@ DEFAULT_CONFIGURATION = {
"member_number_policy": {
"mode": "automatic",
"pattern": DEFAULT_MEMBER_NUMBER_PATTERN,
"allocation_strategy": "next_free",
},
"member_number_sequences": {},
"reminder_policy": {
@@ -160,6 +163,8 @@ class MemberRepository:
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}")
@@ -290,6 +295,28 @@ class MemberRepository:
phone: str = "",
birth_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,
member_number=member_number,
)
def _create_member_locked(
self,
*,
first_name: str,
last_name: str,
nickname: str,
email: str,
phone: str,
birth_date: str,
member_number: str,
) -> Member:
if not first_name.strip() or not last_name.strip():
raise RepositoryError("Vorname und Nachname sind erforderlich.")
@@ -305,7 +332,9 @@ class MemberRepository:
elif policy["mode"] == "manual":
raise RepositoryError("Eine Mitgliedsnummer ist erforderlich.")
else:
selected_number = self._allocate_member_number(policy["pattern"])
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)
@@ -1703,15 +1732,30 @@ class MemberRepository:
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}
return {
"mode": mode,
"pattern": pattern,
"allocation_strategy": allocation_strategy,
}
def save_member_number_policy(self, *, mode: str, pattern: str) -> None:
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()}
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)
@@ -1727,11 +1771,15 @@ class MemberRepository:
config["organization"] = organization
write_json_atomic(self.root / "repository.json", config)
def preview_member_number(self, pattern: str | None = None) -> str:
selected_pattern = pattern or self.get_member_number_policy()["pattern"]
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)[0]
return self._next_available_member_number(config, selected_pattern, selected_strategy)[0]
def _member_directories(self) -> Iterable[Path]:
if not self.members_root.exists():
@@ -1757,9 +1805,11 @@ class MemberRepository:
raise RepositoryError("Ungültige Asset-ID.")
return self.assets_root / asset_id
def _allocate_member_number(self, pattern: str) -> str:
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)
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 = {}
@@ -1768,15 +1818,20 @@ class MemberRepository:
write_json_atomic(self.root / "repository.json", config)
return member_number
def _next_available_member_number(self, config: dict, pattern: str) -> tuple[str, int]:
sequences = config.get("member_number_sequences")
if not isinstance(sequences, dict):
sequences = {}
try:
number = max(1, int(sequences.get(pattern, 1)))
except (TypeError, ValueError):
number = 1
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
@@ -1784,6 +1839,33 @@ class MemberRepository:
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,
@@ -1941,6 +2023,26 @@ def validate_member_number_pattern(pattern: str) -> None:
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)