Spaces:
Running
Running
| """Error reports for the log channel. | |
| Every ERROR-level log record becomes one message: what the bot was doing, then the full traceback, inside a Python code | |
| block (Telegram shows it with a "Python" label and a copy button). An identical error is reported once per couple of | |
| minutes, no more than ``PER_MINUTE`` reports go out in a minute, and secrets are masked before anything is sent. | |
| """ | |
| from __future__ import annotations | |
| import asyncio | |
| import contextvars | |
| import logging | |
| import os | |
| import re | |
| import time | |
| from collections import deque | |
| from dataclasses import dataclass | |
| from itertools import groupby | |
| from typing import Any, Awaitable, Callable, Deque, Dict, Iterable, List, Optional, Set, Tuple | |
| from urllib.parse import urlsplit | |
| from .utils import clip, esc | |
| log = logging.getLogger(__name__) | |
| Send = Callable[[str], Awaitable[None]] | |
| MAX_TEXT = 3800 # UTF-16 units of the code block: Telegram takes 4096 per message | |
| HEAD_LIMIT = 700 # characters of "what was happening" kept when a traceback shares the message | |
| REPEAT_WINDOW = 120.0 # seconds during which an identical error is reported only once | |
| PER_MINUTE = 10 # reports sent per minute at most | |
| # What the bot is working on right now: text, or a function that returns it. Errors logged anywhere below it mention | |
| # it, so a failure deep inside a helper still says which message or file it belongs to. | |
| error_context: "contextvars.ContextVar[Any]" = contextvars.ContextVar("error_context", default=None) | |
| def describe_update(event: Any) -> str: | |
| """One line about a Telethon event: who sent what, where. Never raises.""" | |
| try: | |
| kind = type(event).__qualname__.removesuffix(".Event") # NewMessage / CallbackQuery | |
| parts = [f"sender_id={getattr(event, 'sender_id', None)}", f"chat_id={getattr(event, 'chat_id', None)}"] | |
| data = getattr(event, "data", None) | |
| if data is not None: | |
| parts.append(f"message_id={getattr(event, 'message_id', None)}") | |
| parts.append("data=" + repr(bytes(data).decode("utf-8", "replace"))) | |
| elif hasattr(event, "builder"): # an inline query: what was typed after the bot's name | |
| parts.append("query=" + repr(clip(getattr(event, "text", "") or "", 100))) | |
| else: | |
| parts.append(f"message_id={getattr(event, 'id', None)}") | |
| text = getattr(event, "raw_text", None) | |
| if text: | |
| parts.append("text=" + repr(clip(text, 200))) | |
| file = getattr(getattr(event, "message", None), "file", None) | |
| if file is not None: | |
| parts.append(f"file={getattr(file, 'name', None)!r} ({getattr(file, 'mime_type', None)}, {getattr(file, 'size', None)} bytes)") | |
| return f"{kind}({', '.join(parts)})" | |
| except Exception: # noqa: BLE001 | |
| return type(event).__name__ | |
| _SQL_PARAMS = re.compile(r"\[parameters: [^\n]*\]") # SQLAlchemy prints the values it bound, e.g. a saved session | |
| _TOKENISH = re.compile(r"[A-Za-z0-9_\-+=]{100,}") # a session string is ~350 characters; no path or file name has 100 in a row | |
| def _mask_token(match: "re.Match[str]") -> str: | |
| text = match.group(0) | |
| return "***" if re.search(r"[A-Za-z0-9]", text) else text # a line of dashes is not a secret | |
| def redact(text: str, secrets: Iterable[str] = ()) -> str: | |
| """Hide the configured secrets and anything that looks like a session string.""" | |
| for secret in secrets: | |
| if len(secret) >= 6: | |
| text = text.replace(secret, "***") | |
| return _TOKENISH.sub(_mask_token, _SQL_PARAMS.sub("[parameters: …]", text)) | |
| def secrets_from(cfg: Any) -> List[str]: | |
| """The values of the configuration that must never appear in a report.""" | |
| values = [cfg.bot_token, cfg.api_hash, cfg.session_string, cfg.database_url] | |
| try: | |
| values.append(urlsplit(cfg.database_url).password or "") | |
| except ValueError: | |
| pass | |
| return [v for v in values if v] | |
| def _u16(text: str) -> int: | |
| return len(text.encode("utf-16-le")) // 2 | |
| def _tail(text: str, units: int) -> str: | |
| """The end of ``text`` that fits in ``units`` UTF-16 units, starting at the beginning of a line.""" | |
| used, start = 0, len(text) | |
| for i in range(len(text) - 1, -1, -1): | |
| used += 2 if ord(text[i]) > 0xFFFF else 1 | |
| if used > units: | |
| break | |
| start = i | |
| tail = text[start:] | |
| if start > 0: | |
| newline = tail.find("\n") | |
| tail = tail[newline + 1:] if newline != -1 else tail | |
| return tail | |
| def _head(text: str, units: int) -> str: | |
| """The beginning of ``text`` that fits in ``units`` UTF-16 units, ending at the end of a line.""" | |
| used, end = 0, len(text) | |
| for i, char in enumerate(text): | |
| used += 2 if ord(char) > 0xFFFF else 1 | |
| if used > units: | |
| end = i | |
| break | |
| if end == len(text): | |
| return text | |
| head = text[:end] | |
| newline = head.rfind("\n") | |
| return head[:newline] if newline != -1 else head | |
| _LIBRARY_FRAME = re.compile(r' File ".*[\\/](?:site-packages|dist-packages|lib[\\/]python[\d.]+)[\\/]') | |
| _CUT = "\n…\n" | |
| def _elide(trace: str, room: int) -> str: | |
| """Make a traceback fit ``room`` by replacing runs of frames from libraries (SQLAlchemy, Telethon, asyncio...) with | |
| one line each: the bot's own frames and the error stay. Runs of three or more go first, then every run.""" | |
| if _u16(trace) <= room: | |
| return trace | |
| blocks: List[List[str]] = [] # a frame is its "File" line and the source lines under it; any other line is a block of its own | |
| for line in trace.split("\n"): | |
| if blocks and line.startswith(" ") and blocks[-1][0].startswith(" File "): | |
| blocks[-1].append(line) | |
| else: | |
| blocks.append([line]) | |
| text = trace | |
| for shortest_run in (3, 1): | |
| lines: List[str] = [] | |
| for library, group in groupby(blocks, key=lambda block: bool(_LIBRARY_FRAME.match(block[0]))): | |
| frames = list(group) | |
| if library and len(frames) >= shortest_run: | |
| lines.append(f" [… {len(frames)} library frame{'s' if len(frames) != 1 else ''} …]") | |
| else: | |
| lines.extend(line for frame in frames for line in frame) | |
| text = "\n".join(lines) | |
| if _u16(text) <= room: | |
| break | |
| return text | |
| def _fit(head: str, trace: str) -> str: | |
| """``head`` + the traceback in at most ``MAX_TEXT``. What does not fit goes in this order: frames from libraries, | |
| then the middle of the traceback: its first frames (where it came from) and its end (the error) are what matter.""" | |
| if not trace: | |
| return clip(head, MAX_TEXT // 2) | |
| head = clip(head, HEAD_LIMIT) | |
| room = MAX_TEXT - _u16(head) - 2 | |
| trace = _elide(trace, room) | |
| if _u16(trace) > room: | |
| room -= len(_CUT) | |
| first = room * 2 // 5 | |
| trace = _head(trace, first) + _CUT + _tail(trace, room - first) | |
| return head + "\n\n" + trace | |
| def _signature(exc_info: Any) -> str: | |
| """Same exception type raised on the same line = the same error.""" | |
| etype, _value, tb = exc_info | |
| while tb is not None and tb.tb_next is not None: | |
| tb = tb.tb_next | |
| where = f"{os.path.basename(tb.tb_frame.f_code.co_filename)}:{tb.tb_lineno}" if tb is not None else "" | |
| return f"{getattr(etype, '__name__', etype)}@{where}" | |
| def build_report(record: logging.LogRecord, secrets: Iterable[str] = ()) -> Tuple[str, str]: | |
| """``(signature, html)``: the record's message, what the bot was working on, and the traceback in a Python block.""" | |
| try: | |
| head = record.getMessage() | |
| except Exception: # noqa: BLE001 | |
| head = str(record.msg) | |
| context = error_context.get() | |
| if callable(context): | |
| context = context() | |
| if context and str(context) not in head: | |
| head += "\n" + str(context) | |
| trace, signature = "", f"{record.name}|{head[:100]}" | |
| if record.exc_info and record.exc_info[0] is not None: | |
| trace = logging.Formatter().formatException(record.exc_info) | |
| signature = _signature(record.exc_info) | |
| text = _fit(redact(head, secrets), redact(trace, secrets)) | |
| return signature, f'<pre><code class="language-python">{esc(text)}</code></pre>' | |
| class _Seen: | |
| at: float | |
| repeats: int = 0 | |
| class ErrorReporter: | |
| """Sends the reports to the log channel one after another and keeps repeats and floods out of it.""" | |
| def __init__(self, send: Send, *, repeat_window: float = REPEAT_WINDOW, per_minute: int = PER_MINUTE, | |
| clock: Callable[[], float] = time.monotonic): | |
| self._send = send | |
| self._repeat_window = repeat_window | |
| self._per_minute = per_minute | |
| self._clock = clock | |
| self._seen: Dict[str, _Seen] = {} | |
| self._sent: Deque[float] = deque() # when the reports of the last minute went out | |
| self._skipped = 0 | |
| self._lock = asyncio.Lock() | |
| self._tasks: Set["asyncio.Future[None]"] = set() | |
| def submit(self, signature: str, body: str) -> None: | |
| """Queue one report (inside the event loop). A repeat or a flood is counted and mentioned in the next report.""" | |
| now = self._clock() | |
| seen = self._seen.get(signature) | |
| if seen is not None and now - seen.at < self._repeat_window: | |
| seen.repeats += 1 | |
| return | |
| while self._sent and now - self._sent[0] >= 60: | |
| self._sent.popleft() | |
| if len(self._sent) >= self._per_minute: | |
| self._skipped += 1 | |
| return | |
| notes = [] | |
| if seen is not None and seen.repeats: | |
| notes.append(f"🔁 {seen.repeats} more of this error since the last report.") | |
| if self._skipped: | |
| notes.append(f"⚠️ {self._skipped} other error report(s) were left out to keep this chat readable.") | |
| self._skipped = 0 | |
| self._seen[signature] = _Seen(now) | |
| if len(self._seen) > 200: | |
| self._seen = {k: v for k, v in self._seen.items() if now - v.at < self._repeat_window} | |
| self._sent.append(now) | |
| task = asyncio.ensure_future(self._deliver(body + "".join("\n" + note for note in notes))) | |
| self._tasks.add(task) | |
| task.add_done_callback(self._tasks.discard) | |
| async def _deliver(self, text: str) -> None: | |
| async with self._lock: # in the order they happened | |
| try: | |
| await self._send(text) | |
| except Exception as exc: # noqa: BLE001 | |
| log.warning("could not send an error report to the log channel: %s", exc) | |
| async def flush(self, timeout: float = 5.0) -> None: | |
| """Wait (a little) for the reports that are still on their way.""" | |
| if self._tasks: | |
| await asyncio.wait(set(self._tasks), timeout=timeout) | |
| class ChannelLogHandler(logging.Handler): | |
| """Turns every ERROR record, from any thread, into a report for the log channel.""" | |
| def __init__(self, reporter: ErrorReporter, loop: asyncio.AbstractEventLoop, secrets: Iterable[str] = ()): | |
| super().__init__(level=logging.ERROR) | |
| self._reporter = reporter | |
| self._loop = loop | |
| self._secrets = [s for s in secrets if s] | |
| self._logger: Optional[logging.Logger] = None | |
| def emit(self, record: logging.LogRecord) -> None: | |
| if record.name == __name__: # a problem with the reports themselves must not become a report | |
| return | |
| try: | |
| signature, body = build_report(record, self._secrets) | |
| self._loop.call_soon_threadsafe(self._reporter.submit, signature, body) | |
| except Exception: # noqa: BLE001 - logging must never break the code that logs | |
| pass | |
| def attach(self, logger: logging.Logger) -> None: | |
| logger.addHandler(self) | |
| self._logger = logger | |
| def detach(self) -> None: | |
| if self._logger is not None: | |
| self._logger.removeHandler(self) | |
| self._logger = None | |
| async def aclose(self, timeout: float = 5.0) -> None: | |
| """Stop listening, then give the reports that are on their way a moment to arrive.""" | |
| self.detach() | |
| await asyncio.sleep(0) # reports handed over from another thread land in the loop first | |
| await self._reporter.flush(timeout) | |
| def install(send: Send, secrets: Iterable[str] = (), *, logger: Optional[logging.Logger] = None) -> ChannelLogHandler: | |
| """Send the errors logged on ``logger`` (the root logger by default) through ``send``. Call it inside the event loop.""" | |
| handler = ChannelLogHandler(ErrorReporter(send), asyncio.get_running_loop(), secrets) | |
| handler.attach(logger or logging.getLogger()) | |
| return handler | |