ROMFASTSQL/proxmox/lxc171-claude-agent/discord-bridge/render.py

"""Randare pentru Discord: chunker + un SINGUR loop de editare per CANAL.

T10. Bucket-ul de rate limit al Discord e per canal, deci loop-ul e per canal, coalescent,
cu interval adaptiv 1s -> 5s. Modulul NU importa discord.py: I/O-ul il face adaptorul.
"""

from __future__ import annotations

import asyncio
import contextlib
import logging
import re
import time
from dataclasses import dataclass, field
from typing import Any, Awaitable, Callable

log = logging.getLogger("discord-bridge.render")

MAX_MSG = 2000
ATTACH_OVER = 6000
_FENCE = re.compile(r"^```([A-Za-z0-9_+.\-]*)\s*$")


@dataclass
class Attachment:
    filename: str
    content: str
    preview: str

    kind: str = "attachment"


@dataclass
class Chunks:
    parts: list[str] = field(default_factory=list)
    kind: str = "chunks"


def split_message(text: str, limit: int = MAX_MSG, attach_over: int = ATTACH_OVER,
                  filename: str = "raspuns.md") -> Chunks | Attachment:
    """Chunker-ul, intr-o singura functie.

    - peste `attach_over` caractere -> obiect Attachment (adaptorul il urca in Discord)
    - altfel bucati de cel mult `limit` caractere, fara sa rupa un fence ``` peste granita:
      fence-ul se inchide la finalul bucatii si se redeschide cu acelasi limbaj in urmatoarea.
    """
    text = text or ""
    if len(text) > attach_over:
        head = text[: limit - 200]
        return Attachment(
            filename=filename,
            content=text,
            preview=head + "\n...\n(raspuns lung, atasat integral)",
        )

    parts: list[str] = []
    cur: list[str] = []
    cur_len = 0
    fence_lang: str | None = None  # limbajul fence-ului deschis in bucata curenta

    def flush(reopen: bool) -> None:
        nonlocal cur, cur_len
        if not cur:
            return
        body = "\n".join(cur)
        if fence_lang is not None:
            body += "\n```"
        parts.append(body)
        cur = [f"```{fence_lang}"] if (reopen and fence_lang is not None) else []
        cur_len = sum(len(x) + 1 for x in cur)

    for line in text.split("\n"):
        # o linie singura mai lunga decat limita se taie dur
        pieces = [line] if len(line) <= limit - 10 else [
            line[i: i + limit - 10] for i in range(0, len(line), limit - 10)
        ]
        for piece in pieces:
            need = len(piece) + 1 + (4 if fence_lang is not None else 0)
            if cur and cur_len + need > limit:
                flush(reopen=True)
            cur.append(piece)
            cur_len += len(piece) + 1
            m = _FENCE.match(piece.strip())
            if m:
                fence_lang = None if fence_lang is not None else (m.group(1) or "")
    flush(reopen=False)
    if not parts:
        parts = [""]
    return Chunks(parts=parts)


class ChannelEditLoop:
    """Un loop de editare per canal. Coalescent: conteaza doar ultimul text per tinta."""

    def __init__(
        self,
        channel_id: str,
        edit_fn: Callable[[Any, str], Awaitable[None]],
        *,
        min_interval: float = 1.0,
        max_interval: float = 5.0,
        clock: Callable[[], float] = time.monotonic,
    ):
        self.channel_id = str(channel_id)
        self.edit_fn = edit_fn
        self.min_interval = min_interval
        self.max_interval = max_interval
        self.interval = min_interval
        self.clock = clock
        self.pending: dict[Any, str] = {}
        self.edits = 0
        self.coalesced = 0
        self.errors = 0
        self._wake = asyncio.Event()
        self._task: asyncio.Task | None = None
        self._stop = False

    # ------------------------------------------------------------ interfata
    def queue(self, target: Any, text: str) -> None:
        """Ultima versiune castiga; actualizarile intermediare se pierd intentionat."""
        if target in self.pending:
            self.coalesced += 1
        self.pending[target] = text
        self._wake.set()

    def note_rate_limited(self) -> None:
        """Adaptorul semnaleaza un 429: urcam direct la intervalul maxim."""
        self.interval = self.max_interval

    def start(self) -> asyncio.Task:
        if self._task is None or self._task.done():
            self._stop = False
            self._task = asyncio.create_task(self._run())
        return self._task

    async def stop(self, flush: bool = True) -> None:
        self._stop = True
        self._wake.set()
        if self._task is not None:
            with contextlib.suppress(asyncio.CancelledError, Exception):
                await asyncio.wait_for(self._task, 5.0)
            self._task = None
        if flush:
            await self.flush()

    async def flush(self) -> None:
        """Trimite imediat tot ce e in asteptare (finalul unui tur)."""
        while self.pending:
            target, text = next(iter(self.pending.items()))
            del self.pending[target]
            await self._edit(target, text)

    # -------------------------------------------------------------- interne
    async def _edit(self, target: Any, text: str) -> None:
        try:
            await self.edit_fn(target, text)
            self.edits += 1
        except Exception:
            self.errors += 1
            log.warning("editare esuata pe canalul %s", self.channel_id, exc_info=True)

    async def _run(self) -> None:
        while not self._stop:
            if not self.pending:
                self._wake.clear()
                with contextlib.suppress(asyncio.TimeoutError):
                    await asyncio.wait_for(self._wake.wait(), self.max_interval)
                # liniste -> coboram spre intervalul minim
                self.interval = max(self.min_interval, self.interval / 1.5)
                continue
            batch = self.pending
            self.pending = {}
            for target, text in batch.items():
                await self._edit(target, text)
            if len(batch) > 1:
                self.interval = min(self.max_interval, self.interval * 1.5)
            await asyncio.sleep(self.interval)


class RenderManager:
    """Un singur ChannelEditLoop per canal."""

    def __init__(self, edit_fn: Callable[[Any, str], Awaitable[None]], **kw):
        self.edit_fn = edit_fn
        self.kw = kw
        self.loops: dict[str, ChannelEditLoop] = {}

    def loop_for(self, channel_id: str) -> ChannelEditLoop:
        key = str(channel_id)
        loop = self.loops.get(key)
        if loop is None:
            loop = self.loops[key] = ChannelEditLoop(key, self.edit_fn, **self.kw)
            loop.start()
        return loop

    def queue(self, channel_id: str, target: Any, text: str) -> None:
        self.loop_for(channel_id).queue(target, text)

    async def stop_all(self) -> None:
        for loop in list(self.loops.values()):
            await loop.stop()
        self.loops.clear()


def footer(model: str, duration_ms: int, cost_usd: float, thread_total: float) -> str:
    """Subsolul cerut la T11: model, durata, cost tur, cost cumulat pe fir."""
    return (
        f"-# {model} · {duration_ms / 1000:.1f}s · ${cost_usd:.4f} tur · ${thread_total:.4f} fir"
    )