report.py - master - makk/Stone_parser

report.py
146 строк5 KB
makk
stone_monitor
28 май 2026, 09:45
28 май 2026, 09:4506d9882
"""
Stone Monitor — форматирование отчёта для Telegram.
"""
import logging
from datetime import datetime, timezone

from config import CHANNEL_CATEGORIES

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


CATEGORY_NAMES = {
    "top_realty": "🏗 Топ рынка недвижимости",
    "media": "📰 СМИ",
    "competitors": "🏢 Конкуренты",
    "consultants": "🤝 Консультанты",
    "architects": "🎨 Архбюро",
    "extra": "🔖 Дополнительно",
}

SENTIMENT_ICONS = {
    "positive": "🟢",
    "neutral": "⚪",
    "negative": "🔴",
}


def format_report(mentions: list, stats: dict = None) -> str:
    """
    Формирует итоговый отчёт.

    Args:
        mentions: список dict с полями channel, category, url, summary, sentiment, is_comment, text
        stats: dict с ключами total_raw, total_24h (опционально)

    Returns:
        str: HTML-форматированный отчёт для Telegram
    """
    if not mentions:
        return format_empty_report(stats)

    # Статистика коллектора
    total_raw = stats.get("total_raw", "?") if stats else "?"
    total_24h = stats.get("total_24h", "?") if stats else "?"

    # Статистика упоминаний
    posts = [m for m in mentions if not m.get("is_comment")]
    comments = [m for m in mentions if m.get("is_comment")]
    total = len(mentions)

    # Группировка по категориям
    by_category = {}
    for m in mentions:
        cat = m.get("category", "unknown")
        if cat not in by_category:
            by_category[cat] = []
        by_category[cat].append(m)

    lines = []
    lines.append(f"<b>📊 STONE MONITOR — {datetime.now().strftime('%d.%m.%Y')}</b>")
    lines.append(f"Проанализировано <b>{total_raw}</b> сообщений в ТГ, за последние сутки сообщений <b>{total_24h}</b>, с упоминанием Stone <b>{total}</b> (посты: {len(posts)}, комм.: {len(comments)})")
    lines.append("")

    # Статистика по категориям
    lines.append("<b>По категориям:</b>")
    for cat_key, (cat_name, _) in CHANNEL_CATEGORIES.items():
        cat_mentions = by_category.get(cat_key, [])
        if cat_mentions:
            cat_posts = len([m for m in cat_mentions if not m.get("is_comment")])
            cat_comments = len([m for m in cat_mentions if m.get("is_comment")])
            lines.append(f"  {CATEGORY_NAMES.get(cat_key, cat_key)}: {len(cat_mentions)} (посты: {cat_posts}, комм.: {cat_comments})")
    lines.append("")
    lines.append("─" * 25)

    # Список упоминаний по категориям
    counter = 1
    for cat_key, (cat_name, _) in CHANNEL_CATEGORIES.items():
        cat_mentions = by_category.get(cat_key, [])
        if not cat_mentions:
            continue

        lines.append(f"\n<b>{CATEGORY_NAMES.get(cat_key, cat_key)}</b>")

        for m in cat_mentions:
            sentiment_icon = SENTIMENT_ICONS.get(m.get("sentiment", "neutral"), "⚪")
            type_label = "💬" if m.get("is_comment") else "📝"
            summary = m.get("summary", "")[:200]

            lines.append(
                f"{counter}. {sentiment_icon} {type_label} @{m['channel']}: "
                f"{summary}"
            )
            counter += 1

    lines.append("")
    lines.append("─" * 25)
    lines.append("<i>Stone Monitor v1.1 — ежедневный отчёт</i>")

    return "\n".join(lines)


def format_empty_report(stats: dict = None) -> str:
    """Отчёт при отсутствии упоминаний."""
    total_raw = stats.get("total_raw", "?") if stats else "?"
    total_24h = stats.get("total_24h", "?") if stats else "?"
    lines = [
        f"<b>📊 STONE MONITOR — {datetime.now().strftime('%d.%m.%Y')}</b>",
        f"Проанализировано <b>{total_raw}</b> сообщений в ТГ, за последние сутки сообщений <b>{total_24h}</b>",
        "",
        "✅ Упоминаний Stone за вчера не найдено.",
        "",
        "<i>Stone Monitor v1.1 — ежедневный отчёт</i>",
    ]
    return "\n".join(lines)


def format_channels_report(stats: dict) -> str:
    """Статистика по каналам (для кнопки/команды «Каналы»)."""
    channels = stats.get("channels", {}) if stats else {}

    lines = []
    lines.append("<b>📡 Статистика по каналам (за сутки):</b>")
    lines.append("")

    if not channels:
        lines.append("  Нет данных.")
    else:
        for ch in sorted(channels.keys()):
            count = channels[ch]
            short_name = ch[:18] + "…" if len(ch) > 18 else ch
            lines.append(f"  @{short_name}: {count}")

    return "\n".join(lines)


def split_message(text: str, max_length: int = 4000) -> list:
    """Разбивка длинного сообщения на части."""
    parts = []
    for paragraph in text.split("\n\n"):
        if not parts:
            parts.append(paragraph)
        elif len(parts[-1]) + len(paragraph) + 2 <= max_length:
            parts[-1] += "\n\n" + paragraph
        else:
            parts.append(paragraph)
    return parts