Мониторинг видимости в нейросетях

import httpx
import asyncio
import json
from datetime import datetime
from dataclasses import dataclass, asdict

@dataclass
class CitationCheck:
    query: str
    timestamp: str
    cited: bool
    position: int | None
    total_sources: int
    all_citations: list[str]

async def check_citation(
    query: str,
    domain: str,
    api_key: str,
) -> CitationCheck:
    """Проверяет цитирование домена в ответе Perplexity."""
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            "https://api.perplexity.ai/chat/completions",
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json",
            },
            json={
                "model": "sonar",
                "messages": [
                    {"role": "user", "content": query}
                ],
            },
            timeout=30,
        )

    data = resp.json()
    citations = data.get("citations", [])

    position = None
    for i, url in enumerate(citations):
        if domain in url:
            position = i
            break

    return CitationCheck(
        query=query,
        timestamp=datetime.now().isoformat(),
        cited=position is not None,
        position=position,
        total_sources=len(citations),
        all_citations=citations,
    )

async def run_monitoring(
    queries: list[str],
    domain: str,
    api_key: str,
    output_file: str = "citations.jsonl",
):
    """Запускает мониторинг для списка запросов."""
    tasks = [
        check_citation(q, domain, api_key)
        for q in queries
    ]
    results = await asyncio.gather(*tasks)

    # Сохраняем в JSONL
    with open(output_file, "a") as f:
        for r in results:
            f.write(json.dumps(asdict(r), ensure_ascii=False) + "\n")

    # Статистика
    cited = sum(1 for r in results if r.cited)
    rate = cited / len(results) * 100

    print(f"Citation Rate: {rate:.1f}% ({cited}/{len(results)})")

    for r in results:
        status = f"[pos {r.position}]" if r.cited else "[---]"
        print(f"  {status} {r.query}")

    return results

queries = [
    "что такое GEO оптимизация",
    "как оптимизировать контент для AI поисковиков",
    "schema.org для нейросетей",
    "E-E-A-T для AI поиска",
    "perplexity оптимизация контента",
]

asyncio.run(
    run_monitoring(queries, "example.com", "pplx-xxx")
)

import re
from collections import Counter, defaultdict
from datetime import datetime

AI_BOT_PATTERNS = {
    "GPTBot": r"GPTBot",
    "ChatGPT-User": r"ChatGPT-User",
    "PerplexityBot": r"PerplexityBot",
    "ClaudeBot": r"ClaudeBot",
    "Google-Extended": r"Google-Extended",
    "Applebot-Extended": r"Applebot-Extended",
    "Bytespider": r"Bytespider",
    "cohere-ai": r"cohere-ai",
}

LOG_PATTERN = re.compile(
    r'(?P<ip>\S+) \S+ \S+ \[(?P<time>[^\]]+)\] '
    r'"(?P<method>\S+) (?P<path>\S+) \S+" '
    r'(?P<status>\d+) (?P<size>\d+) '
    r'"(?P<referer>[^"]*)" "(?P<ua>[^"]*)"'
)

def analyze_ai_traffic(log_path: str) -> dict:
    """Анализирует AI-бот трафик из Nginx-логов."""
    bot_pages = defaultdict(Counter)  # bot -> {page: count}
    bot_daily = defaultdict(Counter)  # bot -> {date: count}

    with open(log_path) as f:
        for line in f:
            match = LOG_PATTERN.match(line)
            if not match:
                continue

            ua = match.group("ua")
            path = match.group("path")
            status = match.group("status")

            # Только успешные запросы
            if not status.startswith("2"):
                continue

            for bot_name, pattern in AI_BOT_PATTERNS.items():
                if re.search(pattern, ua, re.IGNORECASE):
                    bot_pages[bot_name][path] += 1

                    # Парсим дату
                    time_str = match.group("time")
                    date = datetime.strptime(
                        time_str.split(":")[0], "%d/%b/%Y"
                    ).strftime("%Y-%m-%d")
                    bot_daily[bot_name][date] += 1
                    break

    return {
        "pages_by_bot": {
            bot: dict(pages.most_common(20))
            for bot, pages in bot_pages.items()
        },
        "daily_by_bot": {
            bot: dict(sorted(daily.items()))
            for bot, daily in bot_daily.items()
        },
        "total_by_bot": {
            bot: sum(pages.values())
            for bot, pages in bot_pages.items()
        },
    }

stats = analyze_ai_traffic("/var/log/nginx/access.log")

print("=== AI Bot Traffic ===")
for bot, total in sorted(
    stats["total_by_bot"].items(), key=lambda x: -x[1]
):
    print(f"{bot}: {total} запросов")
    top_pages = stats["pages_by_bot"][bot]
    for page, count in list(top_pages.items())[:5]:
        print(f"  {page}: {count}")

// server/api/geo/visibility.get.ts
import { defineEventHandler, getQuery } from 'h3'

interface VisibilityData {
  date: string
  citationRate: number
  aiTraffic: number
  crawlFrequency: number
  topQueries: { query: string; cited: boolean }[]
}

export default defineEventHandler(async (event) => {
  const query = getQuery(event)
  const days = Number(query.days) || 30

  // В реальном проекте данные берутся из БД
  // Здесь — пример структуры ответа

  const data: VisibilityData = {
    date: new Date().toISOString(),
    citationRate: 34.5,
    aiTraffic: 1247,
    crawlFrequency: 523,
    topQueries: [
      { query: 'что такое GEO', cited: true },
      { query: 'оптимизация для AI поисковиков', cited: true },
      { query: 'schema.org для AI', cited: false },
      { query: 'E-E-A-T нейропоиск', cited: true },
    ],
  }

  return data
})

import asyncio
import json
from datetime import datetime
from pathlib import Path

from monitor import run_monitoring

DOMAIN = "example.com"
API_KEY = "pplx-xxx"  # Из переменных окружения
OUTPUT_DIR = Path("/opt/geo-monitor/data")

QUERIES = [
    "что такое GEO оптимизация",
    "как AI поисковики выбирают источники",
    "schema.org для нейросетей",
    "E-E-A-T для AI поиска",
    "мониторинг видимости в AI",
    "оптимизация для perplexity",
    "chatgpt search оптимизация",
    "GEO кейсы до после",
]

async def main():
    today = datetime.now().strftime("%Y-%m-%d")
    output_file = OUTPUT_DIR / f"citations-{today}.jsonl"

    results = await run_monitoring(
        QUERIES, DOMAIN, API_KEY, str(output_file)
    )

    # Суммарный отчёт
    cited = sum(1 for r in results if r.cited)
    report = {
        "date": today,
        "citation_rate": cited / len(results) * 100,
        "cited_count": cited,
        "total_queries": len(results),
    }

    report_file = OUTPUT_DIR / "daily-reports.jsonl"
    with open(report_file, "a") as f:
        f.write(json.dumps(report) + "\n")

asyncio.run(main())