Spaces:
Sleeping
Sleeping
File size: 2,165 Bytes
1f213fe | 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 | import copy
from typing import Any, Dict, List
class ChatApp:
def __init__(self, channels: List[str], noise_channels: List[str]):
all_channels = list(dict.fromkeys(channels + noise_channels))
self.channels = all_channels
self.messages: Dict[str, List[Dict]] = {ch: [] for ch in all_channels}
self._episode_messages: List[Dict] = []
def snapshot(self) -> Dict[str, Any]:
return {
"channels": self.channels,
"messages_posted_this_episode": copy.deepcopy(self._episode_messages),
}
def post_message(self, args: Dict[str, Any]) -> Dict[str, Any]:
channel = args.get("channel", "")
text = args.get("text", "")
if not channel or not text:
return {"ok": False, "error": "Both 'channel' and 'text' are required"}
if channel not in self.channels:
return {"ok": False, "error": f"Channel '{channel}' not found. Use chat.list_channels to see available channels."}
msg = {"channel": channel, "text": text}
self.messages[channel].append(msg)
self._episode_messages.append(msg)
return {"ok": True, "data": {"channel": channel, "delivered": True}, "side_effects": [f"message posted to {channel}"]}
def read_channel(self, args: Dict[str, Any]) -> Dict[str, Any]:
channel = args.get("channel", "")
limit = min(args.get("limit", 20), 50)
if channel not in self.channels:
return {"ok": False, "error": f"Channel '{channel}' not found"}
msgs = self.messages[channel][-limit:]
return {"ok": True, "data": {"channel": channel, "messages": copy.deepcopy(msgs)}}
def list_channels(self, args: Dict[str, Any]) -> Dict[str, Any]:
return {"ok": True, "data": self.channels}
def search(self, args: Dict[str, Any]) -> Dict[str, Any]:
query = args.get("query", "").lower()
results = []
for ch, msgs in self.messages.items():
for m in msgs:
if query in m["text"].lower():
results.append({"channel": ch, "text": m["text"]})
return {"ok": True, "data": results[:20]}
|