|
| 1 | +"""Appeals: a child can ask for a second look at a rejection or a penalty. |
| 2 | +
|
| 3 | +With ``appeals_enabled`` on, a child's card lists what they may appeal: a |
| 4 | +chore or reward claim a parent rejected, or a penalty applied to them, from |
| 5 | +the last ``APPEAL_WINDOW_HOURS`` hours. One appeal per item, with an optional |
| 6 | +message. The parents get a push (``appeal_submitted``, on by default like a |
| 7 | +help request) and the appeal waits on the panel's Today page. |
| 8 | +
|
| 9 | +Deciding it: |
| 10 | + * **Uphold** — nothing changes; the child sees the parent's note. |
| 11 | + * **Overturn** — a rejected chore pays its points, a rejected reward claim |
| 12 | + is approved there and then, a penalty is undone (and any open make-it-right |
| 13 | + offer for it is closed). |
| 14 | +
|
| 15 | +A penalty the child has already earned back with its make-it-right chore |
| 16 | +can't be appealed: it has been put right already. Wishlist redemptions keep |
| 17 | +their own decline flow and are never appealable. |
| 18 | +""" |
| 19 | + |
| 20 | +from __future__ import annotations |
| 21 | + |
| 22 | +import logging |
| 23 | +from datetime import datetime, timedelta |
| 24 | +from typing import Any |
| 25 | + |
| 26 | +from homeassistant.util import dt as dt_util |
| 27 | + |
| 28 | +from . import notify_strings |
| 29 | +from .const import NOTIF_TYPE_APPEAL_DECIDED, NOTIF_TYPE_APPEAL_SUBMITTED |
| 30 | +from .models import RewardClaim, clean_label, generate_id, parse_datetime |
| 31 | + |
| 32 | +_LOGGER = logging.getLogger(__name__) |
| 33 | + |
| 34 | +APPEAL_WINDOW_HOURS = 48 |
| 35 | +APPEAL_MESSAGE_MAX = 200 |
| 36 | +APPEAL_NOTE_MAX = 200 |
| 37 | +APPEAL_KINDS = ("chore", "reward", "penalty") |
| 38 | +# Decided appeals stay on the child's card this long. |
| 39 | +APPEAL_SHOW_DECIDED_HOURS = 24 |
| 40 | +APPEAL_KEEP = 100 |
| 41 | +# How close a make-it-right offer's creation must be to the penalty's |
| 42 | +# transaction to count as that penalty's offer. |
| 43 | +_OFFER_MATCH = timedelta(minutes=2) |
| 44 | +_PENALTY_PREFIX = "Penalty: " |
| 45 | + |
| 46 | + |
| 47 | +def _when(value) -> datetime | None: |
| 48 | + try: |
| 49 | + return parse_datetime(value) |
| 50 | + except (TypeError, ValueError): |
| 51 | + return None |
| 52 | + |
| 53 | + |
| 54 | +class AppealsMixin: |
| 55 | + """Children appeal rejections and penalties; parents uphold or overturn.""" |
| 56 | + |
| 57 | + def appeals_enabled(self) -> bool: |
| 58 | + v = self.storage.get_setting("appeals_enabled", False) |
| 59 | + return v is True or str(v).lower() == "true" |
| 60 | + |
| 61 | + def _appeals(self) -> list[dict[str, Any]]: |
| 62 | + return self.storage.get_section("appeals", list) |
| 63 | + |
| 64 | + def _appealed_targets(self) -> set[str]: |
| 65 | + return {a.get("target_id") for a in self._appeals()} |
| 66 | + |
| 67 | + def _penalty_offer(self, child_id: str, name: str, at: datetime) -> dict[str, Any] | None: |
| 68 | + """The make-it-right offer that penalty transaction created, if any.""" |
| 69 | + for offer in self.storage.get_section("penalty_redemptions", list): |
| 70 | + if offer.get("child_id") != child_id or offer.get("penalty_name") != name: |
| 71 | + continue |
| 72 | + created = _when(offer.get("created_at")) |
| 73 | + if created is not None and abs(created - at) <= _OFFER_MATCH: |
| 74 | + return offer |
| 75 | + return None |
| 76 | + |
| 77 | + def appealable_for_child(self, child_id: str) -> list[dict[str, Any]]: |
| 78 | + """What this child may still appeal, newest first.""" |
| 79 | + if not self.appeals_enabled(): |
| 80 | + return [] |
| 81 | + now = dt_util.now() |
| 82 | + cutoff = now - timedelta(hours=APPEAL_WINDOW_HOURS) |
| 83 | + taken = self._appealed_targets() |
| 84 | + out: list[dict[str, Any]] = [] |
| 85 | + for row in self.storage.get_rejections(): |
| 86 | + if row.get("child_id") != child_id or row.get("wish") or row.get("kind") not in ("chore", "reward"): |
| 87 | + continue |
| 88 | + at = _when(row.get("rejected_at")) |
| 89 | + if at is None or at < cutoff or row.get("id") in taken: |
| 90 | + continue |
| 91 | + out.append( |
| 92 | + { |
| 93 | + "kind": row["kind"], |
| 94 | + "target_id": row["id"], |
| 95 | + "item_id": row.get("item_id", ""), |
| 96 | + "name": row.get("item_name", ""), |
| 97 | + "points": int(row.get("points", 0) or 0), |
| 98 | + "at": at.isoformat(), |
| 99 | + } |
| 100 | + ) |
| 101 | + for txn in self.storage.get_points_transactions(): |
| 102 | + if txn.child_id != child_id or not (txn.reason or "").startswith(_PENALTY_PREFIX) or txn.points >= 0: |
| 103 | + continue |
| 104 | + if txn.created_at < cutoff or txn.id in taken: |
| 105 | + continue |
| 106 | + name = txn.reason[len(_PENALTY_PREFIX) :] |
| 107 | + offer = self._penalty_offer(child_id, name, txn.created_at) |
| 108 | + if offer is not None and offer.get("status") == "redeemed": |
| 109 | + continue |
| 110 | + out.append( |
| 111 | + { |
| 112 | + "kind": "penalty", |
| 113 | + "target_id": txn.id, |
| 114 | + "item_id": "", |
| 115 | + "name": name, |
| 116 | + "points": -txn.points, |
| 117 | + "at": txn.created_at.isoformat(), |
| 118 | + } |
| 119 | + ) |
| 120 | + out.sort(key=lambda i: i["at"], reverse=True) |
| 121 | + return out[:5] |
| 122 | + |
| 123 | + async def async_appeal(self, child_id: str, kind: str, target_id: str, message: str = "") -> dict[str, Any]: |
| 124 | + """A child appeals one item from their appealable list.""" |
| 125 | + if not self.appeals_enabled(): |
| 126 | + raise ValueError("Appeals are switched off.") |
| 127 | + child = self.get_child(child_id) |
| 128 | + if child is None: |
| 129 | + raise ValueError(f"Child {child_id} not found") |
| 130 | + if target_id in self._appealed_targets(): |
| 131 | + raise ValueError("That has already been appealed.") |
| 132 | + item = next( |
| 133 | + (i for i in self.appealable_for_child(child_id) if i["kind"] == kind and i["target_id"] == target_id), None |
| 134 | + ) |
| 135 | + if item is None: |
| 136 | + raise ValueError("That can't be appealed.") |
| 137 | + now = dt_util.now() |
| 138 | + record = { |
| 139 | + "id": generate_id(), |
| 140 | + "child_id": child_id, |
| 141 | + "kind": kind, |
| 142 | + "target_id": target_id, |
| 143 | + "item_id": item["item_id"], |
| 144 | + "item_name": item["name"], |
| 145 | + "points": item["points"], |
| 146 | + "message": clean_label(message, APPEAL_MESSAGE_MAX), |
| 147 | + "status": "open", |
| 148 | + "created_at": now.isoformat(), |
| 149 | + } |
| 150 | + appeals = self._appeals() |
| 151 | + appeals.append(record) |
| 152 | + _trim(appeals) |
| 153 | + await self.storage.async_save() |
| 154 | + payload = { |
| 155 | + "appeal_id": record["id"], |
| 156 | + "child_id": child_id, |
| 157 | + "child_name": child.name, |
| 158 | + "kind": kind, |
| 159 | + "item_name": record["item_name"], |
| 160 | + "points": record["points"], |
| 161 | + "message": record["message"], |
| 162 | + } |
| 163 | + self.hass.bus.async_fire("taskmate_appeal_submitted", {**payload, "timestamp": now.isoformat()}) |
| 164 | + if getattr(self, "notifications", None): |
| 165 | + await self.notifications.fire( |
| 166 | + NOTIF_TYPE_APPEAL_SUBMITTED, |
| 167 | + {**payload, "message_text": f" — “{record['message']}”" if record["message"] else ""}, |
| 168 | + ) |
| 169 | + await self.async_refresh() |
| 170 | + return dict(record) |
| 171 | + |
| 172 | + def open_appeals(self) -> list[dict[str, Any]]: |
| 173 | + """Appeals waiting for a parent, oldest first, with the child's name.""" |
| 174 | + names = {c.id: c.name for c in self.storage.get_children()} |
| 175 | + return [ |
| 176 | + {**a, "child_name": names.get(a.get("child_id"), "")} for a in self._appeals() if a.get("status") == "open" |
| 177 | + ] |
| 178 | + |
| 179 | + def appeals_for_child(self, child_id: str) -> list[dict[str, Any]]: |
| 180 | + """This child's open appeals and the ones decided in the last day, compact, newest first.""" |
| 181 | + cutoff = dt_util.now() - timedelta(hours=APPEAL_SHOW_DECIDED_HOURS) |
| 182 | + out = [] |
| 183 | + for a in reversed(self._appeals()): |
| 184 | + if a.get("child_id") != child_id: |
| 185 | + continue |
| 186 | + if a.get("status") != "open": |
| 187 | + decided = _when(a.get("decided_at")) |
| 188 | + if decided is None or decided < cutoff: |
| 189 | + continue |
| 190 | + row = {k: a[k] for k in ("id", "kind", "item_name", "points", "status") if k in a} |
| 191 | + if a.get("note"): |
| 192 | + row["note"] = a["note"] |
| 193 | + out.append(row) |
| 194 | + return out[:5] |
| 195 | + |
| 196 | + async def async_decide_appeal(self, appeal_id: str, overturn: bool, note: str = "") -> dict[str, Any]: |
| 197 | + """A parent upholds the original decision or overturns it.""" |
| 198 | + appeal = next((a for a in self._appeals() if a.get("id") == appeal_id), None) |
| 199 | + if appeal is None: |
| 200 | + raise ValueError("That appeal no longer exists.") |
| 201 | + if appeal.get("status") != "open": |
| 202 | + raise ValueError("That appeal has already been decided.") |
| 203 | + child = self.get_child(appeal["child_id"]) |
| 204 | + if child is None: |
| 205 | + raise ValueError("The child is no longer here.") |
| 206 | + if overturn: |
| 207 | + await self._async_overturn(appeal, child) |
| 208 | + now = dt_util.now() |
| 209 | + appeal["status"] = "overturned" if overturn else "upheld" |
| 210 | + appeal["decided_at"] = now.isoformat() |
| 211 | + appeal["note"] = clean_label(note, APPEAL_NOTE_MAX) |
| 212 | + await self.storage.async_save() |
| 213 | + payload = { |
| 214 | + "appeal_id": appeal["id"], |
| 215 | + "child_id": child.id, |
| 216 | + "child_name": child.name, |
| 217 | + "kind": appeal["kind"], |
| 218 | + "item_name": appeal["item_name"], |
| 219 | + "points": appeal["points"], |
| 220 | + "overturned": bool(overturn), |
| 221 | + "note": appeal["note"], |
| 222 | + } |
| 223 | + self.hass.bus.async_fire("taskmate_appeal_decided", {**payload, "timestamp": now.isoformat()}) |
| 224 | + if getattr(self, "notifications", None): |
| 225 | + await self.notifications.fire( |
| 226 | + NOTIF_TYPE_APPEAL_DECIDED, |
| 227 | + { |
| 228 | + **payload, |
| 229 | + "outcome": notify_strings.text( |
| 230 | + self.hass, "appeal_outcome_overturned" if overturn else "appeal_outcome_upheld" |
| 231 | + ), |
| 232 | + "note_text": f": {appeal['note']}" if appeal["note"] else "", |
| 233 | + }, |
| 234 | + only_recipients={f"child:{child.id}"}, |
| 235 | + ) |
| 236 | + await self.async_refresh() |
| 237 | + return dict(appeal) |
| 238 | + |
| 239 | + async def _async_overturn(self, appeal: dict[str, Any], child) -> None: |
| 240 | + kind = appeal["kind"] |
| 241 | + if kind == "chore": |
| 242 | + points = int(appeal.get("points", 0) or 0) |
| 243 | + if points > 0: |
| 244 | + await self.async_add_points(child.id, points, reason=f"Appeal: {appeal['item_name']}") |
| 245 | + return |
| 246 | + if kind == "reward": |
| 247 | + reward = self.get_reward(appeal.get("item_id", "")) |
| 248 | + if reward is None: |
| 249 | + raise ValueError("That reward no longer exists.") |
| 250 | + claim = RewardClaim(reward_id=reward.id, child_id=child.id, claimed_at=dt_util.now()) |
| 251 | + self.storage.add_reward_claim(claim) |
| 252 | + try: |
| 253 | + await self.async_approve_reward(claim.id) |
| 254 | + except ValueError: |
| 255 | + self.storage.remove_reward_claim(claim.id) |
| 256 | + raise |
| 257 | + return |
| 258 | + if kind == "penalty": |
| 259 | + txn = next((t for t in self.storage.get_points_transactions() if t.id == appeal["target_id"]), None) |
| 260 | + if txn is None: |
| 261 | + raise ValueError("That penalty has already been taken back.") |
| 262 | + offer = self._penalty_offer(child.id, appeal["item_name"], txn.created_at) |
| 263 | + if offer is not None and offer.get("status") == "redeemed": |
| 264 | + raise ValueError("That penalty was already earned back.") |
| 265 | + if offer is not None and offer.get("status") == "open": |
| 266 | + offer["status"] = "cancelled" |
| 267 | + await self.async_undo_transaction(txn.id) |
| 268 | + return |
| 269 | + raise ValueError(f"Unknown appeal kind {kind}") |
| 270 | + |
| 271 | + |
| 272 | +def _trim(appeals: list[dict[str, Any]]) -> None: |
| 273 | + """Keep every open appeal and the newest decided ones.""" |
| 274 | + decided = [a for a in appeals if a.get("status") != "open"] |
| 275 | + extra = len(decided) - APPEAL_KEEP |
| 276 | + if extra > 0: |
| 277 | + drop = {id(a) for a in decided[:extra]} |
| 278 | + appeals[:] = [a for a in appeals if id(a) not in drop] |
0 commit comments