Агент готовит ежемесячный отчёт для клиентов агентства: 3 часа работы за 15 минут

import asyncio
import anthropic
import requests
import json
from datetime import datetime, timedelta
from googleapiclient.discovery import build
from google.oauth2.service_account import Credentials

# ─── Конфигурация ───────────────────────────────────────────
DIRECT_TOKEN      = "YOUR_DIRECT_TOKEN"
METRIKA_TOKEN     = "YOUR_METRIKA_TOKEN"
WEBMASTER_TOKEN   = "YOUR_WEBMASTER_TOKEN"
CLAUDE_API_KEY    = "YOUR_CLAUDE_KEY"
TELEGRAM_TOKEN    = "YOUR_BOT_TOKEN"
TELEGRAM_CHAT_ID  = "YOUR_CHAT_ID"
GOOGLE_CREDS_FILE = "service_account.json"   # Google Service Account JSON

# Конфигурация клиентов
CLIENTS = [
  {
      "name": "ООО Климат-Сервис",
      "direct_login": "klimat-service",
      "metrika_counter": "12345678",
      "webmaster_host": "https://klimat-service.ru",
      "telegram_chat": "-100123456789",
  },
  # ... добавить остальных клиентов
]

DIRECT_URL  = "https://api.direct.yandex.com/json/v5/"
METRIKA_URL = "https://api-metrika.yandex.net/stat/v1/data"
WM_URL      = "https://api.webmaster.yandex.net/v4/user/"

# ─── Сбор данных ─────────────────────────────────────────────

def get_direct_data(client_login: str) -> dict:
  """Расход, клики и конверсии за текущий и прошлый месяц."""
  now = datetime.now()
  date_to   = now.strftime("%Y-%m-%d")
  date_from = (now - timedelta(days=30)).strftime("%Y-%m-%d")
  prev_from = (now - timedelta(days=60)).strftime("%Y-%m-%d")
  prev_to   = (now - timedelta(days=31)).strftime("%Y-%m-%d")

  headers = {
      "Authorization": f"Bearer {DIRECT_TOKEN}",
      "Client-Login": client_login,
      "Accept-Language": "ru",
  }

  def fetch_stats(date_f, date_t, period_name):
      payload = {
          "params": {
              "SelectionCriteria": {"DateFrom": date_f, "DateTo": date_t},
              "Goals": [], "AttributionModels": ["LC"],
              "FieldNames": ["CampaignName","Cost","Clicks","Impressions","Conversions","Ctr"],
              "ReportName": f"monthly_{period_name}_{date_f}",
              "ReportType": "CAMPAIGN_PERFORMANCE_REPORT",
              "DateRangeType": "CUSTOM_DATE",
              "Format": "JSON",
              "IncludeVAT": "YES",
              "IncludeDiscount": "NO",
          }
      }
      r = requests.post(DIRECT_URL + "reports", headers=headers, json=payload)
      if r.status_code == 201:
          import time; time.sleep(5)
          r = requests.post(DIRECT_URL + "reports", headers=headers, json=payload)
      r.raise_for_status()
      data = r.json().get("data", [])
      total_cost = sum(float(row.get("Cost", 0)) for row in data) / 1_000_000
      total_conv = sum(int(row.get("Conversions", 0)) for row in data)
      total_clicks = sum(int(row.get("Clicks", 0)) for row in data)
      cpl = round(total_cost / total_conv, 0) if total_conv > 0 else 0
      return {
          "cost": round(total_cost, 0),
          "conversions": total_conv,
          "clicks": total_clicks,
          "cpl": cpl,
      }

  return {
      "current": fetch_stats(date_from, date_to, "current"),
      "previous": fetch_stats(prev_from, prev_to, "previous"),
  }

