import re from collections.abc import Iterable, Iterator from importlib import import_module from typing import Any, Literal, Self, get_args from pydantic import BaseModel, Field, ValidationError, model_validator MAX_PATTERN = 200 MAX_TEXT = 4096 MAX_NAME = 120 WatchKind = Literal[ "keyword", "new_message", "message_deleted", "message_edited", "peer_online" ] WATCH_KINDS: tuple[str, ...] = get_args(WatchKind) sre_parser = import_module("re._parser") _REPEATS = {sre_parser.MAX_REPEAT, sre_parser.MIN_REPEAT, sre_parser.POSSESSIVE_REPEAT} class WatchParamsError(ValueError): pass def _subpatterns(value: object) -> Iterator[Any]: if isinstance(value, sre_parser.SubPattern): yield value elif isinstance(value, tuple | list): for item in value: yield from _subpatterns(item) def _repeat_depth(items: Iterable[tuple[Any, Any]]) -> int: depth = 0 for op, value in items: if op in _REPEATS: low, high, body = value depth = max(depth, _repeat_depth(body) + (high > max(low, 1))) else: inner = (_repeat_depth(sub) for sub in _subpatterns(value)) depth = max(depth, *inner, 0) return depth def compile_pattern(pattern: str) -> re.Pattern[str]: try: compiled = re.compile(pattern, re.IGNORECASE) except re.error as exc: msg = f"invalid regex: {exc}" raise ValueError(msg) from exc if _repeat_depth(sre_parser.parse(pattern)) > 1: msg = "regex repeats inside a repeat, like (a+)+, which can hang" raise ValueError(msg) return compiled class KeywordParams(BaseModel): pattern: str = Field(min_length=1, max_length=MAX_PATTERN) regex: bool = False chat_id: int | None = None sender_id: int | None = None @model_validator(mode="before") @classmethod def _legacy_keyword(cls, data: object) -> object: if isinstance(data, dict) and "pattern" not in data and "keyword" in data: return {**data, "pattern": data["keyword"]} return data @model_validator(mode="after") def _valid_pattern(self) -> Self: if not self.pattern.strip(): msg = "pattern is blank" raise ValueError(msg) if self.regex: compile_pattern(self.pattern) return self class NewMessageParams(BaseModel): chat_id: int | None = None sender_id: int | None = None @model_validator(mode="after") def _scoped(self) -> Self: if self.chat_id is None and self.sender_id is None: msg = "set chat_id, sender_id or both" raise ValueError(msg) return self class DeletedParams(BaseModel): chat_id: int | None = None class EditedParams(BaseModel): chat_id: int | None = None class PeerOnlineParams(BaseModel): peer_id: int WatchParams = ( KeywordParams | NewMessageParams | DeletedParams | EditedParams | PeerOnlineParams ) PARAMS_BY_KIND: dict[str, type[WatchParams]] = { "keyword": KeywordParams, "new_message": NewMessageParams, "message_deleted": DeletedParams, "message_edited": EditedParams, "peer_online": PeerOnlineParams, } def _describe(exc: ValidationError) -> str: parts = [] for error in exc.errors(): where = ".".join(str(part) for part in error["loc"]) message = error["msg"].removeprefix("Value error, ") parts.append(f"{where}: {message}" if where else message) return "; ".join(parts) def parse_params(kind: str, params: dict[str, Any]) -> WatchParams: model = PARAMS_BY_KIND.get(kind) if model is None: msg = f"unknown kind {kind!r}, expected one of {', '.join(WATCH_KINDS)}" raise WatchParamsError(msg) try: return model.model_validate(params) except ValidationError as exc: raise WatchParamsError(_describe(exc)) from exc def normalize_params(kind: str, params: dict[str, Any]) -> dict[str, Any]: return parse_params(kind, params).model_dump(exclude_none=True) def clean_name(name: str | None) -> str | None: if name is None: return None name = name.strip()[:MAX_NAME] return name or None