"""Shared helper classes for gateway platform adapters. Extracts common patterns that were duplicated across 5-7 adapters: message deduplication, text batch aggregation, markdown stripping, and thread participation tracking. """ import asyncio import json import logging import re import time from pathlib import Path from typing import TYPE_CHECKING, Dict from utils import atomic_json_write if TYPE_CHECKING: from gateway.platforms.base import MessageEvent logger = logging.getLogger(__name__) # ─── Message Deduplication ──────────────────────────────────────────────────── class MessageDeduplicator: """TTL-based message deduplication cache. Replaces the identical ``_seen_messages`` / ``_is_duplicate()`` pattern previously duplicated in discord, slack, dingtalk, wecom, weixin, mattermost, and feishu adapters. Usage:: self._dedup = MessageDeduplicator() # In message handler: if self._dedup.is_duplicate(msg_id): return """ def __init__(self, max_size: int = 2000, ttl_seconds: float = 300): self._seen: Dict[str, float] = {} self._max_size = max_size self._ttl = ttl_seconds def is_duplicate(self, msg_id: str) -> bool: """Return True if *msg_id* was already seen within the TTL window.""" if not msg_id: return False now = time.time() if msg_id in self._seen: if now - self._seen[msg_id] < self._ttl: return True # Entry has expired — remove it and treat as new del self._seen[msg_id] self._seen[msg_id] = now if len(self._seen) > self._max_size: cutoff = now - self._ttl self._seen = {k: v for k, v in self._seen.items() if v > cutoff} if len(self._seen) > self._max_size: # TTL pruning alone does not cap the cache when every entry is # still fresh. Keep the newest entries so the helper's # max_size bound is enforced under sustained traffic. newest = sorted( self._seen.items(), key=lambda item: item[1], )[-self._max_size:] self._seen = dict(newest) return False def contains(self, msg_id: str) -> bool: """Return whether *msg_id* is live in the cache without inserting it.""" if not msg_id: return False seen_at = self._seen.get(msg_id) if seen_at is None: return False if time.time() - seen_at < self._ttl: return True del self._seen[msg_id] return False def discard(self, msg_id: str) -> None: """Release a claimed message ID after cancelled/failed handoff.""" self._seen.pop(msg_id, None) def clear(self): """Clear all tracked messages.""" self._seen.clear() # ─── Text Batch Aggregation ────────────────────────────────────────────────── class TextBatchAggregator: """Aggregates rapid-fire text events into single messages. Replaces the ``_enqueue_text_event`` / ``_flush_text_batch`` pattern previously duplicated in telegram, discord, matrix, wecom, and feishu. Usage:: self._text_batcher = TextBatchAggregator( handler=self._message_handler, batch_delay=0.6, split_threshold=1900, ) # In message dispatch: if msg_type == MessageType.TEXT and self._text_batcher.is_enabled(): self._text_batcher.enqueue(event, session_key) return """ def __init__( self, handler, *, batch_delay: float = 0.6, split_delay: float = 2.0, split_threshold: int = 4000, ): self._handler = handler self._batch_delay = batch_delay self._split_delay = split_delay self._split_threshold = split_threshold self._pending: Dict[str, "MessageEvent"] = {} self._pending_tasks: Dict[str, asyncio.Task] = {} def is_enabled(self) -> bool: """Return True if batching is active (delay > 0).""" return self._batch_delay > 0 def enqueue(self, event: "MessageEvent", key: str) -> None: """Add *event* to the pending batch for *key*.""" chunk_len = len(event.text or "") existing = self._pending.get(key) if not existing: event._last_chunk_len = chunk_len # type: ignore[attr-defined] self._pending[key] = event else: existing.text = f"{existing.text}\n{event.text}" existing._last_chunk_len = chunk_len # type: ignore[attr-defined] # Cancel prior flush timer, start a new one prior = self._pending_tasks.get(key) if prior and not prior.done(): prior.cancel() self._pending_tasks[key] = asyncio.create_task(self._flush(key)) async def _flush(self, key: str) -> None: """Wait then dispatch the batched event for *key*.""" current_task = self._pending_tasks.get(key) pending = self._pending.get(key) last_len = getattr(pending, "_last_chunk_len", 0) if pending else 0 # Use longer delay when the last chunk looks like a split message delay = self._split_delay if last_len >= self._split_threshold else self._batch_delay await asyncio.sleep(delay) event = self._pending.pop(key, None) if event: try: await self._handler(event) except Exception: logger.exception("[TextBatchAggregator] Error dispatching batched event for %s", key) if self._pending_tasks.get(key) is current_task: self._pending_tasks.pop(key, None) def cancel_all(self) -> None: """Cancel all pending flush tasks.""" for task in self._pending_tasks.values(): if not task.done(): task.cancel() self._pending_tasks.clear() self._pending.clear() # ─── Markdown Stripping ────────────────────────────────────────────────────── # Pre-compiled regexes for performance _RE_BOLD = re.compile(r"\*\*(.+?)\*\*", re.DOTALL) _RE_ITALIC_STAR = re.compile(r"\*(.+?)\*", re.DOTALL) _RE_BOLD_UNDER = re.compile(r"\b__(?![\s_])(.+?)(? str: """Strip markdown formatting for plain-text platforms (SMS, iMessage, etc.). Replaces the identical ``_strip_markdown()`` functions previously duplicated in sms.py, bluebubbles.py, and feishu.py. """ text = _RE_BOLD.sub(r"\1", text) text = _RE_ITALIC_STAR.sub(r"\1", text) text = _RE_BOLD_UNDER.sub(r"\1", text) text = _RE_ITALIC_UNDER.sub(r"\1", text) text = _RE_CODE_BLOCK.sub("", text) text = _RE_INLINE_CODE.sub(r"\1", text) text = _RE_HEADING.sub("", text) text = _RE_LINK.sub(r"\1", text) text = _RE_MULTI_NEWLINE.sub("\n\n", text) return text.strip() # ─── Thread Participation Tracking ─────────────────────────────────────────── class ThreadParticipationTracker: """Persistent tracking of threads the bot has participated in. Replaces the identical ``_load/_save_participated_threads`` + ``_mark_thread_participated`` pattern previously duplicated in discord.py and matrix.py. Usage:: self._threads = ThreadParticipationTracker("discord") # Check membership: if thread_id in self._threads: ... # Mark participation: self._threads.mark(thread_id) """ _MAX_TRACKED = 500 def __init__(self, platform_name: str, max_tracked: int = 500): self._platform = platform_name self._max_tracked = max_tracked self._threads: dict[str, None] = { str(thread_id): None for thread_id in self._load() } def _state_path(self) -> Path: from hermes_constants import get_hermes_home return get_hermes_home() / f"{self._platform}_threads.json" def _load(self) -> list[str]: path = self._state_path() if path.exists(): try: data = json.loads(path.read_text(encoding="utf-8")) if isinstance(data, list): return [str(thread_id) for thread_id in data] except Exception: pass return [] def _save(self) -> None: path = self._state_path() thread_list = list(self._threads) if len(thread_list) > self._max_tracked: thread_list = thread_list[-self._max_tracked:] self._threads = dict.fromkeys(thread_list) atomic_json_write(path, thread_list, indent=None) def mark(self, thread_id: str) -> None: """Mark *thread_id* as participated and persist.""" if thread_id not in self._threads: self._threads[thread_id] = None self._save() def __contains__(self, thread_id: str) -> bool: return thread_id in self._threads def clear(self) -> None: self._threads.clear() # ─── Phone Number Redaction ────────────────────────────────────────────────── def redact_phone(phone: str) -> str: """Redact a phone number for logging, preserving country code and last 4. Replaces the identical ``_redact_phone()`` functions in signal.py, sms.py, and bluebubbles.py. """ if not phone: return "" if len(phone) <= 8: return phone[:2] + "****" + phone[-2:] if len(phone) > 4 else "****" return phone[:4] + "****" + phone[-4:] # ─── GFM Markdown Table → Bullet Conversion ───────────────────────────────── # Shared by Discord and Telegram adapters. Discord calls # convert_table_to_bullets() directly; Telegram imports the primitives # but keeps its own MarkdownV2-aware renderer. # Matches a GFM table delimiter row: optional outer pipes, cells of dashes # (with optional alignment colons) separated by '|'. # Requires at least one internal '|' so lone '---' rules are NOT matched. TABLE_SEPARATOR_RE = re.compile( r'^\s*\|?\s*:?-+:?\s*(?:\|\s*:?-+:?\s*){1,}\|?\s*$' ) def is_table_row(line: str) -> bool: """Return True if *line* could plausibly be a table data row.""" stripped = line.strip() return bool(stripped) and '|' in stripped def split_markdown_table_row(line: str) -> list[str]: """Split a GFM table row into stripped cell values. Thin delegate to the canonical implementation in :mod:`agent.markdown_tables` (``split_table_row``) so the three formerly byte-identical copies (here, ``agent/markdown_tables.py``, ``weixin._split_table_row``) share one body. """ from agent.markdown_tables import split_table_row return split_table_row(line) def _render_table_block(table_block: list[str]) -> str: """Render a detected GFM table as bold-heading + bullet groups. Uses the same alignment logic as Telegram's renderer: for non-row-label tables, ``data_cells = cells`` (the full row) and the bullet whose value duplicates the heading is skipped. This keeps header→value alignment correct. """ if len(table_block) < 3: return "\n".join(table_block) headers = split_markdown_table_row(table_block[0]) if len(headers) < 2: return "\n".join(table_block) first_data_row = ( split_markdown_table_row(table_block[2]) if len(table_block) > 2 else [] ) has_row_label_col = len(first_data_row) == len(headers) + 1 rendered_groups: list[str] = [] for index, row in enumerate(table_block[2:], start=1): cells = split_markdown_table_row(row) if has_row_label_col: heading = cells[0] if cells and cells[0] else f"Row {index}" data_cells = cells[1:] else: heading = next((cell for cell in cells if cell), f"Row {index}") data_cells = cells if len(data_cells) < len(headers): data_cells.extend([""] * (len(headers) - len(data_cells))) elif len(data_cells) > len(headers): data_cells = data_cells[: len(headers)] bullets: list[str] = [] for header, value in zip(headers, data_cells): if not has_row_label_col and value == heading: continue bullets.append(f"• {header}: {value}") group_lines = [f"**{heading}**", *bullets] rendered_groups.append("\n".join(group_lines)) return "\n\n".join(rendered_groups) def convert_table_to_bullets(text: str) -> str: """Rewrite GFM pipe tables into bold-heading + bullet groups. Tables inside fenced code blocks are left alone. """ if '|' not in text or '-' not in text: return text lines = text.split('\n') out: list[str] = [] in_fence = False i = 0 while i < len(lines): line = lines[i] stripped = line.lstrip() if stripped.startswith('```'): in_fence = not in_fence out.append(line) i += 1 continue if in_fence: out.append(line) i += 1 continue if ( '|' in line and i + 1 < len(lines) and TABLE_SEPARATOR_RE.match(lines[i + 1]) ): table_block = [line, lines[i + 1]] j = i + 2 while j < len(lines) and is_table_row(lines[j]): table_block.append(lines[j]) j += 1 out.append(_render_table_block(table_block)) i = j continue out.append(line) i += 1 return '\n'.join(out) # ─── Mention-pattern compilation ───────────────────────────────────────────── def compile_mention_patterns( raw, *, log_prefix: str, platform_label: str | None = None, display_label: str | None = None, defaults: 'list[str] | None' = None, logger_: 'logging.Logger | None' = None, ) -> 'list[re.Pattern]': """Compile regex wake-word/mention patterns from config or env values. Two adapter families share this logic: * **Config-style** (dingtalk, telegram): pass ``platform_label`` (e.g. ``"dingtalk"``). ``raw`` is the value from ``config.extra`` after env fallback parsing; must be a list or string, anything else logs a warning and yields ``[]``. Non-string entries are skipped. A summary info log is emitted when patterns load. * **Wakeword-style** (photon, bluebubbles): pass ``defaults``. ``raw`` may be None (use defaults), a string (JSON list or comma/newline separated), a list, or a scalar (wrapped in a list). Entries are coerced via ``str()``. ``log_prefix`` is interpolated into every log message so per-adapter log output stays byte-identical to the historical inline implementations. """ log = logger_ or logger if platform_label is not None: # Config-style (dingtalk/telegram) semantics. display = display_label or platform_label patterns = raw if patterns is None: return [] if isinstance(patterns, str): patterns = [patterns] if not isinstance(patterns, list): log.warning( "[%s] %s mention_patterns must be a list or string; got %s", log_prefix, platform_label, type(patterns).__name__, ) return [] compiled: list[re.Pattern] = [] for pattern in patterns: if not isinstance(pattern, str) or not pattern.strip(): continue try: compiled.append(re.compile(pattern, re.IGNORECASE)) except re.error as exc: log.warning( "[%s] Invalid %s mention pattern %r: %s", log_prefix, display, pattern, exc, ) if compiled: log.info( "[%s] Loaded %d %s mention pattern(s)", log_prefix, len(compiled), display, ) return compiled # Wakeword-style (photon/bluebubbles) semantics. if raw is None: patterns = list(defaults or []) elif isinstance(raw, str): text = raw.strip() try: loaded = json.loads(text) if text else [] except Exception: loaded = None patterns = loaded if isinstance(loaded, list) else [ part.strip() for line in text.splitlines() for part in line.split(",") ] elif isinstance(raw, list): patterns = raw else: patterns = [raw] compiled = [] for pattern in patterns: text = str(pattern).strip() if not text: continue try: compiled.append(re.compile(text, re.IGNORECASE)) except re.error as exc: log.warning("[%s] Invalid mention pattern %r: %s", log_prefix, text, exc) return compiled # ─── Fence-Aware Markdown Chunking ─────────────────────────────────────────── # Shared core for the fence-aware markdown chunkers that previously lived as # near-duplicates in gateway/stream_consumer.py, gateway/platforms/yuanbao.py # (MarkdownProcessor — the richest version, which this core is derived from), # and gateway/platforms/weixin.py. Each caller keeps its own knobs: # # * stream_consumer: newline-preferred splitting + close/reopen fence # balancing (``prefer_paragraphs=False, balance_fences=True``) # * yuanbao: atomic-block extraction + paragraph-boundary splitting, fences # kept intact as atoms (``prefer_paragraphs=True, balance_fences=False``) # * weixin: keeps its own block splitter (anchored ``_FENCE_RE``, per-line # rstrip semantics) but reuses ``greedy_pack_blocks`` for packing. # # The typing helpers below use ``Optional``/``Callable`` from ``typing`` to # match the module's existing import style. def text_has_unclosed_fence(text: str) -> bool: """Return True when *text* ends inside an unclosed ``` code fence. Scans line by line, toggling in/out state on lines starting with ```. An odd number of toggles means the trailing fence is unclosed. """ in_fence = False for line in text.split('\n'): if line.startswith('```'): in_fence = not in_fence return in_fence def text_ends_with_table_row(text: str) -> bool: """True when the last non-empty line starts and ends with ``|``.""" trimmed = text.rstrip() if not trimmed: return False last_line = trimmed.split('\n')[-1].strip() return last_line.startswith('|') and last_line.endswith('|') def is_fence_atom(text: str) -> bool: """True when an atomic block is a code block (starts with ```).""" return text.lstrip().startswith('```') def is_table_atom(text: str) -> bool: """True when an atomic block is a table (first line is ``|...|``).""" first_line = text.split('\n')[0].strip() return first_line.startswith('|') and first_line.endswith('|') _SENTENCE_END_NEWLINE_RE = re.compile(r'[。！？.!?]\n') def split_at_paragraph_boundary(text, max_chars, len_fn=None): """Find the nearest paragraph boundary within *max_chars*; return (head, tail). Split priority: 1. Blank line (paragraph boundary) 2. Newline after sentence-ending punctuation (CJK and ASCII) 3. Last newline 4. Force split at the *max_chars* window boundary ``head + tail == text`` always holds. *len_fn* allows measuring in custom units (e.g. UTF-16 code units); a binary search finds the largest prefix that fits when it is provided. """ _len = len_fn or len if _len(text) <= max_chars: return text, '' if _len is len: window = text[:max_chars] else: lo, hi = 0, len(text) while lo < hi: mid = (lo + hi + 1) // 2 if _len(text[:mid]) <= max_chars: lo = mid else: hi = mid - 1 window = text[:lo] # 1. Prefer the last blank line (\n\n) as paragraph boundary pos = window.rfind('\n\n') if pos > 0: return text[:pos + 2], text[pos + 2:] # 2. Then the last newline following sentence-ending punctuation best_pos = -1 for m in _SENTENCE_END_NEWLINE_RE.finditer(window): best_pos = m.end() if best_pos > 0: return text[:best_pos], text[best_pos:] # 3. Fallback: last newline pos = window.rfind('\n') if pos > 0: return text[:pos + 1], text[pos + 1:] # 4. No valid split point: force split at the window boundary cut = len(window) return text[:cut], text[cut:] def split_markdown_atoms(text: str) -> "list[str]": """Split markdown into indivisible "atomic blocks". Atoms are: fenced code blocks (``` ... ``` inclusive), tables (consecutive ``|...|`` lines), and plain paragraphs separated by blank lines. Blank lines are separators and belong to no atom. """ lines = text.split('\n') atoms: "list[str]" = [] current_lines: "list[str]" = [] in_fence = False def _is_table_line(line: str) -> bool: stripped = line.strip() return stripped.startswith('|') and stripped.endswith('|') def _flush_current() -> None: if current_lines: atom = '\n'.join(current_lines) if atom.strip(): atoms.append(atom) current_lines.clear() for line in lines: if in_fence: current_lines.append(line) if line.startswith('```') and len(current_lines) > 1: in_fence = False _flush_current() elif line.startswith('```'): _flush_current() in_fence = True current_lines.append(line) elif _is_table_line(line): if current_lines and not _is_table_line(current_lines[-1]): _flush_current() current_lines.append(line) elif line.strip() == '': _flush_current() else: if current_lines and _is_table_line(current_lines[-1]): _flush_current() current_lines.append(line) _flush_current() return atoms def infer_block_separator(prev_chunk: str, next_chunk: str) -> str: """Infer the separator (``'\\n'`` or ``'\\n\\n'``) between two chunks. Single newline when the boundary sits at a code fence or a continued table; paragraph separator otherwise. """ prev_trimmed = prev_chunk.rstrip() next_trimmed = next_chunk.lstrip() if prev_trimmed.endswith('```') or next_trimmed.startswith('```'): return '\n' if text_ends_with_table_row(prev_chunk): first_line = next_trimmed.split('\n')[0].strip() if next_trimmed else '' if first_line.startswith('|') and first_line.endswith('|'): return '\n' return '\n\n' def merge_streaming_fences(chunks: "list[str]") -> "list[str]": """Stream-aware fence merge: rejoin chunks truncated mid-fence. While chunk *i* has an unclosed fence and a successor exists, merge the successor into it using :func:`infer_block_separator`. """ if not chunks: return [] result: "list[str]" = [] i = 0 while i < len(chunks): current = chunks[i] while text_has_unclosed_fence(current) and i + 1 < len(chunks): sep = infer_block_separator(current, chunks[i + 1]) current = current + sep + chunks[i + 1] i += 1 result.append(current) i += 1 return result def balance_fences_across_chunks(chunks: "list[str]") -> "list[str]": """Close orphaned ``` fences at each chunk boundary and reopen on the next. When a split lands inside a triple-backtick code block, close the fence at the end of the head chunk and reopen it (with the original language tag) at the start of the next, so every delivered chunk is fence-balanced on its own. """ if len(chunks) <= 1: return chunks out: "list[str]" = [] carry_lang = None for chunk in chunks: prefix = f"```{carry_lang}\n" if carry_lang is not None else "" in_code = carry_lang is not None lang = carry_lang or "" for line in chunk.split("\n"): stripped = line.strip() if stripped.startswith("```"): if in_code: in_code = False lang = "" else: in_code = True tag = stripped[3:].strip() lang = tag.split()[0] if tag else "" body = prefix + chunk if in_code: body += "\n```" carry_lang = lang else: carry_lang = None out.append(body) return out def greedy_pack_blocks(blocks, max_length, len_fn=None, sep="\n\n", overflow=None): """Greedily pack pre-split *blocks* into chunks of at most *max_length*. Blocks are joined with *sep* while they fit. A block that alone exceeds the limit is passed to *overflow(block)* (which must return a list of chunks) when provided, else emitted as-is. """ _len = len_fn or len packed: "list[str]" = [] current = "" for block in blocks: candidate = block if not current else f"{current}{sep}{block}" if _len(candidate) <= max_length: current = candidate continue if current: packed.append(current) current = "" if _len(block) <= max_length: current = block continue if overflow is not None: packed.extend(overflow(block)) else: packed.append(block) if current: packed.append(current) return packed def split_text_fence_aware( text, limit, len_fn=None, *, prefer_paragraphs=True, balance_fences=False, ): """Split markdown text into chunks of at most *limit*, respecting fences. Two strategies, selected by ``prefer_paragraphs``: ``prefer_paragraphs=True`` (yuanbao-derived, the richest): Extract atomic blocks (code fences, tables, paragraphs), greedily merge them up to *limit*, split still-oversized non-atomic chunks at paragraph boundaries, then re-merge small neighbours. Code blocks and tables are never split in the middle; a single atom larger than *limit* is emitted oversize rather than broken. ``prefer_paragraphs=False`` (stream_consumer-derived): Newline-preferred hard splitting with headroom reserved for fence markers when the text contains ```. ``balance_fences=True`` post-processes the chunks so a split inside a code block closes the fence on the head chunk and reopens it (with the language tag) on the tail — required by callers whose chunks are delivered as independent messages that each must render standalone. """ _len = len_fn or len if not text: return [] if prefer_paragraphs: chunks = _chunk_markdown_paragraphs(text, limit, len_fn) else: chunks = _chunk_newline_preferred(text, limit, _len) if balance_fences: chunks = balance_fences_across_chunks(chunks) return chunks def _chunk_markdown_paragraphs(text, max_chars, len_fn=None): """Yuanbao-derived paragraph/atom chunking pipeline (see module docs).""" _len = len_fn or len if _len(text) <= max_chars: return [text] # Phase 1: Extract atomic blocks atoms = split_markdown_atoms(text) # Phase 2: Greedy merge chunks: "list[str]" = [] indivisible_set: "set[int]" = set() current_parts: "list[str]" = [] current_len = 0 def _flush_parts() -> None: if current_parts: chunks.append('\n\n'.join(current_parts)) for atom in atoms: atom_len = _len(atom) sep_len = 2 if current_parts else 0 projected_len = current_len + sep_len + atom_len if projected_len > max_chars and current_parts: _flush_parts() current_parts = [] current_len = 0 sep_len = 0 if (not current_parts and atom_len > max_chars and (is_fence_atom(atom) or is_table_atom(atom))): indivisible_set.add(len(chunks)) chunks.append(atom) continue current_parts.append(atom) current_len += sep_len + atom_len _flush_parts() # Phase 3: Split still-oversized chunks at paragraph boundaries result: "list[str]" = [] for idx, chunk in enumerate(chunks): if _len(chunk) <= max_chars: result.append(chunk) continue if idx in indivisible_set: result.append(chunk) continue if text_has_unclosed_fence(chunk): result.append(chunk) continue remaining = chunk while _len(remaining) > max_chars: head, remaining = split_at_paragraph_boundary( remaining, max_chars, len_fn=len_fn, ) if not head: head, remaining = remaining[:max_chars], remaining[max_chars:] if head: result.append(head) if remaining: result.append(remaining) # Phase 4: Merge small trailing/leading chunks with neighbours if len(result) > 1: merged: "list[str]" = [result[0]] for chunk in result[1:]: prev = merged[-1] combined = prev + '\n\n' + chunk if _len(combined) <= max_chars: merged[-1] = combined else: merged.append(chunk) result = merged return [c for c in result if c] def _chunk_newline_preferred(text, limit, len_fn): """Stream-consumer-derived newline-preferred splitting (no balancing).""" if len_fn(text) <= limit: return [text] # Reserve headroom for the close/reopen fence markers a balancing pass # may add, so balanced chunks stay within the platform limit. split_limit = limit if "```" in text: split_limit = max(limit - 16, limit // 2, 1) # Local import: gateway.platforms.base is heavyweight and pulls config; # helpers must stay import-light for adapters that import it first. from gateway.platforms.base import _custom_unit_to_cp chunks: "list[str]" = [] remaining = text while len_fn(remaining) > split_limit: _cp_budget = _custom_unit_to_cp(remaining, split_limit, len_fn) split_at = remaining.rfind("\n", 0, _cp_budget) if split_at < _cp_budget // 2: split_at = _cp_budget chunks.append(remaining[:split_at]) remaining = remaining[split_at:].lstrip("\n") if remaining: chunks.append(remaining) return chunks