feat: allow deleting housekeeper tasks

This commit is contained in:
Marcel Peterkau
2026-06-21 18:09:04 +02:00
parent 3e9f347435
commit c717d6806b
6 changed files with 110 additions and 2 deletions
+31 -1
View File
@@ -64,10 +64,12 @@ class Housekeeper:
now = datetime.now().astimezone().isoformat(timespec="seconds")
items = _items_by_key(working)
successful_scopes: set[tuple[str, str]] = set()
member_ids = set(self.repository.list_member_ids())
_remove_orphaned_member_items(items, member_ids)
rules = load_rules(self.repository.root)
repository_config = self.repository.get_configuration()
for member_id in self.repository.list_member_ids():
for member_id in sorted(member_ids):
try:
member, contributions = self.repository.preflight_member_record(member_id)
except RepositoryError as exc:
@@ -126,6 +128,23 @@ class Housekeeper:
write_json_atomic(self.state_path, working)
return _open_findings(working["items"])
def delete_task(self, key: str) -> list[HousekeeperFinding]:
selected_key = key.strip()
if not selected_key:
raise RepositoryError("Der Task hat keinen gültigen Key.")
with _exclusive_lock(self.lock_path):
state = self._load_state()
items = _items_by_key(state)
item = items.get(selected_key)
if item is None:
return _open_findings(list(items.values()))
if item.get("action") != "task":
raise RepositoryError("Nur Hausmeister-Tasks können manuell gelöscht werden.")
del items[selected_key]
state["items"] = sorted(items.values(), key=lambda value: str(value.get("key", "")))
write_json_atomic(self.state_path, state)
return _open_findings(state["items"])
@staticmethod
def _refresh_record_integrity_task(
items: dict[str, dict[str, Any]],
@@ -351,6 +370,7 @@ def _open_findings(items: list[dict[str, Any]]) -> list[HousekeeperFinding]:
title=str(item.get("title", item.get("key", "Hausmeister"))),
detail=str(item.get("detail", "")),
due_date=due_date,
key=str(item.get("key", "")),
)
)
severity_order = {"error": 0, "warning": 1, "info": 2}
@@ -360,6 +380,16 @@ def _open_findings(items: list[dict[str, Any]]) -> list[HousekeeperFinding]:
)
def _remove_orphaned_member_items(items: dict[str, dict[str, Any]], member_ids: set[str]) -> None:
orphaned_keys = [
key
for key, item in items.items()
if item.get("member_id") and str(item["member_id"]) not in member_ids
]
for key in orphaned_keys:
del items[key]
@contextmanager
def _exclusive_lock(path: Path):
path.parent.mkdir(parents=True, exist_ok=True)