86 lines
2.8 KiB
Python
86 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
from typing import Any
|
|
|
|
|
|
def chunk_message(text: str, limit: int = 1900) -> list[str]:
|
|
if len(text) <= limit:
|
|
return [text]
|
|
|
|
chunks: list[str] = []
|
|
remaining = text
|
|
while remaining:
|
|
if len(remaining) <= limit:
|
|
chunks.append(remaining)
|
|
break
|
|
|
|
split_at = remaining.rfind("\n", 0, limit)
|
|
if split_at <= 0:
|
|
split_at = limit
|
|
|
|
chunks.append(remaining[:split_at])
|
|
remaining = remaining[split_at:]
|
|
if remaining.startswith("\n"):
|
|
remaining = remaining[1:]
|
|
|
|
return chunks
|
|
|
|
|
|
def summarize_error(body: Any, fallback: str) -> str:
|
|
if isinstance(body, dict):
|
|
err = body.get("error")
|
|
if isinstance(err, dict):
|
|
msg = err.get("message")
|
|
if isinstance(msg, str) and msg.strip():
|
|
return _compact_error_text(msg)
|
|
|
|
msg = body.get("message")
|
|
if isinstance(msg, str) and msg.strip():
|
|
return _compact_error_text(msg)
|
|
|
|
return _compact_error_text(fallback)
|
|
|
|
|
|
def _compact_error_text(text: str, limit: int = 240) -> str:
|
|
clean = text.strip()
|
|
if not clean:
|
|
return "Unknown provider error"
|
|
|
|
lower = clean.lower()
|
|
if "<!doctype html" in lower or "<html" in lower:
|
|
title_match = re.search(r"<title>\s*([^<]+?)\s*</title>", clean, flags=re.IGNORECASE | re.DOTALL)
|
|
title = re.sub(r"\s+", " ", title_match.group(1)).strip() if title_match else "HTML error page"
|
|
code_match = re.search(r"(?:error code\s*)?([45]\d{2})", clean, flags=re.IGNORECASE)
|
|
code = code_match.group(1) if code_match else None
|
|
phrase = "Bad gateway" if "bad gateway" in lower else title
|
|
if "cloudflare" in lower or "openrouter.ai" in lower:
|
|
if code:
|
|
return f"Cloudflare/OpenRouter returned HTTP {code} {phrase}"
|
|
return f"Cloudflare/OpenRouter returned {phrase}"
|
|
if code:
|
|
return f"Provider returned HTTP {code} HTML error page: {title}"
|
|
return f"Provider returned HTML error page: {title}"
|
|
|
|
clean = re.sub(r"\s+", " ", clean)
|
|
if len(clean) > limit:
|
|
return clean[: limit - 1].rstrip() + "…"
|
|
return clean
|
|
|
|
|
|
def build_transcript_block(display_name: str, transcript: str) -> str:
|
|
clean_name = display_name.strip() or "Unknown User"
|
|
clean_transcript = transcript.strip()
|
|
return f"[{clean_name}]\n{clean_transcript}"
|
|
|
|
|
|
def command_channel_error(current_channel_id: int | None, allowed_channel_id: int | None) -> str | None:
|
|
if allowed_channel_id is None:
|
|
return "⚠️ This bot doesn't have a command channel yet. Use `/set_output` first."
|
|
|
|
if current_channel_id != allowed_channel_id:
|
|
return f"⚠️ This command only works in <#{allowed_channel_id}>."
|
|
|
|
return None
|