asxpi/tg-channel-analysis
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add multi-user features: i18n, payments, multi-model AI, focus areas

Browse source 
Replace single report type with composable analysis: depth levels
(basic/standard/full) + focus area multi-select (psychology, business,
marketing, content, audience, sentiment). Multi-step inline keyboard
flow guides users through selection.

- i18n: English + Russian, auto-detect from Telegram, /lang override
- AI providers: Anthropic + OpenRouter via AIClient abstraction
- Telegram Stars payments with per-depth pricing and free trial
- SQLite (aiosqlite) for users, analyses, payments tracking
- User middleware for auto-registration and language detection
- Report persistence: save .md locally, offer file download
- New commands: /features, /prices, /lang
- Composable prompt system: depth modifiers + focus area fragments
This commit is contained in:
Sergei Poljanski 2026-02-23 01:18:12 +02:00
commit c30b7e4675
27 changed files with 1155 additions and 280 deletions
Download patch file Download diff file Expand all files Collapse all files

106
bot/services/ai_client.py Normal file
View file

@ -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)

70
bot/services/analyzer.py
View file

@ -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,

18
bot/services/report_saver.py Normal file
View file

@ -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