Агент мониторит упоминания бренда в Telegram и предупреждает о кризисе до того как он стал кризисом

import asyncio
import anthropic
import requests
from telethon import TelegramClient, events
from datetime import datetime, timezone

# === Конфигурация ===
# Получить на my.telegram.org → API development tools
TG_API_ID = 12345678
TG_API_HASH = "your_api_hash_here"
TG_SESSION_NAME = "brand_monitor_session"

# Claude API и Telegram Bot для алертов
CLAUDE_API_KEY = "your_claude_key"
ALERT_BOT_TOKEN = "your_bot_token"   # @BotFather
ALERT_CHAT_ID = "your_chat_id"       # куда слать алерты

# Каналы для мониторинга (username без @)
CHANNELS_TO_MONITOR = [
  "durov",              # примеры — замени на реальные
  "techcrunch_ru",
  "vc_ru",
  "roem_ru",
  # ... добавляй нужные
]

# Ключевые слова (в нижнем регистре)
BRAND_KEYWORDS = [
  "мой бренд",          # название компании
  "mybrand",            # транслит/домен
  "имя CEO",            # публичные персоны
  "продукт X",          # названия продуктов
]

# Порог уровня для немедленного алерта (2=критика, 3=угроза, 4=кризис)
ALERT_THRESHOLD = 2

# Дайджест — список всех упоминаний за день
daily_digest = []


def check_keywords(text: str) -> bool:
  """Проверяем наличие ключевых слов в сообщении."""
  text_lower = text.lower()
  return any(kw in text_lower for kw in BRAND_KEYWORDS)


def classify_with_claude(text: str, channel: str) -> dict:
  """Передаём текст в Claude для классификации тональности."""
  client = anthropic.Anthropic(api_key=CLAUDE_API_KEY)

  prompt = f"""Ты PR-аналитик. Оцени репутационный риск для бренда в тексте из Telegram-канала.

Канал: {channel}
Текст сообщения:
---
{text[:2000]}
---

Верни JSON с полями:
- level: число от 1 до 4
1 = упоминание (нейтральное, информационное)
2 = критика (негативная оценка, жалоба)
3 = угроза (призыв к бойкоту, фейк, серьёзный негатив)
4 = кризис (вирусный, массовое распространение, СМИ)
- sentiment: "positive" | "negative" | "neutral"
- summary: краткое описание на русском (1 предложение, максимум 100 символов)
- action: рекомендуемое действие на русском (1 предложение)

Отвечай только JSON, без markdown-обёртки."""

  message = client.messages.create(
      model="claude-sonnet-4-6",
      max_tokens=256,
      messages=[{"role": "user", "content": prompt}]
  )

  import json
  try:
      return json.loads(message.content[0].text)
  except Exception:
      return {"level": 1, "sentiment": "neutral",
              "summary": "Не удалось классифицировать", "action": "Проверить вручную"}


def send_alert(channel: str, text: str, classification: dict, msg_url: str):
  """Отправляем алерт в Telegram."""
  level = classification.get("level", 1)
  icons = {1: "📌", 2: "⚠️", 3: "🔴", 4: "🚨"}
  level_names = {1: "Упоминание", 2: "Критика", 3: "Угроза", 4: "КРИЗИС"}

  icon = icons.get(level, "📌")
  level_name = level_names.get(level, "Упоминание")

  alert_text = (
      f"{icon} <b>Уровень {level}: {level_name}</b>\n"
      f"Канал: <b>{channel}</b>\n"
      f"Оценка: {classification.get('summary', '')}\n"
      f"Действие: {classification.get('action', '')}\n"
      f"\n<a href='{msg_url}'>Открыть сообщение</a>\n"
      f"\n<pre>{text[:300]}{'...' if len(text) > 300 else ''}</pre>"
  )

  requests.post(
      f"https://api.telegram.org/bot{ALERT_BOT_TOKEN}/sendMessage",
      json={
          "chat_id": ALERT_CHAT_ID,
          "text": alert_text,
          "parse_mode": "HTML",
          "disable_web_page_preview": True,
      }
  )


def send_daily_digest():
  """Отправляем дайджест всех упоминаний за день."""
  if not daily_digest:
      return

  today = datetime.now(timezone.utc).strftime("%d.%m.%Y")
  pos = sum(1 for m in daily_digest if m["sentiment"] == "positive")
  neg = sum(1 for m in daily_digest if m["sentiment"] == "negative")
  neu = sum(1 for m in daily_digest if m["sentiment"] == "neutral")

  digest_text = (
      f"📊 <b>Дайджест упоминаний за {today}</b>\n"
      f"Всего: {len(daily_digest)} | "
      f"✅ {pos} позитивных | "
      f"❌ {neg} негативных | "
      f"➖ {neu} нейтральных\n\n"
  )

  for m in daily_digest[-10:]:  # последние 10 для краткости
      icon = "✅" if m["sentiment"] == "positive" else "❌" if m["sentiment"] == "negative" else "➖"
      digest_text += f"{icon} [{m['channel']}] {m['summary']}\n"

  requests.post(
      f"https://api.telegram.org/bot{ALERT_BOT_TOKEN}/sendMessage",
      json={"chat_id": ALERT_CHAT_ID, "text": digest_text, "parse_mode": "HTML"}
  )
  daily_digest.clear()


async def main():
  """Основной цикл мониторинга."""
  client = TelegramClient(TG_SESSION_NAME, TG_API_ID, TG_API_HASH)

  @client.on(events.NewMessage(chats=CHANNELS_TO_MONITOR))
  async def handler(event):
      text = event.message.message
      if not text or not check_keywords(text):
          return

      channel = event.chat.username or str(event.chat_id)
      msg_url = f"https://t.me/{channel}/{event.message.id}"

      print(f"[{datetime.now().strftime('%H:%M:%S')}] Упоминание в @{channel}")

      # Классифицируем через Claude
      classification = classify_with_claude(text, channel)

      # Сохраняем в дайджест
      daily_digest.append({
          "channel": channel,
          "sentiment": classification.get("sentiment", "neutral"),
          "summary": classification.get("summary", ""),
          "url": msg_url,
      })

      # Алерт при угрозе
      if classification.get("level", 1) >= ALERT_THRESHOLD:
          send_alert(channel, text, classification, msg_url)
          print(f"  → Алерт отправлен (уровень {classification.get('level')})")

  await client.start()
  print("Мониторинг запущен. Слушаем каналы...")

  # Дайджест каждые 24 часа (86400 секунд)
  async def digest_loop():
      while True:
          await asyncio.sleep(86400)
          send_daily_digest()

  await asyncio.gather(
      client.run_until_disconnected(),
      digest_loop()
  )


if __name__ == "__main__":
  asyncio.run(main())