Spaces:
Sleeping
Sleeping
File size: 4,535 Bytes
2415446 05e7f80 2415446 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 | """Persist platform messages belonging to FCC-managed conversations."""
from datetime import UTC, datetime
from typing import Any
class ManagedMessageLog:
"""Track managed inbound and outbound messages in insertion order."""
def __init__(self, *, cap: int | None = None) -> None:
self._items: dict[str, list[dict[str, Any]]] = {}
self._ids: dict[str, set[str]] = {}
self._cap = cap
@property
def cap(self) -> int | None:
return self._cap
@classmethod
def from_json(cls, raw_log: Any, *, cap: int | None = None) -> "ManagedMessageLog":
"""Load current and legacy message-log entries."""
log = cls(cap=cap)
if not isinstance(raw_log, dict):
return log
for chat_key, items in raw_log.items():
if not isinstance(chat_key, str) or not isinstance(items, list):
continue
for item in items:
if not isinstance(item, dict):
continue
message_id = item.get("message_id")
direction = str(item.get("direction") or "")
kind = str(item.get("kind") or "")
if message_id is None or direction not in {"in", "out"} or not kind:
continue
log._append(
chat_key,
str(message_id),
ts=str(item.get("ts") or ""),
direction=direction,
kind=kind,
)
return log
def to_json(self) -> dict[str, list[dict[str, Any]]]:
return {chat_key: list(items) for chat_key, items in self._items.items()}
def record(
self,
*,
platform: str,
chat_id: str,
message_id: str,
direction: str,
kind: str,
) -> bool:
"""Record one managed platform message."""
if direction not in {"in", "out"}:
raise ValueError("Managed message direction must be 'in' or 'out'")
if not kind:
raise ValueError("Managed message kind cannot be empty")
return self._append(
make_chat_key(platform, chat_id),
str(message_id),
ts=datetime.now(UTC).isoformat(),
direction=direction,
kind=str(kind),
)
def ids_for_chat(self, platform: str, chat_id: str) -> list[str]:
chat_key = make_chat_key(platform, chat_id)
return [str(item["message_id"]) for item in self._items.get(chat_key, [])]
def remove_ids(self, platform: str, chat_id: str, message_ids: set[str]) -> bool:
chat_key = make_chat_key(platform, chat_id)
if not message_ids or chat_key not in self._items:
return False
before_count = len(self._items[chat_key])
retained = [
item
for item in self._items[chat_key]
if str(item["message_id"]) not in message_ids
]
if retained:
self._items[chat_key] = retained
self._ids[chat_key] = {str(item["message_id"]) for item in retained}
else:
self._items.pop(chat_key)
self._ids.pop(chat_key, None)
return len(retained) != before_count
def clear_chat(self, platform: str, chat_id: str) -> bool:
chat_key = make_chat_key(platform, chat_id)
removed = self._items.pop(chat_key, None) is not None
self._ids.pop(chat_key, None)
return removed
def _append(
self,
chat_key: str,
message_id: str,
*,
ts: str,
direction: str,
kind: str,
) -> bool:
seen = self._ids.setdefault(chat_key, set())
if message_id in seen:
return False
self._items.setdefault(chat_key, []).append(
{
"message_id": message_id,
"ts": ts,
"direction": direction,
"kind": kind,
}
)
seen.add(message_id)
self._trim(chat_key)
return True
def _trim(self, chat_key: str) -> None:
if self._cap is None or self._cap <= 0:
return
items = self._items.get(chat_key, [])
if len(items) <= self._cap:
return
retained = items[-self._cap :]
self._items[chat_key] = retained
self._ids[chat_key] = {str(item["message_id"]) for item in retained}
def make_chat_key(platform: str, chat_id: str) -> str:
return f"{platform}:{chat_id}"
|