init

bot/services/analyzer.py

@@ -0,0 +1,124 @@
+import asyncio
+import logging
+from collections.abc import Callable, Coroutine
+from typing import Any
+
+import anthropic
+
+from bot.config import settings
+from bot.models import ReportType
+from bot.prompts.chunk_summary import build_chunk_prompt
+from bot.prompts.synthesis import build_synthesis_prompt
+
+log = logging.getLogger(__name__)
+
+# Only 1 concurrent request to stay within rate limits
+_semaphore = asyncio.Semaphore(1)
+
+MAX_RETRIES = 5
+
+
+async def _call_claude(
+    client: anthropic.AsyncAnthropic,
+    system: str,
+    user: str,
+    max_tokens: int,
+    on_progress: "ProgressCallback | None" = None,
+) -> str:
+    for attempt in range(MAX_RETRIES):
+        try:
+            async with _semaphore:
+                kwargs: dict[str, Any] = {
+                    "model": settings.claude_model,
+                    "max_tokens": max_tokens,
+                    "system": system,
+                    "messages": [{"role": "user", "content": user}],
+                }
+                # Adaptive thinking only works on Opus 4.6
+                if "opus-4-6" in settings.claude_model:
+                    kwargs["thinking"] = {"type": "adaptive"}
+
+                async with client.messages.stream(**kwargs) as stream:
+                    response = await stream.get_final_message()
+
+            return "".join(
+                block.text for block in response.content if block.type == "text"
+            )
+        except anthropic.RateLimitError as e:
+            wait = getattr(e, "retry_after", None) or 60
+            log.warning("Rate limited, waiting %ds (attempt %d/%d)", wait, attempt + 1, MAX_RETRIES)
+            if on_progress:
+                await on_progress(f"Rate limited, waiting {wait}s...")
+            await asyncio.sleep(wait)
+        except anthropic.APIStatusError as e:
+            if e.status_code >= 500:
+                wait = 10 * (attempt + 1)
+                log.warning("Server error %d, retrying in %ds", e.status_code, wait)
+                await asyncio.sleep(wait)
+            else:
+                raise
+
+    raise RuntimeError("Max retries exceeded due to rate limiting")
+
+
+ProgressCallback = Callable[[str], Coroutine[Any, Any, None]]
+
+
+async def analyze_channel(
+    report_type: ReportType,
+    chunks: list[str],
+    channel_title: str,
+    channel_username: str | None,
+    subscribers: int | None,
+    msg_count: int,
+    on_progress: ProgressCallback | None = None,
+) -> str:
+    client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
+
+    total = len(chunks)
+    summaries: list[str] = []
+
+    for i, chunk_text in enumerate(chunks, 1):
+        if on_progress:
+            await on_progress(f"Analyzing chunk {i}/{total}...")
+
+        prompt = build_chunk_prompt(
+            report_type, channel_title, chunk_text, i, total
+        )
+        summary = await _call_claude(
+            client,
+            system="You are an expert Telegram channel analyst.",
+            user=prompt,
+            max_tokens=4096,
+            on_progress=on_progress,
+        )
+        summaries.append(summary)
+        log.info("Chunk %d/%d summarized (%d chars)", i, total, len(summary))
+
+        # Wait 60s between chunks — rate limit is 30K input tokens/min
+        if i < total:
+            if on_progress:
+                await on_progress(f"Chunk {i}/{total} done. Cooling down 60s for rate limit...")
+            await asyncio.sleep(60)
+
+    if on_progress:
+        await on_progress("Generating final report...")
+
+    synthesis_prompt = build_synthesis_prompt(
+        report_type,
+        channel_title,
+        channel_username,
+        subscribers,
+        msg_count,
+        summaries,
+    )
+    report = await _call_claude(
+        client,
+        system="You are an expert Telegram channel analyst producing a final report.",
+        user=synthesis_prompt,
+        max_tokens=16000,
+        on_progress=on_progress,
+    )
+
+    log.info("Final report generated (%d chars)", len(report))
+    return report

bot/services/chunker.py

