import asyncio import re from dataclasses import dataclass, field from datetime import datetime, timedelta from pathlib import Path from typing import Any, Dict, List, Optional, Tuple, Type, Union from uuid import UUID import lxml # nosec import nio from ..html_markdown import HTML_PROCESSOR from ..utils import AutoStrEnum, auto from .model_item import ModelItem OptionalExceptionType = Union[Type[None], Type[Exception]] @dataclass class Account(ModelItem): user_id: str = field() display_name: str = "" avatar_url: str = "" first_sync_done: bool = False profile_updated: Optional[datetime] = None def __lt__(self, other: "Account") -> bool: name = self.display_name or self.user_id[1:] other_name = other.display_name or other.user_id[1:] return name < other_name @property def filter_string(self) -> str: return self.display_name @dataclass class Room(ModelItem): room_id: str = field() given_name: str = "" display_name: str = "" avatar_url: str = "" plain_topic: str = "" topic: str = "" inviter_id: str = "" inviter_name: str = "" inviter_avatar: str = "" left: bool = False typing_members: List[str] = field(default_factory=list) encrypted: bool = False invite_required: bool = True guests_allowed: bool = True can_invite: bool = False can_send_messages: bool = False can_set_name: bool = False can_set_topic: bool = False can_set_avatar: bool = False can_set_encryption: bool = False can_set_join_rules: bool = False can_set_guest_access: bool = False # Event.serialized last_event: Optional[Dict[str, Any]] = field(default=None, repr=False) def __lt__(self, other: "Room") -> bool: # Order: Invited rooms > joined rooms > left rooms. # Within these categories, sort by date then by name. # Left rooms may still have an inviter_id, so check left first. return ( self.left, other.inviter_id, other.last_event["date"] if other.last_event else datetime.fromtimestamp(0), self.display_name.lower() or self.room_id, ) < ( other.left, self.inviter_id, self.last_event["date"] if self.last_event else datetime.fromtimestamp(0), other.display_name.lower() or other.room_id, ) @property def filter_string(self) -> str: return " ".join(( self.display_name, self.topic, re.sub(r"<.*?>", "", self.last_event["inline_content"]) if self.last_event else "", )) @dataclass class Member(ModelItem): user_id: str = field() display_name: str = "" avatar_url: str = "" typing: bool = False power_level: int = 0 invited: bool = False def __lt__(self, other: "Member") -> bool: # Sort by name, but have members with higher power-level first and # invited-but-not-joined members last name = (self.display_name or self.user_id[1:]).lower() other_name = (other.display_name or other.user_id[1:]).lower() return ( self.invited, other.power_level, name, ) < ( other.invited, self.power_level, other_name, ) @property def filter_string(self) -> str: return self.display_name class UploadStatus(AutoStrEnum): Uploading = auto() Caching = auto() Error = auto() @dataclass class Upload(ModelItem): uuid: UUID = field() task: asyncio.Task = field() monitor: nio.TransferMonitor = field() filepath: Path = field() total_size: int = 0 uploaded: int = 0 speed: float = 0 time_left: Optional[timedelta] = None status: UploadStatus = UploadStatus.Uploading error: OptionalExceptionType = type(None) error_args: Tuple[Any, ...] = () start_date: datetime = field(init=False, default_factory=datetime.now) def __lt__(self, other: "Upload") -> bool: # Sort from newest upload to oldest. return self.start_date > other.start_date class TypeSpecifier(AutoStrEnum): none = auto() profile_change = auto() membership_change = auto() @dataclass class Event(ModelItem): source: Optional[nio.Event] = field() client_id: str = field() event_id: str = field() date: datetime = field() sender_id: str = field() sender_name: str = field() sender_avatar: str = field() content: str = "" inline_content: str = "" type_specifier: TypeSpecifier = TypeSpecifier.none target_id: str = "" target_name: str = "" target_avatar: str = "" is_local_echo: bool = False local_event_type: Optional[Type[nio.Event]] = None media_url: str = "" media_title: str = "" media_width: int = 0 media_height: int = 0 media_duration: int = 0 media_size: int = 0 media_mime: str = "" media_crypt_dict: Dict[str, Any] = field(default_factory=dict) thumbnail_url: str = "" thumbnail_width: int = 0 thumbnail_height: int = 0 thumbnail_crypt_dict: Dict[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: if not self.inline_content: self.inline_content = HTML_PROCESSOR.filter_inline(self.content) def __lt__(self, other: "Event") -> bool: # Sort events from newest to oldest. return True means return False. return self.date > other.date @property def event_type(self) -> Type: if self.local_event_type: return self.local_event_type return type(self.source) @property def links(self) -> List[str]: urls: List[str] = [] if self.content.strip(): urls += [link[2] for link in lxml.html.iterlinks(self.content)] if self.media_url: urls.append(self.media_url) return urls @dataclass class Device(ModelItem): device_id: str = field() ed25519_key: str = field() trusted: bool = False blacklisted: bool = False display_name: str = "" last_seen_ip: str = "" last_seen_date: str = ""