# Copyright Mirage authors & contributors # SPDX-License-Identifier: LGPL-3.0-or-later """`ModelItem` subclasses definitions.""" import json from dataclasses import asdict, dataclass, field from datetime import datetime, timedelta from pathlib import Path from typing import Any, Dict, List, Optional, Set, Tuple, Type, Union from uuid import UUID import lxml # nosec import nio from ..presence import Presence from ..utils import AutoStrEnum, auto, strip_html_tags from .model_item import ModelItem OptionalExceptionType = Union[Type[None], Type[Exception]] ZERO_DATE = datetime.fromtimestamp(0) class TypeSpecifier(AutoStrEnum): """Enum providing clarification of purpose for some matrix events.""" Unset = auto() ProfileChange = auto() MembershipChange = auto() class PingStatus(AutoStrEnum): """Enum for the status of a homeserver ping operation.""" Done = auto() Pinging = auto() Failed = auto() class RoomNotificationOverride(AutoStrEnum): """Possible per-room notification override settings, as displayed in the left sidepane's context menu when right-clicking a room. """ UseDefaultSettings = auto() AllEvents = auto() HighlightsOnly = auto() IgnoreEvents = auto() @dataclass(eq=False) class Homeserver(ModelItem): """A homeserver we can connect to. The `id` field is the server's URL.""" id: str = field() name: str = field() site_url: str = field() country: str = field() ping: int = -1 status: PingStatus = PingStatus.Pinging stability: float = -1 downtimes_ms: List[float] = field(default_factory=list) def __lt__(self, other: "Homeserver") -> bool: return (self.name.lower(), self.id) < (other.name.lower(), other.id) @dataclass(eq=False) class Account(ModelItem): """A logged in matrix account.""" id: str = field() order: int = -1 display_name: str = "" avatar_url: str = "" max_upload_size: int = 0 profile_updated: datetime = ZERO_DATE connecting: bool = False total_unread: int = 0 total_highlights: int = 0 local_unreads: bool = False ignored_users: Set[str] = field(default_factory=set) # For some reason, Account cannot inherit Presence, because QML keeps # complaining type error on unknown file presence_support: bool = False save_presence: bool = True presence: Presence.State = Presence.State.offline currently_active: bool = False last_active_at: datetime = ZERO_DATE status_msg: str = "" def __lt__(self, other: "Account") -> bool: return (self.order, self.id) < (other.order, other.id) @dataclass(eq=False) class PushRule(ModelItem): """A push rule configured for one of our account.""" id: Tuple[str, str] = field() # (kind.value, rule_id) kind: nio.PushRuleKind = field() rule_id: str = field() order: int = field() default: bool = field() enabled: bool = True conditions: List[Dict[str, Any]] = field(default_factory=list) pattern: str = "" actions: List[Dict[str, Any]] = field(default_factory=list) notify: bool = False highlight: bool = False bubble: bool = False sound: str = "" # usually "default" when set urgency_hint: bool = False def __lt__(self, other: "PushRule") -> bool: return ( self.kind is nio.PushRuleKind.underride, self.kind is nio.PushRuleKind.sender, self.kind is nio.PushRuleKind.room, self.kind is nio.PushRuleKind.content, self.kind is nio.PushRuleKind.override, self.order, self.id, ) < ( other.kind is nio.PushRuleKind.underride, other.kind is nio.PushRuleKind.sender, other.kind is nio.PushRuleKind.room, other.kind is nio.PushRuleKind.content, other.kind is nio.PushRuleKind.override, other.order, other.id, ) @dataclass class Room(ModelItem): """A matrix room we are invited to, are or were member of.""" id: str = field() for_account: str = "" given_name: str = "" display_name: str = "" main_alias: 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) federated: bool = True encrypted: bool = False unverified_devices: bool = False invite_required: bool = True guests_allowed: bool = True default_power_level: int = 0 own_power_level: int = 0 can_invite: bool = False can_kick: bool = False can_redact_all: 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 can_set_power_levels: bool = False last_event_date: datetime = ZERO_DATE unreads: int = 0 highlights: int = 0 local_unreads: bool = False notification_setting: RoomNotificationOverride = \ RoomNotificationOverride.UseDefaultSettings lexical_sorting: bool = False pinned: bool = False # Allowed keys: "last_event_date", "unreads", "highlights", "local_unreads" # Keys in this dict will override their corresponding item fields for the # __lt__ method. This is used when we want to lock a room's position, # e.g. to avoid having the room move around when it is focused in the GUI _sort_overrides: Dict[str, Any] = field(default_factory=dict) def _sorting(self, key: str) -> Any: return self._sort_overrides.get(key, getattr(self, key)) def __lt__(self, other: "Room") -> bool: by_activity = not self.lexical_sorting return ( self.for_account, other.pinned, self.left, # Left rooms may have an inviter_id, check them first bool(other.inviter_id), bool(by_activity and other._sorting("highlights")), bool(by_activity and other._sorting("unreads")), bool(by_activity and other._sorting("local_unreads")), other._sorting("last_event_date") if by_activity else ZERO_DATE, (self.display_name or self.id).lower(), self.id, ) < ( other.for_account, self.pinned, other.left, bool(self.inviter_id), bool(by_activity and self._sorting("highlights")), bool(by_activity and self._sorting("unreads")), bool(by_activity and self._sorting("local_unreads")), self._sorting("last_event_date") if by_activity else ZERO_DATE, (other.display_name or other.id).lower(), other.id, ) @dataclass(eq=False) class AccountOrRoom(Account, Room): """The left sidepane in the GUI lists a mixture of accounts and rooms giving a tree view illusion. Since all items in a QML ListView must have the same available properties, this class inherits both `Account` and `Room` to fulfill that purpose. """ type: Union[Type[Account], Type[Room]] = Account account_order: int = -1 def __lt__(self, other: "AccountOrRoom") -> bool: # type: ignore by_activity = not self.lexical_sorting return ( self.account_order, self.id if self.type is Account else self.for_account, other.type is Account, other.pinned, self.left, bool(other.inviter_id), bool(by_activity and other._sorting("highlights")), bool(by_activity and other._sorting("unreads")), bool(by_activity and other._sorting("local_unreads")), other._sorting("last_event_date") if by_activity else ZERO_DATE, (self.display_name or self.id).lower(), self.id, ) < ( other.account_order, other.id if other.type is Account else other.for_account, self.type is Account, self.pinned, other.left, bool(self.inviter_id), bool(by_activity and self._sorting("highlights")), bool(by_activity and self._sorting("unreads")), bool(by_activity and self._sorting("local_unreads")), self._sorting("last_event_date") if by_activity else ZERO_DATE, (other.display_name or other.id).lower(), other.id, ) @dataclass(eq=False) class Member(ModelItem): """A member in a matrix room.""" id: str = field() display_name: str = "" avatar_url: str = "" typing: bool = False power_level: int = 0 invited: bool = False ignored: bool = False profile_updated: datetime = ZERO_DATE last_read_event: str = "" presence: Presence.State = Presence.State.offline currently_active: bool = False last_active_at: datetime = ZERO_DATE status_msg: str = "" def __lt__(self, other: "Member") -> bool: return ( self.invited, other.power_level, self.ignored, Presence.State.offline if self.ignored else self.presence, (self.display_name or self.id[1:]).lower(), self.id, ) < ( other.invited, self.power_level, other.ignored, Presence.State.offline if other.ignored else other.presence, (other.display_name or other.id[1:]).lower(), other.id, ) class TransferStatus(AutoStrEnum): """Enum describing the status of an upload operation.""" Preparing = auto() Transfering = auto() Caching = auto() Error = auto() @dataclass(eq=False) class Transfer(ModelItem): """Represent a running or failed file upload/download operation.""" id: UUID = field() is_upload: bool = field() filepath: Path = Path("-") total_size: int = 0 transferred: int = 0 speed: float = 0 time_left: timedelta = timedelta(0) paused: bool = False status: TransferStatus = TransferStatus.Preparing error: OptionalExceptionType = type(None) error_args: Tuple[Any, ...] = () start_date: datetime = field(init=False, default_factory=datetime.now) def __lt__(self, other: "Transfer") -> bool: return (self.start_date, self.id) > (other.start_date, other.id) @dataclass(eq=False) class Event(ModelItem): """A matrix state event or message.""" id: str = field() event_id: str = field() event_type: Type[nio.Event] = field() date: datetime = field() sender_id: str = field() sender_name: str = field() sender_avatar: str = field() fetch_profile: bool = False content: str = "" inline_content: str = "" reason: str = "" links: List[str] = field(default_factory=list) mentions: List[Tuple[str, str]] = field(default_factory=list) type_specifier: TypeSpecifier = TypeSpecifier.Unset target_id: str = "" target_name: str = "" target_avatar: str = "" redacter_id: str = "" redacter_name: str = "" # {user_id: server_timestamp} - QML can't parse dates from JSONified dicts last_read_by: Dict[str, int] = field(default_factory=dict) read_by_count: int = 0 is_local_echo: bool = False source: Optional[nio.Event] = None media_url: str = "" media_http_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) media_local_path: Union[str, Path] = "" thumbnail_url: str = "" thumbnail_mime: str = "" thumbnail_width: int = 0 thumbnail_height: int = 0 thumbnail_crypt_dict: Dict[str, Any] = field(default_factory=dict) def __lt__(self, other: "Event") -> bool: return (self.date, self.id) > (other.date, other.id) @property def plain_content(self) -> str: """Plaintext version of the event's content.""" if isinstance(self.source, nio.RoomMessageText): return self.source.body return strip_html_tags(self.content) @staticmethod def parse_links(text: str) -> List[str]: """Return list of URLs (`` tags) present in the content.""" ignore = [] if "" in text or "mention" in text: parser = lxml.html.etree.HTMLParser() tree = lxml.etree.fromstring(text, parser) ignore = [ lxml.etree.tostring(matching_element) for ugly_disgusting_xpath in [ # Match mx-reply > blockquote > second a (user ID link) "//mx-reply/blockquote/a[count(preceding-sibling::*)<=1]", # Match tags with a mention class '//a[contains(concat(" ",normalize-space(@class)," ")' '," mention ")]', ] for matching_element in tree.xpath(ugly_disgusting_xpath) ] if not text.strip(): return [] return [ url for el, attrib, url, pos in lxml.html.iterlinks(text) if lxml.etree.tostring(el) not in ignore ] def serialized_field(self, field: str) -> Any: if field == "source": source_dict = asdict(self.source) if self.source else {} return json.dumps(source_dict) return super().serialized_field(field)