from dataclasses import dataclass, field from datetime import datetime from typing import Dict, Optional from paircd.client import Client from paircd.channel import Channel from paircd.reply import QUIT @dataclass class Server: clients_by_nick: Dict[str, Client] = field(default_factory=dict) channels_by_name: Dict[str, Channel] = field(default_factory=dict) create_time: datetime = field(default_factory=datetime.utcnow) def add_client(self, client: Client) -> None: self.clients_by_nick[client.nickname] = client def get_channel_by_name(self, name: str, create: bool = True) -> Optional[Channel]: if name not in self.channels_by_name and create: self.channels_by_name[name] = Channel(name=name) return self.channels_by_name.get(name) def get_client_by_name(self, name: str) -> Optional[Client]: return self.clients_by_nick.get(name) def remove_client_by_name(self, name: str) -> None: del self.clients_by_nick[name] def rename_client(self, name: str, new_name: str) -> None: if name not in self.clients_by_nick: raise KeyError(f"No client on server with nick {name}") if new_name in self.clients_by_nick: raise KeyError(f"Client already exists with nick {new_name}") self.clients_by_nick[new_name] = self.clients_by_nick[name] del self.clients_by_nick[name] async def disconnect_client(self, nick: str, quit_msg: str) -> None: client = self.clients_by_nick[nick] if client.closed: return client.log(f"is quitting: {quit_msg}") msg = QUIT(quit_msg, prefix=client.id()) for client in self.clients_by_nick.values(): if client.nickname == nick: continue client.write_message(msg) for channel_name in client.channels: channel = self.get_channel_by_name(channel_name, create=False) if channel is None: continue channel.remove_client_by_nick(client.nickname) if channel.get_channel_size() == 0: del self.channels_by_name[channel_name] self.remove_client_by_name(client.nickname) await client.close()