105 lines
3.1 KiB
Python
105 lines
3.1 KiB
Python
import re
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
|
|
from utils.logging import logger
|
|
from utils.read.models import WatchView
|
|
from utils.watch_rules import (
|
|
MAX_TEXT,
|
|
DeletedParams,
|
|
EditedParams,
|
|
KeywordParams,
|
|
NewMessageParams,
|
|
PeerOnlineParams,
|
|
WatchParamsError,
|
|
compile_pattern,
|
|
parse_params,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class Seen:
|
|
chat_id: int
|
|
message_id: int
|
|
sender_id: int | None
|
|
text: str | None
|
|
outgoing: bool = False
|
|
date: datetime | None = None
|
|
edit_date: datetime | None = None
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class KeywordRule:
|
|
watch_id: int
|
|
params: KeywordParams
|
|
regex: re.Pattern[str] | None
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class Rules:
|
|
keyword: list[KeywordRule] = field(default_factory=list)
|
|
new_message: list[tuple[int, NewMessageParams]] = field(default_factory=list)
|
|
deleted: list[tuple[int, DeletedParams]] = field(default_factory=list)
|
|
edited: list[tuple[int, EditedParams]] = field(default_factory=list)
|
|
online: list[tuple[int, PeerOnlineParams]] = field(default_factory=list)
|
|
|
|
def __len__(self) -> int:
|
|
return (
|
|
len(self.keyword)
|
|
+ len(self.new_message)
|
|
+ len(self.deleted)
|
|
+ len(self.edited)
|
|
+ len(self.online)
|
|
)
|
|
|
|
|
|
def build_rules(watches: list[WatchView]) -> Rules:
|
|
rules = Rules()
|
|
for watch in watches:
|
|
if not watch.enabled:
|
|
continue
|
|
try:
|
|
params = parse_params(watch.kind, watch.params)
|
|
except WatchParamsError as exc:
|
|
logger.warning(f"[yellow]Watch {watch.id} skipped:[/] {exc}")
|
|
continue
|
|
match params:
|
|
case KeywordParams():
|
|
regex = compile_pattern(params.pattern) if params.regex else None
|
|
rules.keyword.append(KeywordRule(watch.id, params, regex))
|
|
case NewMessageParams():
|
|
rules.new_message.append((watch.id, params))
|
|
case DeletedParams():
|
|
rules.deleted.append((watch.id, params))
|
|
case EditedParams():
|
|
rules.edited.append((watch.id, params))
|
|
case PeerOnlineParams():
|
|
rules.online.append((watch.id, params))
|
|
return rules
|
|
|
|
|
|
def _in_scope(chat_id: int | None, sender_id: int | None, seen: Seen) -> bool:
|
|
if chat_id is not None and chat_id != seen.chat_id:
|
|
return False
|
|
return sender_id is None or sender_id == seen.sender_id
|
|
|
|
|
|
def match_keyword(rule: KeywordRule, seen: Seen) -> bool:
|
|
params = rule.params
|
|
if not seen.text or not _in_scope(params.chat_id, params.sender_id, seen):
|
|
return False
|
|
sample = seen.text[:MAX_TEXT]
|
|
if rule.regex is not None:
|
|
return rule.regex.search(sample) is not None
|
|
return params.pattern.casefold() in sample.casefold()
|
|
|
|
|
|
def match_new_message(params: NewMessageParams, seen: Seen) -> bool:
|
|
return _in_scope(params.chat_id, params.sender_id, seen)
|
|
|
|
|
|
def match_chat_scope(params: DeletedParams | EditedParams, seen: Seen) -> bool:
|
|
if params.chat_id is None:
|
|
return seen.chat_id > 0
|
|
return params.chat_id == seen.chat_id
|