from datetime import UTC, datetime, timedelta from typing import Literal from pydantic import BaseModel, field_validator DEFAULT_LIMIT = 50 MAX_LIMIT = 500 _EPOCH = datetime(1970, 1, 1, tzinfo=UTC) _MICROSECOND = timedelta(microseconds=1) ChatKind = Literal["private", "group", "channel"] HitSource = Literal["text", "stt", "version", "data"] FoundWhere = Literal[ "text", "button", "link", "forward", "reply", "file", "preview", "data" ] def encode_cursor(date: datetime, chat_id: int, message_id: int) -> str: return f"{(date - _EPOCH) // _MICROSECOND}_{chat_id}_{message_id}" def decode_cursor(cursor: str) -> tuple[datetime, int, int]: micros, chat_id, message_id = (int(part) for part in cursor.split("_")) return _EPOCH + micros * _MICROSECOND, chat_id, message_id class SearchFilters(BaseModel): account_id: int query: str | None = None chat_id: int | None = None sender_id: int | None = None kind: ChatKind | None = None has_media: bool | None = None deleted: bool | None = None edited: bool | None = None date_from: datetime | None = None date_to: datetime | None = None regex: str | None = None everywhere: bool = False cursor: str | None = None limit: int = DEFAULT_LIMIT @field_validator("cursor") @classmethod def _valid_cursor(cls, value: str | None) -> str | None: if value is not None: decode_cursor(value) return value class Found(BaseModel): where: FoundWhere path: str value: str button: str | None = None class SearchHit(BaseModel): chat_id: int message_id: int date: datetime sender_id: int | None text: str | None has_media: bool media_kind: str | None edited_at: datetime | None deleted_at: datetime | None source: HitSource extracted_text: str | None = None version_text: str | None = None version_at: datetime | None = None found: Found | None = None class SearchPage(BaseModel): hits: list[SearchHit] next_cursor: str | None terms: list[str] scanned_to: datetime | None = None