def get_metrika_data(counter_id: str) -> dict:
  """Трафик по источникам за текущий и прошлый месяц."""
  now = datetime.now()
  date_to   = now.strftime("%Y-%m-%d")
  date_from = (now - timedelta(days=30)).strftime("%Y-%m-%d")
  prev_from = (now - timedelta(days=60)).strftime("%Y-%m-%d")
  prev_to   = (now - timedelta(days=31)).strftime("%Y-%m-%d")

  headers = {"Authorization": f"OAuth {METRIKA_TOKEN}"}

  def fetch(date_f, date_t):
      params = {
          "ids": counter_id,
          "date1": date_f, "date2": date_t,
          "metrics": "ym:s:visits,ym:s:users,ym:s:bounceRate,ym:s:goalReachesAny",
          "dimensions": "ym:s:trafficSourceName",
          "sort": "-ym:s:visits",
          "limit": 8,
      }
      r = requests.get(METRIKA_URL, headers=headers, params=params)
      r.raise_for_status()
      result = r.json()
      sources = []
      total_visits = 0
      for row in result.get("data", []):
          v = int(row["metrics"][0])
          total_visits += v
          sources.append({
              "source": row["dimensions"][0]["name"],
              "visits": v,
              "bounce": round(row["metrics"][2], 1),
          })
      return {"total_visits": total_visits, "sources": sources[:5]}

  return {
      "current": fetch(date_from, date_to),
      "previous": fetch(prev_from, prev_to),
  }

def get_webmaster_data(host_url: str) -> dict:
  """Позиции и индексация из Яндекс.Вебмастер."""
  headers = {"Authorization": f"OAuth {WEBMASTER_TOKEN}"}
  # Получить user_id
  r = requests.get(WM_URL + "me", headers=headers)
  r.raise_for_status()
  user_id = r.json()["user_id"]

  # Хосты пользователя
  r = requests.get(WM_URL + f"{user_id}/hosts", headers=headers)
  r.raise_for_status()
  hosts = r.json().get("hosts", [])
  host_id = None
  for h in hosts:
      if host_url in h.get("unicode_host_url", ""):
          host_id = h["host_id"]
          break
  if not host_id:
      return {"error": "host not found", "top10": 0, "indexed": 0}

  # Статистика запросов
  r = requests.get(
      WM_URL + f"{user_id}/hosts/{host_id}/search-queries/popular",
      headers=headers,
      params={"query_indicator": "TOTAL_SHOWS,TOTAL_CLICKS,AVG_CLICK_POSITION", "limit": 100},
  )
  r.raise_for_status()
  queries = r.json().get("queries", [])
  top10 = sum(1 for q in queries if q.get("indicators", {}).get("AVG_CLICK_POSITION", 99) <= 10)

  # Индексация
  r = requests.get(WM_URL + f"{user_id}/hosts/{host_id}/summary", headers=headers)
  r.raise_for_status()
  summary = r.json()
  indexed = summary.get("indexing", {}).get("yandex", {}).get("indexedDocumentsCount", 0)

  return {"top10_queries": top10, "indexed_pages": indexed}

# ─── Генерация отчёта через Claude ───────────────────────────

