analyzer.py - master - makk/Stone_parser

analyzer.py
156 строк7 KB
makk
stone_monitor
28 май 2026, 09:45
28 май 2026, 09:4506d9882
"""
Stone Monitor — AI-анализ: disambiguation Stone-девелопер + саммари.
"""
import json
import logging
import re

from openai import OpenAI
from google import genai
from google.genai import types

from config import (
    OPENAI_API_KEY, GEMINI_API_KEY,
    AI_TEMPERATURE, AI_MAX_TOKENS,
    GEMINI_MODEL_ID, OPENAI_MODEL_ID,
)

logger = logging.getLogger("stone_monitor.analyzer")

# === ИНИЦИАЛИЗАЦИЯ ===
openai_client = OpenAI(api_key=OPENAI_API_KEY)
gemini_client = genai.Client(api_key=GEMINI_API_KEY)

# === ПРОМПТ ===
DISAMBIGUATION_PROMPT = """Ты — аналитик рынка недвижимости. Твоя задача — определить, относится ли Telegram-сообщение к девелоперу Stone, и дать краткое саммари.

Stone — российский девелопер жилой и коммерческой недвижимости (Москва, Россия).
Проекты: жилые комплексы, бизнес-центры, кварталы, офисные здания.
Ключевые слова-маркеры девелопера: «девелопер», «застройщик», «ЖК», «жилой комплекс», «проект», «строительство», «продажи квартир», «офис», «квартал», «Stone», «Стоун», «недвижимость», «апартаменты», «бизнес-центр», «BTS», «built-to-suit».

НЕ ЯВЛЯЕТСЯ девелопером Stone:
- stone как материал (камень, гранит, мрамор, natural stone)
- Rolling Stone (журнал/музыка)
- Stone Brewing (пивоварня)
- StoneX (финансовая компания)
- Sharon Stone, Oliver Stone (актёры/режиссёры)
- любые другие употребления слова stone вне контекста недвижимости

Для каждого сообщения верни JSON-объект:
{
  "is_stone_developer": true/false,
  "summary": "краткое саммари на русском, 1 предложение. Опиши: что сказано о Stone, в каком ключе (позитив/нейтрал/негатив), какой аспект (новый проект, продажи, партнёрство, проблема и т.д.). Если is_stone_developer=false, напиши 'Не относится к девелоперу Stone'",
  "sentiment": "positive/neutral/negative"
}

Пример правильного ответа:
{
  "is_stone_developer": true,
  "summary": "Stone начал строительство нового жилого квартала на 50 тыс. кв.м в районе Хорошёво-Мнёвники, нейтральное упоминание",
  "sentiment": "neutral"
}

Проанализируй следующие сообщения. Верни ТОЛЬКО JSON-массив объектов, без дополнительного текста:
"""


def _build_analysis_input(items: list) -> str:
    """Формирует текст для AI-анализа (пакетная обработка)."""
    lines = []
    for i, item in enumerate(items):
        channel = item.get("channel", "?")
        text = item.get("text", "")
        is_comment = "комментарий" if item.get("is_comment") else "пост"
        lines.append(
            f"[{i}] Канал: @{channel} | Тип: {is_comment}\n"
            f"Текст: {text[:600]}"
        )
    return "\n\n".join(lines)


def analyze_mentions(items: list) -> list:
    """
    Анализирует список сообщений через AI.
    Возвращает только те, где is_stone_developer == true,
    с добавленным полем summary и sentiment.
    """
    if not items:
        return []

    logger.info(f"AI analysis: {len(items)} items")

    full_input = _build_analysis_input(items)
    prompt = DISAMBIGUATION_PROMPT + "\n\n" + full_input

    try:
        response = gemini_client.models.generate_content(
            model=GEMINI_MODEL_ID,
            contents=prompt,
            config=types.GenerateContentConfig(
                temperature=AI_TEMPERATURE,
                max_output_tokens=AI_MAX_TOKENS,
            ),
        )
        raw_result = response.text
    except Exception as e:
        logger.warning(f"Gemini failed: {e}, trying OpenAI fallback...")
        try:
            response = openai_client.chat.completions.create(
                model=OPENAI_MODEL_ID,
                messages=[
                    {"role": "system", "content": DISAMBIGUATION_PROMPT},
                    {"role": "user", "content": full_input},
                ],
                temperature=AI_TEMPERATURE,
                max_tokens=AI_MAX_TOKENS,
            )
            raw_result = response.choices[0].message.content
        except Exception as e2:
            logger.error(f"OpenAI fallback also failed: {e2}")
            return []

    # Парсим JSON-ответ
    try:
        clean = raw_result.strip()

        # Убираем markdown ```json ... ``` или ``` ... ```
        clean = re.sub(r"^```[a-z]*\n?", "", clean)
        clean = re.sub(r"\n?```$", "", clean)
        clean = clean.strip()

        # Пробуем прямой парсинг
        try:
            analyses = json.loads(clean)
        except json.JSONDecodeError:
            # Regex-fallback: вытаскиваем JSON-массив [...] из текста
            match = re.search(r"\[.*\]", clean, re.DOTALL)
            if match:
                analyses = json.loads(match.group(0))
            else:
                raise

        if isinstance(analyses, dict):
            analyses = [analyses]
    except json.JSONDecodeError as e:
        logger.error(f"JSON parse error: {e}\nRaw: {raw_result[:500]}")
        return []

    # Объединяем AI-результаты с исходными элементами
    result = []
    for i, analysis in enumerate(analyses):
        if i >= len(items):
            break
        if analysis.get("is_stone_developer"):
            item = items[i].copy()
            item["summary"] = analysis.get("summary", "")
            item["sentiment"] = analysis.get("sentiment", "neutral")
            result.append(item)

    # Статистика
    posts_count = len([r for r in result if not r.get("is_comment")])
    comments_count = len([r for r in result if r.get("is_comment")])
    logger.info(
        f"AI result: {len(result)} Stone mentions ({posts_count} posts, {comments_count} comments) of {len(items)} checked"
    )

    return result