diff --git a/.env.example b/.env.example index c8750ca..7292ce3 100644 --- a/.env.example +++ b/.env.example @@ -4,3 +4,11 @@ TELEGRAM_API_HASH= TELEGRAM_PHONE= ANTHROPIC_API_KEY= CLAUDE_MODEL=claude-opus-4-6 +OPENROUTER_API_KEY= +AVAILABLE_MODELS=claude-haiku-4-5,anthropic/claude-3.5-sonnet +PRICE_BASIC=50 +PRICE_STANDARD=100 +PRICE_FULL=200 +FREE_ANALYSES=1 +DB_PATH=/app/data/bot.db +REPORTS_DIR=/app/data/reports diff --git a/CLAUDE.md b/CLAUDE.md index b583997..7295f22 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -4,40 +4,55 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co ## Overview -Telegram Channel Analyzer Bot — fetches public channel history via Telethon, analyzes with Claude AI (chunked summarization pipeline), delivers reports via aiogram bot. +Telegram Channel Analyzer Bot — multi-user public bot with payments, multi-language (EN/RU), multi-provider AI (Anthropic + OpenRouter). Fetches public channel history via Telethon, analyzes with configurable AI models (chunked summarization pipeline), delivers HTML reports via aiogram bot. ## Stack - **Python 3.12**, async throughout -- **aiogram 3** — Telegram bot interface (commands, inline keyboards, progress messages) +- **aiogram 3** — Telegram bot interface (commands, inline keyboards, payments, progress messages) - **Telethon** — userbot client for reading public channel history - **anthropic** (AsyncAnthropic) — Claude API with streaming +- **httpx** — OpenRouter API calls (transitive dep of anthropic) +- **aiosqlite** — SQLite database for users, usage, payments - **pydantic-settings** — config from environment variables ## Architecture - Both aiogram and Telethon share one asyncio loop (no threads) - Telethon client is attached to the bot instance in `__main__.py` -- Analysis pipeline: Fetch → Chunk (token-bounded) → Summarize each chunk → Synthesize final report +- Analysis flow: Channel → Depth selection → Focus areas (multi-select) → Model → Payment check → Pipeline +- Pipeline: Fetch → Chunk (token-bounded) → Summarize each chunk → Synthesize final report → Save .md → Send - Rate limit handling: semaphore(1), 60s cooldown between chunks, retry with backoff on 429 - Output: Markdown→HTML conversion, split on section boundaries at 4000 chars +- User middleware auto-creates DB user, detects language, injects `lang`/`db_user` into handler data +- In-memory `_sessions` dict tracks multi-step analysis flow per user ## Key Files -- `bot/config.py` — all settings from env vars, `CLAUDE_MODEL` selects the model -- `bot/services/analyzer.py` — Claude API calls, adaptive thinking only for opus-4-6 -- `bot/services/chunker.py` — `MAX_TOKENS_PER_CHUNK` and `CHARS_PER_TOKEN` control chunking -- `bot/prompts/` — prompt templates per report type (chunk_summary.py, synthesis.py) +- `bot/config.py` — all settings from env vars (Anthropic, OpenRouter, pricing, paths) +- `bot/models.py` — `Depth`, `FocusArea` enums, `AnalysisSession` dataclass +- `bot/i18n/` — `Lang` enum, `t()` lookup, all UI strings in `strings.py` +- `bot/db/` — aiosqlite engine, `user_repo`, `usage_repo` +- `bot/middleware/user_middleware.py` — auto-create user, detect lang +- `bot/services/ai_client.py` — `AIClient` ABC, `AnthropicClient`, `OpenRouterClient`, `get_ai_client()` +- `bot/services/analyzer.py` — orchestrates chunk analysis + synthesis with retry +- `bot/services/report_saver.py` — saves .md to `data/reports/` +- `bot/prompts/` — composable prompts: `depth.py`, `focus_areas.py`, `chunk_summary.py`, `synthesis.py` +- `bot/handlers/analyze.py` — multi-step flow (depth→focus→model→pay→run) +- `bot/handlers/payment.py` — pre_checkout handler ## Running - Container-based: `Containerfile` + `compose.yml` - `--login` flag for interactive Telethon session creation -- Session persists in `data/` volume +- Session persists in `data/` volume, DB at `data/bot.db`, reports at `data/reports/` - `.env` file must not have inline comments (Podman/Docker limitation) ## Common Tasks - To change chunk size: edit `MAX_TOKENS_PER_CHUNK` in `bot/services/chunker.py` -- To add a report type: add to `ReportType` enum, add prompts in both `prompts/` files -- To change model: set `CLAUDE_MODEL` env var; thinking params auto-adapt in `analyzer.py` +- To add a focus area: add to `FocusArea` enum, add prompt fragments in `prompts/focus_areas.py`, add i18n strings +- To add an AI model: add to `AVAILABLE_MODELS` env var (use `org/model` format for OpenRouter) +- To change pricing: set `PRICE_BASIC`/`PRICE_STANDARD`/`PRICE_FULL` env vars +- To change free trial count: set `FREE_ANALYSES` env var +- To add a language: add to `Lang` enum, add translations in `i18n/strings.py` diff --git a/bot/__main__.py b/bot/__main__.py index e4345cb..4bc63e5 100644 --- a/bot/__main__.py +++ b/bot/__main__.py @@ -6,7 +6,9 @@ from aiogram import Bot, Dispatcher from telethon import TelegramClient from bot.config import settings -from bot.handlers import analyze, start +from bot.db.engine import get_db, close_db +from bot.handlers import analyze, features, lang, payment, prices, start +from bot.middleware.user_middleware import UserMiddleware SESSION_PATH = "/app/data/analyzer_session" @@ -30,6 +32,10 @@ async def login() -> None: async def main() -> None: + # Init database + await get_db() + log.info("Database initialized") + telethon_client = TelegramClient( SESSION_PATH, settings.telegram_api_id, @@ -46,11 +52,19 @@ async def main() -> None: log.info("Telethon client started") bot = Bot(token=settings.bot_token) - # Attach telethon client to bot instance for handler access bot._telethon_client = telethon_client # type: ignore[attr-defined] dp = Dispatcher() + + # Register middleware + dp.update.middleware(UserMiddleware()) + + # Register routers — payment.pre_checkout must come before analyze dp.include_router(start.router) + dp.include_router(lang.router) + dp.include_router(features.router) + dp.include_router(prices.router) + dp.include_router(payment.router) dp.include_router(analyze.router) log.info("Starting bot polling...") @@ -58,6 +72,7 @@ async def main() -> None: await dp.start_polling(bot) finally: await telethon_client.disconnect() + await close_db() if __name__ == "__main__": diff --git a/bot/config.py b/bot/config.py index 5e0d528..eebab08 100644 --- a/bot/config.py +++ b/bot/config.py @@ -9,7 +9,22 @@ class Settings(BaseSettings): anthropic_api_key: str claude_model: str = "claude-opus-4-6" + openrouter_api_key: str = "" + available_models: str = "claude-haiku-4-5" + + price_basic: int = 50 + price_standard: int = 100 + price_full: int = 200 + free_analyses: int = 1 + + db_path: str = "/app/data/bot.db" + reports_dir: str = "/app/data/reports" + model_config = {"env_file": ".env", "env_file_encoding": "utf-8", "extra": "ignore"} + @property + def models_list(self) -> list[str]: + return [m.strip() for m in self.available_models.split(",") if m.strip()] + settings = Settings() diff --git a/bot/db/__init__.py b/bot/db/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/bot/db/engine.py b/bot/db/engine.py new file mode 100644 index 0000000..799f254 --- /dev/null +++ b/bot/db/engine.py @@ -0,0 +1,54 @@ +import aiosqlite + +from bot.config import settings + +_db: aiosqlite.Connection | None = None + +SCHEMA = """\ +CREATE TABLE IF NOT EXISTS users ( + telegram_id INTEGER PRIMARY KEY, + username TEXT, + first_name TEXT, + lang TEXT NOT NULL DEFAULT 'en', + free_used INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL DEFAULT (datetime('now')) +); + +CREATE TABLE IF NOT EXISTS analyses ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + telegram_id INTEGER NOT NULL REFERENCES users(telegram_id), + channel TEXT NOT NULL, + depth TEXT NOT NULL, + focus_areas TEXT NOT NULL, + model_id TEXT NOT NULL, + stars_paid INTEGER NOT NULL DEFAULT 0, + report_path TEXT, + created_at TEXT NOT NULL DEFAULT (datetime('now')) +); + +CREATE TABLE IF NOT EXISTS payments ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + telegram_id INTEGER NOT NULL REFERENCES users(telegram_id), + telegram_payment_id TEXT NOT NULL UNIQUE, + stars_amount INTEGER NOT NULL, + analysis_id INTEGER REFERENCES analyses(id), + created_at TEXT NOT NULL DEFAULT (datetime('now')) +); +""" + + +async def get_db() -> aiosqlite.Connection: + global _db + if _db is None: + _db = await aiosqlite.connect(settings.db_path) + _db.row_factory = aiosqlite.Row + await _db.executescript(SCHEMA) + await _db.commit() + return _db + + +async def close_db() -> None: + global _db + if _db is not None: + await _db.close() + _db = None diff --git a/bot/db/usage_repo.py b/bot/db/usage_repo.py new file mode 100644 index 0000000..5d7800d --- /dev/null +++ b/bot/db/usage_repo.py @@ -0,0 +1,46 @@ +import json + +from bot.db.engine import get_db + + +async def record_analysis( + telegram_id: int, + channel: str, + depth: str, + focus_areas: list[str], + model_id: str, + stars_paid: int = 0, + report_path: str | None = None, +) -> int: + db = await get_db() + cursor = await db.execute( + "INSERT INTO analyses (telegram_id, channel, depth, focus_areas, model_id, stars_paid, report_path) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + (telegram_id, channel, depth, json.dumps(focus_areas), model_id, stars_paid, report_path), + ) + await db.commit() + return cursor.lastrowid + + +async def update_report_path(analysis_id: int, report_path: str) -> None: + db = await get_db() + await db.execute( + "UPDATE analyses SET report_path = ? WHERE id = ?", (report_path, analysis_id) + ) + await db.commit() + + +async def record_payment( + telegram_id: int, + telegram_payment_id: str, + stars_amount: int, + analysis_id: int | None = None, +) -> int: + db = await get_db() + cursor = await db.execute( + "INSERT INTO payments (telegram_id, telegram_payment_id, stars_amount, analysis_id) " + "VALUES (?, ?, ?, ?)", + (telegram_id, telegram_payment_id, stars_amount, analysis_id), + ) + await db.commit() + return cursor.lastrowid diff --git a/bot/db/user_repo.py b/bot/db/user_repo.py new file mode 100644 index 0000000..3820704 --- /dev/null +++ b/bot/db/user_repo.py @@ -0,0 +1,52 @@ +from bot.db.engine import get_db + + +async def get_or_create( + telegram_id: int, + username: str | None = None, + first_name: str | None = None, + lang: str = "en", +) -> dict: + db = await get_db() + row = await db.execute_fetchall( + "SELECT * FROM users WHERE telegram_id = ?", (telegram_id,) + ) + if row: + return dict(row[0]) + + await db.execute( + "INSERT INTO users (telegram_id, username, first_name, lang) VALUES (?, ?, ?, ?)", + (telegram_id, username, first_name, lang), + ) + await db.commit() + row = await db.execute_fetchall( + "SELECT * FROM users WHERE telegram_id = ?", (telegram_id,) + ) + return dict(row[0]) + + +async def update_lang(telegram_id: int, lang: str) -> None: + db = await get_db() + await db.execute("UPDATE users SET lang = ? WHERE telegram_id = ?", (lang, telegram_id)) + await db.commit() + + +async def increment_free(telegram_id: int) -> int: + db = await get_db() + await db.execute( + "UPDATE users SET free_used = free_used + 1 WHERE telegram_id = ?", + (telegram_id,), + ) + await db.commit() + row = await db.execute_fetchall( + "SELECT free_used FROM users WHERE telegram_id = ?", (telegram_id,) + ) + return row[0][0] + + +async def get_free_used(telegram_id: int) -> int: + db = await get_db() + row = await db.execute_fetchall( + "SELECT free_used FROM users WHERE telegram_id = ?", (telegram_id,) + ) + return row[0][0] if row else 0 diff --git a/bot/handlers/analyze.py b/bot/handlers/analyze.py index c51985c..a9c91ca 100644 --- a/bot/handlers/analyze.py +++ b/bot/handlers/analyze.py @@ -3,7 +3,14 @@ import re from aiogram import F, Router from aiogram.filters import Command -from aiogram.types import CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup, Message +from aiogram.types import ( + CallbackQuery, + FSInputFile, + InlineKeyboardButton, + InlineKeyboardMarkup, + LabeledPrice, + Message, +) from telethon import TelegramClient from telethon.errors import ( ChannelInvalidError, @@ -13,132 +20,273 @@ from telethon.errors import ( UsernameNotOccupiedError, ) -from bot.models import ReportType +from bot.config import settings +from bot.db import usage_repo, user_repo +from bot.i18n import Lang, t +from bot.models import AnalysisSession, Depth, FocusArea from bot.services.analyzer import analyze_channel from bot.services.chunker import chunk_messages from bot.services.fetcher import fetch_channel_messages from bot.services.formatter import split_report +from bot.services.report_saver import save_report log = logging.getLogger(__name__) router = Router() -# channel_username -> store temporarily per user for callback -_pending: dict[int, str] = {} +_sessions: dict[int, AnalysisSession] = {} def _extract_channel(text: str) -> str | None: text = text.strip() - # @username m = re.match(r"@(\w+)", text) if m: return m.group(1) - # https://t.me/username m = re.match(r"https?://t\.me/(\w+)", text) if m: return m.group(1) - # bare username if re.match(r"^\w+$", text): return text return None -def _report_keyboard() -> InlineKeyboardMarkup: - return InlineKeyboardMarkup( - inline_keyboard=[ - [InlineKeyboardButton(text=rt.label, callback_data=f"report:{rt.value}")] - for rt in ReportType - ] - ) +def _depth_keyboard(lang: Lang) -> InlineKeyboardMarkup: + return InlineKeyboardMarkup(inline_keyboard=[ + [InlineKeyboardButton(text=t("depth_basic", lang), callback_data="depth:basic")], + [InlineKeyboardButton(text=t("depth_standard", lang), callback_data="depth:standard")], + [InlineKeyboardButton(text=t("depth_full", lang), callback_data="depth:full")], + ]) +def _focus_keyboard(lang: Lang, selected: set[str]) -> InlineKeyboardMarkup: + rows = [] + for area in FocusArea: + check = "✅" if area.value in selected else "☐" + rows.append([InlineKeyboardButton( + text=f"{check} {t(f'focus_{area.value}', lang)}", + callback_data=f"focus:{area.value}", + )]) + rows.append([InlineKeyboardButton(text=f"✅ {t('done', lang)}", callback_data="focus:done")]) + return InlineKeyboardMarkup(inline_keyboard=rows) + + +def _model_keyboard() -> InlineKeyboardMarkup: + rows = [] + for model_id in settings.models_list: + label = model_id.split("/")[-1] if "/" in model_id else model_id + rows.append([InlineKeyboardButton(text=label, callback_data=f"model:{model_id}")]) + return InlineKeyboardMarkup(inline_keyboard=rows) + + +# Step 0: /analyze @channel @router.message(Command("analyze")) -async def cmd_analyze(message: Message) -> None: +async def cmd_analyze(message: Message, lang: Lang = Lang.EN, **_: object) -> None: args = (message.text or "").split(maxsplit=1) if len(args) < 2: - await message.answer( - "Please provide a channel: /analyze @channel", - parse_mode="HTML", - ) + await message.answer(t("provide_channel", lang), parse_mode="HTML") return channel = _extract_channel(args[1]) if not channel: - await message.answer("Could not parse channel name. Use @username or t.me/username.") + await message.answer(t("bad_channel", lang), parse_mode="HTML") return - _pending[message.from_user.id] = channel + _sessions[message.from_user.id] = AnalysisSession(channel=channel) await message.answer( - f"Channel: @{channel}\n\nChoose report type:", + t("choose_depth", lang, channel=channel), parse_mode="HTML", - reply_markup=_report_keyboard(), + reply_markup=_depth_keyboard(lang), ) -@router.callback_query(F.data.startswith("report:")) -async def on_report_type(callback: CallbackQuery) -> None: +# Step 1: Depth selected +@router.callback_query(F.data.startswith("depth:")) +async def on_depth(callback: CallbackQuery, lang: Lang = Lang.EN, **_: object) -> None: await callback.answer() - user_id = callback.from_user.id - channel = _pending.pop(user_id, None) - if not channel: - await callback.message.answer("Session expired. Please run /analyze again.") + session = _sessions.get(user_id) + if not session: + await callback.message.answer(t("session_expired", lang)) return - report_value = callback.data.split(":", 1)[1] - report_type = ReportType(report_value) + depth_val = callback.data.split(":", 1)[1] + session.depth = Depth(depth_val) - telethon_client: TelegramClient = callback.message.bot.__dict__.get("_telethon_client") + await callback.message.edit_text( + t("choose_focus", lang), + parse_mode="HTML", + reply_markup=_focus_keyboard(lang, set()), + ) + + +# Step 2: Focus area toggle +@router.callback_query(F.data.startswith("focus:")) +async def on_focus(callback: CallbackQuery, lang: Lang = Lang.EN, **_: object) -> None: + user_id = callback.from_user.id + session = _sessions.get(user_id) + if not session: + await callback.answer() + await callback.message.answer(t("session_expired", lang)) + return + + value = callback.data.split(":", 1)[1] + + if value == "done": + if not session.focus_areas: + await callback.answer(t("no_focus_selected", lang), show_alert=True) + return + await callback.answer() + # Show model selection + if len(settings.models_list) == 1: + # Skip model selection if only one available + session.model_id = settings.models_list[0] + await _check_payment_and_run(callback.message, user_id, lang) + else: + await callback.message.edit_text( + t("choose_model", lang), + parse_mode="HTML", + reply_markup=_model_keyboard(), + ) + return + + await callback.answer() + area = FocusArea(value) + if area in session.focus_areas: + session.focus_areas.remove(area) + else: + session.focus_areas.append(area) + + selected = {a.value for a in session.focus_areas} + await callback.message.edit_reply_markup( + reply_markup=_focus_keyboard(lang, selected), + ) + + +# Step 3: Model selected +@router.callback_query(F.data.startswith("model:")) +async def on_model(callback: CallbackQuery, lang: Lang = Lang.EN, **_: object) -> None: + await callback.answer() + user_id = callback.from_user.id + session = _sessions.get(user_id) + if not session: + await callback.message.answer(t("session_expired", lang)) + return + + session.model_id = callback.data.split(":", 1)[1] + await _check_payment_and_run(callback.message, user_id, lang) + + +# Step 4: Payment check + run +async def _check_payment_and_run(message: Message, user_id: int, lang: Lang) -> None: + session = _sessions.get(user_id) + if not session: + return + + free_used = await user_repo.get_free_used(user_id) + price_map = { + Depth.BASIC: settings.price_basic, + Depth.STANDARD: settings.price_standard, + Depth.FULL: settings.price_full, + } + price = price_map[session.depth] + + if free_used < settings.free_analyses: + await user_repo.increment_free(user_id) + await message.edit_text( + t("free_analysis", lang, used=free_used + 1, max=settings.free_analyses), + parse_mode="HTML", + ) + await _run_analysis(message, user_id, lang, stars_paid=0) + else: + # Record analysis first to get ID for payload + analysis_id = await usage_repo.record_analysis( + telegram_id=user_id, + channel=session.channel, + depth=session.depth.value, + focus_areas=[a.value for a in session.focus_areas], + model_id=session.model_id, + stars_paid=price, + ) + await message.answer_invoice( + title=t("invoice_title", lang, depth=session.depth.label), + description=t("invoice_description", lang, channel=session.channel, depth=session.depth.label), + payload=str(analysis_id), + currency="XTR", + prices=[LabeledPrice(label="Analysis", amount=price)], + ) + + +# Payment callback triggers analysis +@router.message(lambda m: m.successful_payment is not None) +async def on_payment_run(message: Message, lang: Lang = Lang.EN, **_: object) -> None: + user_id = message.from_user.id + session = _sessions.get(user_id) + if not session: + return + payment = message.successful_payment + await usage_repo.record_payment( + telegram_id=user_id, + telegram_payment_id=payment.telegram_payment_charge_id, + stars_amount=payment.total_amount, + analysis_id=int(payment.invoice_payload) if payment.invoice_payload.isdigit() else None, + ) + await _run_analysis(message, user_id, lang, stars_paid=payment.total_amount) + + +# Step 5: Run pipeline +async def _run_analysis(message: Message, user_id: int, lang: Lang, stars_paid: int) -> None: + session = _sessions.pop(user_id, None) + if not session: + return + + telethon_client: TelegramClient | None = message.bot.__dict__.get("_telethon_client") if not telethon_client: - # Fallback: try dispatcher data - from aiogram import Dispatcher - # Access via bot's dispatcher isn't directly available in callback, - # so we store it on the bot instance in __main__.py - await callback.message.answer("Internal error: Telethon client not configured.") + await message.answer(t("internal_error", lang)) return - status_msg = await callback.message.answer( - f"Starting {report_type.label} for @{channel}...\n\n" - "Fetching messages...", + status_msg = await message.answer( + t("analysis_starting", lang, channel=session.channel, status=t("fetching", lang)), parse_mode="HTML", ) async def update_status(text: str) -> None: try: await status_msg.edit_text( - f"{report_type.label} for @{channel}\n\n{text}", + t("analysis_starting", lang, channel=session.channel, status=text), parse_mode="HTML", ) except Exception: pass try: - messages, stats = await fetch_channel_messages(telethon_client, channel) + messages, stats = await fetch_channel_messages(telethon_client, session.channel) except (ChannelPrivateError, ChannelInvalidError): - await status_msg.edit_text("Channel is private or does not exist.") + await status_msg.edit_text(t("channel_private", lang)) return except (UsernameInvalidError, UsernameNotOccupiedError): - await status_msg.edit_text("Channel username not found.") + await status_msg.edit_text(t("channel_not_found", lang)) return except FloodWaitError as e: - await status_msg.edit_text(f"Rate limited by Telegram. Retry in {e.seconds}s.") + await status_msg.edit_text(t("flood_wait", lang, s=e.seconds)) return except Exception as e: - log.exception("Failed to fetch channel %s", channel) - await status_msg.edit_text(f"Failed to fetch channel: {e}") + log.exception("Failed to fetch channel %s", session.channel) + await status_msg.edit_text(t("fetch_failed", lang, e=e)) return if not messages: - await status_msg.edit_text("No text messages found in this channel.") + await status_msg.edit_text(t("no_messages", lang)) return - await update_status(f"Fetched {len(messages)} messages. Chunking...") - + await update_status(t("fetched_n", lang, n=len(messages))) chunks = chunk_messages(messages) - await update_status(f"{len(messages)} messages in {len(chunks)} chunks. Analyzing...") + await update_status(t("chunked", lang, n=len(messages), c=len(chunks))) try: report = await analyze_channel( - report_type=report_type, + depth=session.depth, + focus_areas=session.focus_areas, + lang=lang.value, + model_id=session.model_id, chunks=chunks, channel_title=stats["title"], channel_username=stats.get("username"), @@ -147,21 +295,42 @@ async def on_report_type(callback: CallbackQuery) -> None: on_progress=update_status, ) except Exception as e: - log.exception("Analysis failed for %s", channel) - await status_msg.edit_text(f"Analysis failed: {e}") + log.exception("Analysis failed for %s", session.channel) + await status_msg.edit_text(t("analysis_failed", lang, e=e)) return - await update_status("Sending report...") + # Save report + report_path = await save_report(user_id, session.channel, report) + + # Record in DB + await usage_repo.record_analysis( + telegram_id=user_id, + channel=session.channel, + depth=session.depth.value, + focus_areas=[a.value for a in session.focus_areas], + model_id=session.model_id, + stars_paid=stars_paid, + report_path=report_path, + ) + + await update_status(t("sending_report", lang)) parts = split_report(report) for part in parts: try: - await callback.message.answer(part, parse_mode="HTML") + await message.answer(part, parse_mode="HTML") except Exception: - # Fallback: send without formatting - await callback.message.answer(part) + await message.answer(part) + + # Offer file download + await message.answer_document( + FSInputFile(report_path, filename=f"{session.channel}_report.md"), + caption=t("download_report", lang), + ) try: await status_msg.delete() except Exception: pass + + diff --git a/bot/handlers/features.py b/bot/handlers/features.py new file mode 100644 index 0000000..13340d5 --- /dev/null +++ b/bot/handlers/features.py @@ -0,0 +1,16 @@ +from aiogram import Router +from aiogram.filters import Command +from aiogram.types import Message + +from bot.i18n import Lang, t +from bot.models import FocusArea + +router = Router() + + +@router.message(Command("features")) +async def cmd_features(message: Message, lang: Lang = Lang.EN, **_: object) -> None: + lines = [t("features_title", lang)] + for area in FocusArea: + lines.append(f"• {t(f'features_{area.value}', lang)}") + await message.answer("\n".join(lines), parse_mode="HTML") diff --git a/bot/handlers/lang.py b/bot/handlers/lang.py new file mode 100644 index 0000000..be04aa4 --- /dev/null +++ b/bot/handlers/lang.py @@ -0,0 +1,20 @@ +from aiogram import Router +from aiogram.filters import Command +from aiogram.types import Message + +from bot.db import user_repo +from bot.i18n import Lang, t + +router = Router() + + +@router.message(Command("lang")) +async def cmd_lang(message: Message, lang: Lang = Lang.EN, **_: object) -> None: + args = (message.text or "").split(maxsplit=1) + if len(args) < 2 or args[1].strip().lower() not in ("en", "ru"): + await message.answer(t("lang_usage", lang), parse_mode="HTML") + return + + new_lang = args[1].strip().lower() + await user_repo.update_lang(message.from_user.id, new_lang) + await message.answer(t("lang_set", new_lang), parse_mode="HTML") diff --git a/bot/handlers/payment.py b/bot/handlers/payment.py new file mode 100644 index 0000000..8d898b4 --- /dev/null +++ b/bot/handlers/payment.py @@ -0,0 +1,9 @@ +from aiogram import Router +from aiogram.types import PreCheckoutQuery + +router = Router() + + +@router.pre_checkout_query() +async def on_pre_checkout(query: PreCheckoutQuery, **_: object) -> None: + await query.answer(ok=True) diff --git a/bot/handlers/prices.py b/bot/handlers/prices.py new file mode 100644 index 0000000..b92a372 --- /dev/null +++ b/bot/handlers/prices.py @@ -0,0 +1,19 @@ +from aiogram import Router +from aiogram.filters import Command +from aiogram.types import Message + +from bot.config import settings +from bot.i18n import Lang, t + +router = Router() + + +@router.message(Command("prices")) +async def cmd_prices(message: Message, lang: Lang = Lang.EN, **_: object) -> None: + text = t( + "prices", lang, + basic=settings.price_basic, + standard=settings.price_standard, + full=settings.price_full, + ) + await message.answer(text, parse_mode="HTML") diff --git a/bot/handlers/start.py b/bot/handlers/start.py index 1ab704b..c4f5697 100644 --- a/bot/handlers/start.py +++ b/bot/handlers/start.py @@ -2,21 +2,11 @@ from aiogram import Router from aiogram.filters import Command from aiogram.types import Message -router = Router() +from bot.i18n import Lang, t -HELP_TEXT = ( - "Telegram Channel Analyzer\n\n" - "Analyze any public Telegram channel using AI.\n\n" - "Usage:\n" - "/analyze @channel — Start analysis\n" - "/analyze https://t.me/channel — Also works\n\n" - "You'll choose a report type:\n" - "• Content Analysis — topics, tone, themes\n" - "• Content + Stats — above + engagement data\n" - "• Full Audit — comprehensive review with recommendations" -) +router = Router() @router.message(Command("start", "help")) -async def cmd_start(message: Message) -> None: - await message.answer(HELP_TEXT, parse_mode="HTML") +async def cmd_start(message: Message, lang: Lang = Lang.EN, **_: object) -> None: + await message.answer(t("welcome", lang), parse_mode="HTML") diff --git a/bot/i18n/__init__.py b/bot/i18n/__init__.py new file mode 100644 index 0000000..664a058 --- /dev/null +++ b/bot/i18n/__init__.py @@ -0,0 +1,30 @@ +from enum import Enum + +from bot.i18n.strings import STRINGS + + +class Lang(Enum): + EN = "en" + RU = "ru" + + +def t(key: str, lang: Lang | str = Lang.EN, **kwargs: object) -> str: + if isinstance(lang, Lang): + lang_str = lang.value + elif lang in ("en", "ru"): + lang_str = lang + else: + lang_str = "en" + entry = STRINGS.get(key) + if not entry: + return key + text = entry.get(lang_str, entry.get("en", key)) + if kwargs: + text = text.format(**kwargs) + return text + + +def detect_lang(language_code: str | None) -> Lang: + if language_code and language_code.startswith("ru"): + return Lang.RU + return Lang.EN diff --git a/bot/i18n/strings.py b/bot/i18n/strings.py new file mode 100644 index 0000000..399534e --- /dev/null +++ b/bot/i18n/strings.py @@ -0,0 +1,255 @@ +STRINGS: dict[str, dict[str, str]] = { + # /start, /help + "welcome": { + "en": ( + "Telegram Channel Analyzer\n\n" + "Analyze any public Telegram channel using AI.\n\n" + "Commands:\n" + "/analyze @channel — Start analysis\n" + "/lang en|ru — Change language\n" + "/features — Available focus areas\n" + "/prices — Pricing info\n" + ), + "ru": ( + "Анализатор Telegram-каналов\n\n" + "Анализ любого публичного канала с помощью ИИ.\n\n" + "Команды:\n" + "/analyze @channel — Начать анализ\n" + "/lang en|ru — Сменить язык\n" + "/features — Доступные области анализа\n" + "/prices — Информация о ценах\n" + ), + }, + + # /analyze + "provide_channel": { + "en": "Please provide a channel: /analyze @channel", + "ru": "Укажите канал: /analyze @канал", + }, + "bad_channel": { + "en": "Could not parse channel name. Use @username or t.me/username.", + "ru": "Не удалось разобрать имя канала. Используйте @username или t.me/username.", + }, + "choose_depth": { + "en": "Channel: @{channel}\n\nChoose analysis depth:", + "ru": "Канал: @{channel}\n\nВыберите глубину анализа:", + }, + "choose_focus": { + "en": "Select focus areas (tap to toggle, then Done):", + "ru": "Выберите области анализа (нажмите для выбора, затем Готово):", + }, + "choose_model": { + "en": "Choose AI model:", + "ru": "Выберите модель ИИ:", + }, + "session_expired": { + "en": "Session expired. Please run /analyze again.", + "ru": "Сессия истекла. Запустите /analyze снова.", + }, + "no_focus_selected": { + "en": "Please select at least one focus area.", + "ru": "Выберите хотя бы одну область анализа.", + }, + + # Depth labels + "depth_basic": { + "en": "Basic (fast)", + "ru": "Базовый (быстрый)", + }, + "depth_standard": { + "en": "Standard", + "ru": "Стандартный", + }, + "depth_full": { + "en": "Full (detailed)", + "ru": "Полный (детальный)", + }, + + # Focus area labels + "focus_psychology": { + "en": "Psychology & Influence", + "ru": "Психология и влияние", + }, + "focus_business": { + "en": "Business & Monetization", + "ru": "Бизнес и монетизация", + }, + "focus_marketing": { + "en": "Marketing & Growth", + "ru": "Маркетинг и рост", + }, + "focus_content": { + "en": "Content Strategy", + "ru": "Контент-стратегия", + }, + "focus_audience": { + "en": "Audience & Engagement", + "ru": "Аудитория и вовлечённость", + }, + "focus_sentiment": { + "en": "Sentiment Analysis", + "ru": "Анализ тональности", + }, + + # Analysis progress + "fetching": { + "en": "Fetching messages...", + "ru": "Загрузка сообщений...", + }, + "fetched_n": { + "en": "Fetched {n} messages. Chunking...", + "ru": "Загружено {n} сообщений. Разбиение...", + }, + "chunked": { + "en": "{n} messages in {c} chunks. Analyzing...", + "ru": "{n} сообщений в {c} частях. Анализ...", + }, + "analyzing_chunk": { + "en": "Analyzing chunk {i}/{total}...", + "ru": "Анализ части {i}/{total}...", + }, + "chunk_done_cooldown": { + "en": "Chunk {i}/{total} done. Cooling down 60s...", + "ru": "Часть {i}/{total} готова. Пауза 60с...", + }, + "generating_report": { + "en": "Generating final report...", + "ru": "Генерация финального отчёта...", + }, + "sending_report": { + "en": "Sending report...", + "ru": "Отправка отчёта...", + }, + "analysis_starting": { + "en": "Starting analysis for @{channel}...\n\n{status}", + "ru": "Начинаю анализ @{channel}...\n\n{status}", + }, + + # Errors + "channel_private": { + "en": "Channel is private or does not exist.", + "ru": "Канал приватный или не существует.", + }, + "channel_not_found": { + "en": "Channel username not found.", + "ru": "Имя канала не найдено.", + }, + "flood_wait": { + "en": "Rate limited by Telegram. Retry in {s}s.", + "ru": "Ограничение Telegram. Повторите через {s}с.", + }, + "fetch_failed": { + "en": "Failed to fetch channel: {e}", + "ru": "Ошибка получения канала: {e}", + }, + "no_messages": { + "en": "No text messages found in this channel.", + "ru": "В канале не найдено текстовых сообщений.", + }, + "analysis_failed": { + "en": "Analysis failed: {e}", + "ru": "Ошибка анализа: {e}", + }, + "internal_error": { + "en": "Internal error: Telethon client not configured.", + "ru": "Внутренняя ошибка: клиент Telethon не настроен.", + }, + "rate_limited": { + "en": "Rate limited, waiting {s}s...", + "ru": "Лимит запросов, ожидание {s}с...", + }, + + # Payment + "free_analysis": { + "en": "This analysis is free ({used}/{max} free trial).", + "ru": "Этот анализ бесплатный ({used}/{max} пробный).", + }, + "payment_required": { + "en": "This analysis costs {price} Stars.", + "ru": "Этот анализ стоит {price} Stars.", + }, + "invoice_title": { + "en": "Channel Analysis — {depth}", + "ru": "Анализ канала — {depth}", + }, + "invoice_description": { + "en": "Analysis of @{channel} ({depth} depth)", + "ru": "Анализ @{channel} ({depth})", + }, + "payment_success": { + "en": "Payment received! Starting analysis...", + "ru": "Оплата получена! Начинаю анализ...", + }, + + # /lang + "lang_set": { + "en": "Language set to English.", + "ru": "Язык установлен: Русский.", + }, + "lang_usage": { + "en": "Usage: /lang en or /lang ru", + "ru": "Использование: /lang en или /lang ru", + }, + + # /features + "features_title": { + "en": "Available Focus Areas\n", + "ru": "Доступные области анализа\n", + }, + "features_psychology": { + "en": "Psychology & Influence — Persuasion techniques, cognitive biases, emotional triggers", + "ru": "Психология и влияние — Техники убеждения, когнитивные искажения, эмоциональные триггеры", + }, + "features_business": { + "en": "Business & Monetization — Revenue models, product placement, conversion patterns", + "ru": "Бизнес и монетизация — Модели доходов, продвижение продуктов, воронки конверсии", + }, + "features_marketing": { + "en": "Marketing & Growth — Growth tactics, viral mechanics, audience acquisition", + "ru": "Маркетинг и рост — Тактики роста, вирусные механики, привлечение аудитории", + }, + "features_content": { + "en": "Content Strategy — Topics, formats, posting patterns, narrative arcs", + "ru": "Контент-стратегия — Темы, форматы, паттерны публикаций, нарративы", + }, + "features_audience": { + "en": "Audience & Engagement — Interaction patterns, community dynamics, engagement drivers", + "ru": "Аудитория и вовлечённость — Паттерны взаимодействия, динамика сообщества", + }, + "features_sentiment": { + "en": "Sentiment Analysis — Tone distribution, emotional shifts, sentiment by topic", + "ru": "Анализ тональности — Распределение тона, эмоциональные сдвиги, тональность по темам", + }, + + # /prices + "prices": { + "en": ( + "Pricing\n\n" + "First analysis is free!\n\n" + "• Basic — {basic} Stars\n" + "• Standard — {standard} Stars\n" + "• Full — {full} Stars\n\n" + "Payment via Telegram Stars." + ), + "ru": ( + "Цены\n\n" + "Первый анализ бесплатно!\n\n" + "• Базовый — {basic} Stars\n" + "• Стандартный — {standard} Stars\n" + "• Полный — {full} Stars\n\n" + "Оплата через Telegram Stars." + ), + }, + + # Report download + "download_report": { + "en": "Download report", + "ru": "Скачать отчёт", + }, + + # Done button + "done": { + "en": "Done", + "ru": "Готово", + }, +} diff --git a/bot/middleware/__init__.py b/bot/middleware/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/bot/middleware/user_middleware.py b/bot/middleware/user_middleware.py new file mode 100644 index 0000000..4f239b5 --- /dev/null +++ b/bot/middleware/user_middleware.py @@ -0,0 +1,41 @@ +from typing import Any, Awaitable, Callable + +from aiogram import BaseMiddleware +from aiogram.types import TelegramObject, Update + +from bot.db import user_repo +from bot.i18n import Lang, detect_lang + + +class UserMiddleware(BaseMiddleware): + async def __call__( + self, + handler: Callable[[TelegramObject, dict[str, Any]], Awaitable[Any]], + event: TelegramObject, + data: dict[str, Any], + ) -> Any: + user = None + if isinstance(event, Update): + if event.message and event.message.from_user: + user = event.message.from_user + elif event.callback_query and event.callback_query.from_user: + user = event.callback_query.from_user + elif event.pre_checkout_query and event.pre_checkout_query.from_user: + user = event.pre_checkout_query.from_user + elif hasattr(event, "from_user") and event.from_user: + user = event.from_user + + if user: + db_user = await user_repo.get_or_create( + telegram_id=user.id, + username=user.username, + first_name=user.first_name, + lang=detect_lang(user.language_code).value, + ) + data["lang"] = Lang(db_user["lang"]) + data["db_user"] = db_user + else: + data["lang"] = Lang.EN + data["db_user"] = None + + return await handler(event, data) diff --git a/bot/models.py b/bot/models.py index 85b503b..fb32ea2 100644 --- a/bot/models.py +++ b/bot/models.py @@ -1,27 +1,39 @@ -from dataclasses import dataclass +from dataclasses import dataclass, field from datetime import datetime from enum import Enum -class ReportType(Enum): - CONTENT = "content" - CONTENT_STATS = "content_stats" - FULL_AUDIT = "full_audit" +class Depth(Enum): + BASIC = "basic" + STANDARD = "standard" + FULL = "full" @property def label(self) -> str: return { - ReportType.CONTENT: "Content Analysis", - ReportType.CONTENT_STATS: "Content + Stats", - ReportType.FULL_AUDIT: "Full Audit", + Depth.BASIC: "Basic", + Depth.STANDARD: "Standard", + Depth.FULL: "Full", }[self] + +class FocusArea(Enum): + PSYCHOLOGY = "psychology" + BUSINESS = "business" + MARKETING = "marketing" + CONTENT = "content" + AUDIENCE = "audience" + SENTIMENT = "sentiment" + @property - def description(self) -> str: + def label(self) -> str: return { - ReportType.CONTENT: "Topics, tone, themes, content strategy", - ReportType.CONTENT_STATS: "Above + posting frequency, engagement patterns", - ReportType.FULL_AUDIT: "All above + sentiment, audience insights, recommendations", + FocusArea.PSYCHOLOGY: "Psychology & Influence", + FocusArea.BUSINESS: "Business & Monetization", + FocusArea.MARKETING: "Marketing & Growth", + FocusArea.CONTENT: "Content Strategy", + FocusArea.AUDIENCE: "Audience & Engagement", + FocusArea.SENTIMENT: "Sentiment Analysis", }[self] @@ -33,3 +45,11 @@ class ChannelMessage: views: int | None = None forwards: int | None = None replies: int | None = None + + +@dataclass +class AnalysisSession: + channel: str + depth: Depth | None = None + focus_areas: list[FocusArea] = field(default_factory=list) + model_id: str | None = None diff --git a/bot/prompts/chunk_summary.py b/bot/prompts/chunk_summary.py index efd3156..9f0e3c4 100644 --- a/bot/prompts/chunk_summary.py +++ b/bot/prompts/chunk_summary.py @@ -1,4 +1,6 @@ -from bot.models import ReportType +from bot.models import Depth, FocusArea +from bot.prompts.depth import DEPTH_CHUNK_MODIFIERS +from bot.prompts.focus_areas import FOCUS_CHUNK_BULLETS _BASE = """\ You are analyzing a batch of Telegram channel posts. Extract structured insights from this chunk. @@ -10,52 +12,11 @@ Posts: {chunk_text} """ -_CONTENT = """\ -Focus on: -- Main topics and themes discussed -- Tone and communication style -- Content formats (long-form, short updates, lists, etc.) -- Key narratives or recurring ideas -- Notable quotes or standout posts - -Provide a concise structured summary.""" - -_CONTENT_STATS = """\ -Focus on: -- Main topics and themes discussed -- Tone and communication style -- Content formats used -- Key narratives or recurring ideas -- Posting frequency patterns in this batch -- Engagement patterns (which topics get more views/forwards/replies) -- Any notable spikes or drops in engagement - -Provide a concise structured summary with both qualitative and quantitative observations.""" - -_FULL_AUDIT = """\ -Focus on: -- Main topics and themes discussed -- Tone and communication style -- Content formats used -- Key narratives or recurring ideas -- Posting frequency patterns -- Engagement patterns with specific numbers -- Sentiment analysis (positive/negative/neutral distribution) -- Audience interaction patterns -- Content strengths and weaknesses -- Missed opportunities - -Provide a detailed structured summary covering all dimensions.""" - -CHUNK_PROMPTS = { - ReportType.CONTENT: _CONTENT, - ReportType.CONTENT_STATS: _CONTENT_STATS, - ReportType.FULL_AUDIT: _FULL_AUDIT, -} - def build_chunk_prompt( - report_type: ReportType, + depth: Depth, + focus_areas: list[FocusArea], + lang: str, title: str, chunk_text: str, chunk_idx: int, @@ -67,4 +28,14 @@ def build_chunk_prompt( total_chunks=total_chunks, chunk_text=chunk_text, ) - return base + "\n" + CHUNK_PROMPTS[report_type] + parts = [base] + + parts.append(f"Depth: {DEPTH_CHUNK_MODIFIERS[depth]}\n") + + parts.append("Focus on the following areas:") + for area in focus_areas: + parts.append(FOCUS_CHUNK_BULLETS[area]) + + parts.append(f"\nProvide the analysis in {'Russian' if lang == 'ru' else 'English'}.") + + return "\n".join(parts) diff --git a/bot/prompts/depth.py b/bot/prompts/depth.py new file mode 100644 index 0000000..a065e7d --- /dev/null +++ b/bot/prompts/depth.py @@ -0,0 +1,19 @@ +from bot.models import Depth + +DEPTH_CHUNK_MODIFIERS: dict[Depth, str] = { + Depth.BASIC: "Provide a brief, high-level summary. Focus on the most important points only. Be concise.", + Depth.STANDARD: "Provide a structured summary with moderate detail. Cover key patterns and notable examples.", + Depth.FULL: ( + "Provide an exhaustive, detailed analysis. Include specific examples, quotes, " + "numerical data, and subtle patterns. Leave nothing significant out." + ), +} + +DEPTH_SYNTHESIS_MODIFIERS: dict[Depth, str] = { + Depth.BASIC: "Keep the report concise and actionable. Use short sections with bullet points.", + Depth.STANDARD: "Provide a well-structured report with moderate depth. Balance brevity and detail.", + Depth.FULL: ( + "Produce a comprehensive, in-depth report. Include detailed analysis, specific evidence, " + "data-backed observations, and strategic recommendations. Be thorough." + ), +} diff --git a/bot/prompts/focus_areas.py b/bot/prompts/focus_areas.py new file mode 100644 index 0000000..6d9db36 --- /dev/null +++ b/bot/prompts/focus_areas.py @@ -0,0 +1,74 @@ +from bot.models import FocusArea + +FOCUS_CHUNK_BULLETS: dict[FocusArea, str] = { + FocusArea.PSYCHOLOGY: ( + "- Persuasion and influence techniques used\n" + "- Cognitive biases leveraged (scarcity, social proof, authority, etc.)\n" + "- Emotional triggers and manipulation patterns\n" + "- Framing and narrative control techniques" + ), + FocusArea.BUSINESS: ( + "- Revenue models and monetization strategies\n" + "- Product/service placement and promotion patterns\n" + "- Conversion funnels and calls-to-action\n" + "- Pricing psychology and offer structuring" + ), + FocusArea.MARKETING: ( + "- Growth tactics and audience acquisition strategies\n" + "- Viral mechanics and shareability factors\n" + "- Cross-promotion and collaboration patterns\n" + "- Brand positioning and differentiation" + ), + FocusArea.CONTENT: ( + "- Main topics and themes discussed\n" + "- Content formats (long-form, short updates, lists, media, etc.)\n" + "- Posting frequency and schedule patterns\n" + "- Narrative arcs and series/recurring segments\n" + "- Content quality and originality assessment" + ), + FocusArea.AUDIENCE: ( + "- Engagement patterns (views, forwards, replies per content type)\n" + "- Audience interaction and community dynamics\n" + "- Top-performing vs underperforming content\n" + "- Engagement drivers and detractors" + ), + FocusArea.SENTIMENT: ( + "- Overall sentiment distribution (positive/negative/neutral)\n" + "- Sentiment breakdown by topic\n" + "- Emotional tone shifts over time\n" + "- Controversial or polarizing content identification" + ), +} + +FOCUS_SYNTHESIS_SECTIONS: dict[FocusArea, str] = { + FocusArea.PSYCHOLOGY: ( + "## Psychology & Influence\n" + "Analyze persuasion techniques, cognitive biases, emotional triggers, " + "and manipulation patterns found across the channel's content." + ), + FocusArea.BUSINESS: ( + "## Business & Monetization\n" + "Detail revenue models, monetization strategies, product placements, " + "conversion patterns, and business-related content." + ), + FocusArea.MARKETING: ( + "## Marketing & Growth\n" + "Assess growth tactics, viral mechanics, cross-promotion strategies, " + "and brand positioning approaches." + ), + FocusArea.CONTENT: ( + "## Content Strategy\n" + "Analyze topics, formats, posting patterns, narrative arcs, " + "content quality, and overall editorial strategy." + ), + FocusArea.AUDIENCE: ( + "## Audience & Engagement\n" + "Deep dive into engagement metrics, audience interaction patterns, " + "community dynamics, and what drives or kills engagement." + ), + FocusArea.SENTIMENT: ( + "## Sentiment Analysis\n" + "Present sentiment distribution, emotional tone analysis, " + "sentiment by topic, and shifts over time." + ), +} diff --git a/bot/prompts/synthesis.py b/bot/prompts/synthesis.py index 46c95dd..55c54c8 100644 --- a/bot/prompts/synthesis.py +++ b/bot/prompts/synthesis.py @@ -1,4 +1,6 @@ -from bot.models import ReportType +from bot.models import Depth, FocusArea +from bot.prompts.depth import DEPTH_SYNTHESIS_MODIFIERS +from bot.prompts.focus_areas import FOCUS_SYNTHESIS_SECTIONS _BASE = """\ You are producing a final report for a Telegram channel analysis. @@ -13,103 +15,11 @@ Below are the summaries from each chunk of the channel's history: {chunk_summaries} """ -_CONTENT = """\ -Synthesize the chunk summaries into a comprehensive **Content Analysis Report** with these sections: - -## Overview -Brief channel description and positioning. - -## Key Topics & Themes -Main subject areas with examples. - -## Tone & Communication Style -How the channel communicates with its audience. - -## Content Strategy -Formats used, posting patterns, narrative arcs. - -## Notable Content -Standout posts or recurring motifs. - -## Summary -Key takeaways in 3-5 bullet points. - -Use Telegram-friendly formatting (bold, bullet points). Be specific — reference actual content patterns you observed.""" - -_CONTENT_STATS = """\ -Synthesize the chunk summaries into a comprehensive **Content & Stats Report** with these sections: - -## Overview -Brief channel description, subscriber count, and overall activity level. - -## Key Topics & Themes -Main subject areas ranked by frequency and engagement. - -## Tone & Communication Style -How the channel communicates with its audience. - -## Content Strategy -Formats used, posting patterns, narrative arcs. - -## Engagement Analysis -- Average engagement patterns -- Top-performing content types -- Engagement trends over time - -## Posting Patterns -Frequency, schedule consistency, any notable gaps or bursts. - -## Summary -Key takeaways in 5-7 bullet points mixing qualitative and quantitative insights. - -Use Telegram-friendly formatting. Include specific numbers where available.""" - -_FULL_AUDIT = """\ -Synthesize the chunk summaries into a comprehensive **Full Channel Audit** with these sections: - -## Executive Summary -Channel positioning, key metrics, and overall assessment. - -## Key Topics & Themes -Main subject areas ranked by frequency and engagement, with trend analysis. - -## Tone & Communication Style -Detailed analysis of voice, register, and audience relationship. - -## Content Strategy Assessment -Formats, patterns, narrative arcs — what works and what doesn't. - -## Engagement Deep Dive -- Engagement metrics and benchmarks -- Top-performing vs underperforming content -- Engagement drivers and detractors - -## Sentiment Analysis -Overall sentiment distribution, sentiment by topic, shifts over time. - -## Audience Insights -Inferred audience profile, interaction patterns, community dynamics. - -## Strengths -What the channel does well (3-5 points with evidence). - -## Areas for Improvement -Actionable recommendations (3-5 points with specific suggestions). - -## Strategic Recommendations -Forward-looking advice for channel growth and content optimization. - -Use Telegram-friendly formatting. Be specific — back every claim with observed patterns or data.""" - -SYNTHESIS_PROMPTS = { - ReportType.CONTENT: _CONTENT, - ReportType.CONTENT_STATS: _CONTENT_STATS, - ReportType.FULL_AUDIT: _FULL_AUDIT, -} - def build_synthesis_prompt( - report_type: ReportType, + depth: Depth, + focus_areas: list[FocusArea], + lang: str, title: str, username: str | None, subscribers: int | None, @@ -127,4 +37,18 @@ def build_synthesis_prompt( chunk_count=len(chunk_summaries), chunk_summaries=numbered, ) - return base + "\n" + SYNTHESIS_PROMPTS[report_type] + parts = [base] + + parts.append("Synthesize the chunk summaries into a comprehensive report with these sections:\n") + parts.append("## Overview\nBrief channel description and positioning.\n") + + for area in focus_areas: + parts.append(FOCUS_SYNTHESIS_SECTIONS[area] + "\n") + + parts.append("## Key Takeaways\nSummarize the most important findings.\n") + + parts.append(DEPTH_SYNTHESIS_MODIFIERS[depth]) + parts.append(f"\nWrite the entire report in {'Russian' if lang == 'ru' else 'English'}.") + parts.append("Use Telegram-friendly formatting (bold, bullet points). Be specific — reference actual content patterns.") + + return "\n".join(parts) diff --git a/bot/services/ai_client.py b/bot/services/ai_client.py new file mode 100644 index 0000000..f96fa56 --- /dev/null +++ b/bot/services/ai_client.py @@ -0,0 +1,106 @@ +import json +import logging +from abc import ABC, abstractmethod +from collections.abc import Callable, Coroutine +from typing import Any + +import anthropic +import httpx + +from bot.config import settings + +log = logging.getLogger(__name__) + +ProgressCallback = Callable[[str], Coroutine[Any, Any, None]] + + +class AIClient(ABC): + @abstractmethod + async def complete( + self, + system: str, + user: str, + max_tokens: int, + on_progress: ProgressCallback | None = None, + ) -> str: ... + + +class AnthropicClient(AIClient): + def __init__(self, model_id: str) -> None: + self.model_id = model_id + self.client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key) + + async def complete( + self, + system: str, + user: str, + max_tokens: int, + on_progress: ProgressCallback | None = None, + ) -> str: + kwargs: dict[str, Any] = { + "model": self.model_id, + "max_tokens": max_tokens, + "system": system, + "messages": [{"role": "user", "content": user}], + } + if "opus-4-6" in self.model_id: + kwargs["thinking"] = {"type": "adaptive"} + + async with self.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" + ) + + +class OpenRouterClient(AIClient): + BASE_URL = "https://openrouter.ai/api/v1/chat/completions" + + def __init__(self, model_id: str) -> None: + self.model_id = model_id + + async def complete( + self, + system: str, + user: str, + max_tokens: int, + on_progress: ProgressCallback | None = None, + ) -> str: + headers = { + "Authorization": f"Bearer {settings.openrouter_api_key}", + "Content-Type": "application/json", + } + payload = { + "model": self.model_id, + "max_tokens": max_tokens, + "messages": [ + {"role": "system", "content": system}, + {"role": "user", "content": user}, + ], + "stream": True, + } + + collected = [] + async with httpx.AsyncClient(timeout=300) as client: + async with client.stream("POST", self.BASE_URL, headers=headers, json=payload) as resp: + resp.raise_for_status() + async for line in resp.aiter_lines(): + if not line.startswith("data: "): + continue + data = line[6:] + if data == "[DONE]": + break + chunk = json.loads(data) + delta = chunk.get("choices", [{}])[0].get("delta", {}) + content = delta.get("content", "") + if content: + collected.append(content) + + return "".join(collected) + + +def get_ai_client(model_id: str) -> AIClient: + if "/" in model_id: + return OpenRouterClient(model_id) + return AnthropicClient(model_id) diff --git a/bot/services/analyzer.py b/bot/services/analyzer.py index 4172dc7..dd8f0da 100644 --- a/bot/services/analyzer.py +++ b/bot/services/analyzer.py @@ -4,46 +4,32 @@ from collections.abc import Callable, Coroutine from typing import Any import anthropic +import httpx -from bot.config import settings -from bot.models import ReportType +from bot.models import Depth, FocusArea from bot.prompts.chunk_summary import build_chunk_prompt from bot.prompts.synthesis import build_synthesis_prompt +from bot.services.ai_client import AIClient, get_ai_client log = logging.getLogger(__name__) -# Only 1 concurrent request to stay within rate limits _semaphore = asyncio.Semaphore(1) - MAX_RETRIES = 5 +ProgressCallback = Callable[[str], Coroutine[Any, Any, None]] -async def _call_claude( - client: anthropic.AsyncAnthropic, + +async def _call_with_retry( + client: AIClient, system: str, user: str, max_tokens: int, - on_progress: "ProgressCallback | None" = None, + 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" - ) + return await client.complete(system, user, max_tokens, on_progress) 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) @@ -57,15 +43,22 @@ async def _call_claude( await asyncio.sleep(wait) else: raise + except httpx.HTTPStatusError as e: + if e.response.status_code == 429 or e.response.status_code >= 500: + wait = 10 * (attempt + 1) + log.warning("HTTP %d, retrying in %ds", e.response.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]] + raise RuntimeError("Max retries exceeded") async def analyze_channel( - report_type: ReportType, + depth: Depth, + focus_areas: list[FocusArea], + lang: str, + model_id: str, chunks: list[str], channel_title: str, channel_username: str | None, @@ -73,8 +66,7 @@ async def analyze_channel( msg_count: int, on_progress: ProgressCallback | None = None, ) -> str: - client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key) - + client = get_ai_client(model_id) total = len(chunks) summaries: list[str] = [] @@ -83,9 +75,9 @@ async def analyze_channel( await on_progress(f"Analyzing chunk {i}/{total}...") prompt = build_chunk_prompt( - report_type, channel_title, chunk_text, i, total + depth, focus_areas, lang, channel_title, chunk_text, i, total ) - summary = await _call_claude( + summary = await _call_with_retry( client, system="You are an expert Telegram channel analyst.", user=prompt, @@ -95,24 +87,20 @@ async def analyze_channel( 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 on_progress(f"Chunk {i}/{total} done. Cooling down 60s...") 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, + depth, focus_areas, lang, + channel_title, channel_username, subscribers, + msg_count, summaries, ) - report = await _call_claude( + report = await _call_with_retry( client, system="You are an expert Telegram channel analyst producing a final report.", user=synthesis_prompt, diff --git a/bot/services/report_saver.py b/bot/services/report_saver.py new file mode 100644 index 0000000..595ebab --- /dev/null +++ b/bot/services/report_saver.py @@ -0,0 +1,18 @@ +import os +from datetime import datetime, timezone + +from bot.config import settings + + +async def save_report( + telegram_id: int, + channel: str, + report_md: str, +) -> str: + os.makedirs(settings.reports_dir, exist_ok=True) + ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") + filename = f"{telegram_id}_{channel}_{ts}.md" + path = os.path.join(settings.reports_dir, filename) + with open(path, "w", encoding="utf-8") as f: + f.write(report_md) + return path diff --git a/pyproject.toml b/pyproject.toml index 218b36f..d3ee41f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,10 +1,11 @@ [project] name = "tg-channel-analyzer" -version = "0.1.0" +version = "0.2.0" requires-python = ">=3.11" dependencies = [ "aiogram>=3.24,<4", "telethon>=1.42,<2", "anthropic>=0.80,<1", "pydantic-settings>=2.0", + "aiosqlite>=0.20", ]