def generate_report(client: dict, direct: dict, metrika: dict, webmaster: dict) -> str:
  anthropic_client = anthropic.Anthropic(api_key=CLAUDE_API_KEY)

  d_cur = direct["current"]
  d_prv = direct["previous"]
  m_cur = metrika["current"]
  m_prv = metrika["previous"]

  cost_delta = round((d_cur["cost"] - d_prv["cost"]) / d_prv["cost"] * 100, 1) if d_prv["cost"] else 0
  conv_delta = round((d_cur["conversions"] - d_prv["conversions"]) / d_prv["conversions"] * 100, 1) if d_prv["conversions"] else 0
  traffic_delta = round((m_cur["total_visits"] - m_prv["total_visits"]) / m_prv["total_visits"] * 100, 1) if m_prv["total_visits"] else 0

  prompt = f"""Ты помощник по digital-маркетингу. Напиши ежемесячный отчёт для клиента агентства.

КЛИЕНТ: {client["name"]}
ПЕРИОД: {datetime.now().strftime("%B %Y")}

ДАННЫЕ РЕКЛАМЫ (Яндекс.Директ):
- Расход: {d_cur["cost"]:,.0f} ₽ ({cost_delta:+.1f}% vs прошлый месяц)
- Лиды: {d_cur["conversions"]} ({conv_delta:+.1f}% vs прошлый месяц)
- Кликов: {d_cur["clicks"]}
- Стоимость лида: {d_cur["cpl"]:,.0f} ₽ (прошлый: {d_prv["cpl"]:,.0f} ₽)

ДАННЫЕ САЙТА (Яндекс.Метрика):
- Визитов: {m_cur["total_visits"]} ({traffic_delta:+.1f}% vs прошлый месяц)
- Источники трафика: {json.dumps(m_cur["sources"][:3], ensure_ascii=False)}

SEO (Яндекс.Вебмастер):
- Запросов в топ-10: {webmaster.get("top10_queries", "н/д")}
- Проиндексировано страниц: {webmaster.get("indexed_pages", "н/д")}

Напиши отчёт на русском языке из четырёх разделов:
1. «Введение» — 3-4 предложения о месяце в целом
2. «Что сделали» — конкретные работы с результатами
3. «На что обратить внимание» — 1-2 проблемные точки или риска
4. «План на следующий месяц» — 3 конкретных приоритета

Пиши как профессионал который общается с клиентом-предпринимателем: конкретно, без маркетинговой воды, с числами. Каждый раздел — 3-5 предложений."""

  message = anthropic_client.messages.create(
      model="claude-sonnet-4-6",
      max_tokens=1500,
      messages=[{"role": "user", "content": prompt}]
  )
  return message.content[0].text

# ─── Экспорт в Google Docs ────────────────────────────────────

def export_to_docs(client_name: str, report_text: str) -> str:
  """Создаёт Google Doc с отчётом, возвращает URL."""
  creds = Credentials.from_service_account_file(
      GOOGLE_CREDS_FILE,
      scopes=["https://www.googleapis.com/auth/documents",
              "https://www.googleapis.com/auth/drive"]
  )
  docs_service  = build("docs", "v1", credentials=creds)
  drive_service = build("drive", "v3", credentials=creds)

  title = f"Отчёт {client_name} — {datetime.now().strftime('%B %Y')}"
  doc = docs_service.documents().create(body={"title": title}).execute()
  doc_id = doc["documentId"]

  # Вставляем текст
  docs_service.documents().batchUpdate(
      documentId=doc_id,
      body={"requests": [
          {"insertText": {"location": {"index": 1}, "text": report_text}}
      ]}
  ).execute()

  # Даём права на просмотр по ссылке
  drive_service.permissions().create(
      fileId=doc_id,
      body={"type": "anyone", "role": "reader"},
  ).execute()

  return f"https://docs.google.com/document/d/{doc_id}/edit"

# ─── Оркестратор для одного клиента ──────────────────────────

def process_client(client: dict) -> None:
  print(f"  Обрабатываем: {client['name']}")

  direct    = get_direct_data(client["direct_login"])
  metrika   = get_metrika_data(client["metrika_counter"])
  webmaster = get_webmaster_data(client["webmaster_host"])

  report_text = generate_report(client, direct, metrika, webmaster)
  doc_url     = export_to_docs(client["name"], report_text)

  # Уведомляем менеджера в Telegram
  msg = (
      f"Отчёт готов: {client['name']}\n"
      f"Ссылка: {doc_url}\n"
      f"Нужна правка ~5 мин перед отправкой клиенту"
  )
  requests.post(
      f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage",
      json={"chat_id": client.get("telegram_chat", TELEGRAM_CHAT_ID), "text": msg}
  )
  print(f"  Готово: {doc_url}")

# ─── Запуск ───────────────────────────────────────────────────

if __name__ == "__main__":
  print(f"Старт: {len(CLIENTS)} клиентов")
  for i, client in enumerate(CLIENTS):
      process_client(client)
  print("Все отчёты готовы.")