@@ -0,0 +1,45 @@
+from bot.models import ChannelMessage
+
+MAX_TOKENS_PER_CHUNK = 25_000
+CHARS_PER_TOKEN = 3
+
+
+def _format_message(msg: ChannelMessage) -> str:
+    parts = [f"[{msg.date:%Y-%m-%d %H:%M}]"]
+    stats = []
+    if msg.views is not None:
+        stats.append(f"{msg.views} views")
+    if msg.forwards is not None:
+        stats.append(f"{msg.forwards} fwd")
+    if msg.replies is not None:
+        stats.append(f"{msg.replies} replies")
+    if stats:
+        parts.append(f"({', '.join(stats)})")
+    parts.append(f"\n{msg.text}")
+    return " ".join(parts)
+
+
+def chunk_messages(messages: list[ChannelMessage]) -> list[str]:
+    """Split messages into token-bounded chunks."""
+    max_chars = MAX_TOKENS_PER_CHUNK * CHARS_PER_TOKEN
+    chunks: list[str] = []
+    current_lines: list[str] = []
+    current_len = 0
+
+    for msg in messages:
+        formatted = _format_message(msg)
+        entry = formatted + "\n---\n"
+        entry_len = len(entry)
+
+        if current_len + entry_len > max_chars and current_lines:
+            chunks.append("".join(current_lines))
+            current_lines = []
+            current_len = 0
+
+        current_lines.append(entry)
+        current_len += entry_len
+
+    if current_lines:
+        chunks.append("".join(current_lines))
+
+    return chunks

bot/services/fetcher.py

@@ -0,0 +1,43 @@
+import logging
+
+from telethon import TelegramClient
+from telethon.tl.functions.channels import GetFullChannelRequest
+
+from bot.models import ChannelMessage
+
+log = logging.getLogger(__name__)
+
+
+async def fetch_channel_messages(
+    client: TelegramClient,
+    channel: str,
+    limit: int | None = None,
+) -> tuple[list[ChannelMessage], dict]:
+    """Fetch messages and basic stats from a public channel."""
+    entity = await client.get_entity(channel)
+    full = await client(GetFullChannelRequest(entity))
+
+    stats = {
+        "title": entity.title,
+        "username": getattr(entity, "username", None),
+        "subscribers": full.full_chat.participants_count,
+    }
+
+    messages: list[ChannelMessage] = []
+    async for msg in client.iter_messages(entity, limit=limit):
+        if not msg.text:
+            continue
+        messages.append(
+            ChannelMessage(
+                id=msg.id,
+                date=msg.date,
+                text=msg.text,
+                views=msg.views,
+                forwards=msg.forwards,
+                replies=msg.replies.replies if msg.replies else None,
+            )
+        )
+
+    messages.reverse()
+    log.info("Fetched %d text messages from %s", len(messages), channel)
+    return messages, stats

bot/services/formatter.py

@@ -0,0 +1,66 @@
+import re
+
+MAX_MSG_LEN = 4000
+
+
+def _escape_md2(text: str) -> str:
+    """Escape special chars for MarkdownV2, preserving bold/headers/bullets."""
+    # We use HTML instead — MarkdownV2 escaping is too fragile
+    raise NotImplementedError("Use HTML mode")
+
+
+def _md_to_html(text: str) -> str:
+    """Minimal Markdown-to-HTML for Telegram's supported subset."""
+    # Bold: **text** or __text__
+    text = re.sub(r"\*\*(.+?)\*\*", r"<b>\1</b>", text)
+    text = re.sub(r"__(.+?)__", r"<b>\1</b>", text)
+    # Italic: *text* (but not inside bold tags)
+    text = re.sub(r"(?<!</b>)\*(.+?)\*(?!<)", r"<i>\1</i>", text)
+    # Inline code
+    text = re.sub(r"`(.+?)`", r"<code>\1</code>", text)
+    # Headers: ## Title -> bold
+    text = re.sub(r"^#{1,3}\s+(.+)$", r"<b>\1</b>", text, flags=re.MULTILINE)
+    # Escape HTML entities in remaining text (but preserve our tags)
+    # Already handled — Telegram is lenient with unescaped text in HTML mode
+    return text
+
+
+def split_report(report: str) -> list[str]:
+    """Split a report into Telegram-safe HTML messages."""
+    html = _md_to_html(report)
+
+    if len(html) <= MAX_MSG_LEN:
+        return [html]
+
+    # Split on section boundaries (bold headers on their own line)
+    sections = re.split(r"(?=\n<b>[^<]+</b>\n)", html)
+
+    messages: list[str] = []
+    current = ""
+
+    for section in sections:
+        if not section.strip():
+            continue
+        if len(current) + len(section) > MAX_MSG_LEN:
+            if current.strip():
+                messages.append(current.strip())
+            # If a single section is too long, split on paragraphs
+            if len(section) > MAX_MSG_LEN:
+                paragraphs = section.split("\n\n")
+                current = ""
+                for para in paragraphs:
+                    if len(current) + len(para) + 2 > MAX_MSG_LEN:
+                        if current.strip():
+                            messages.append(current.strip())
+                        current = para + "\n\n"
+                    else:
+                        current += para + "\n\n"
+            else:
+                current = section
+        else:
+            current += section
+
+    if current.strip():
+        messages.append(current.strip())
+
+    return messages if messages else [html[:MAX_MSG_LEN]]