146 lines
4.1 KiB
Python
146 lines
4.1 KiB
Python
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
|