vpn-telegram-bot/services/tickets.py

"""Тикеты поддержки: тексты, лимиты, уведомления."""

from __future__ import annotations

import html
import logging
from datetime import datetime
from typing import Any

from aiogram import Bot

from config import Settings

logger = logging.getLogger(__name__)

SUBJECT_MAX = 100
BODY_MAX = 2000
MAX_OPEN_PER_USER = 3

STATUS_LABELS: dict[str, str] = {
    "open": "🟢 Открыт",
    "answered": "💬 Есть ответ",
    "closed": "🔒 Закрыт",
}

STATUS_LABELS_SHORT: dict[str, str] = {
    "open": "Открыт",
    "answered": "Ответ",
    "closed": "Закрыт",
}


def status_label(status: str, *, short: bool = False) -> str:
    key = (status or "").strip().lower()
    table = STATUS_LABELS_SHORT if short else STATUS_LABELS
    return table.get(key, status or "—")


def clamp_subject(text: str) -> str:
    return (text or "").strip()[:SUBJECT_MAX]


def clamp_body(text: str) -> str:
    return (text or "").strip()[:BODY_MAX]


def _dt(value: Any) -> str:
    if value is None:
        return ""
    if isinstance(value, datetime):
        return value.isoformat()
    return str(value)


def ticket_to_dict(row: dict[str, Any], *, include_user: bool = False) -> dict[str, Any]:
    out: dict[str, Any] = {
        "id": int(row["id"]),
        "telegram_id": int(row["telegram_id"]),
        "subject": str(row.get("subject") or ""),
        "status": str(row.get("status") or "open"),
        "status_label": status_label(str(row.get("status") or "open"), short=True),
        "created_at": _dt(row.get("created_at")),
        "updated_at": _dt(row.get("updated_at")),
        "message_count": int(row.get("message_count") or 0),
    }
    if include_user:
        name = row.get("full_name") or row.get("login")
        username = row.get("username")
        if name:
            display = str(name)
        elif username:
            display = f"@{username}"
        else:
            display = f"ID {out['telegram_id']}"
        out["user"] = {
            "telegram_id": out["telegram_id"],
            "display_name": display,
            "username": username,
            "login": row.get("login"),
        }
    return out


def message_to_dict(row: dict[str, Any]) -> dict[str, Any]:
    return {
        "id": int(row["id"]),
        "ticket_id": int(row["ticket_id"]),
        "author_role": str(row.get("author_role") or ""),
        "author_id": int(row["author_id"]) if row.get("author_id") is not None else None,
        "body": str(row.get("body") or ""),
        "created_at": _dt(row.get("created_at")),
    }


def format_ticket_card(ticket: dict[str, Any], messages: list[dict[str, Any]]) -> str:
    tid = int(ticket["id"])
    subject = html.escape(str(ticket.get("subject") or ""))
    status = status_label(str(ticket.get("status") or "open"))
    lines = [
        f"<b>🎫 Тикет #{tid}</b>",
        f"Тема: <b>{subject}</b>",
        f"Статус: {status}",
        "",
    ]
    for msg in messages[-12:]:
        role = str(msg.get("author_role") or "")
        if role == "user":
            who = "Вы"
        elif role == "ai":
            who = "ИИ-поддержка"
        else:
            who = "Поддержка"
        body = html.escape(str(msg.get("body") or ""))
        lines.append(f"<b>{who}:</b>\n{body}")
        lines.append("")
    return "\n".join(lines).strip()


def format_admin_ticket_card(ticket: dict[str, Any], messages: list[dict[str, Any]]) -> str:
    tid = int(ticket["id"])
    subject = html.escape(str(ticket.get("subject") or ""))
    status = status_label(str(ticket.get("status") or "open"))
    uid = int(ticket["telegram_id"])
    uname = ticket.get("username")
    fname = ticket.get("full_name") or ticket.get("login") or ""
    user_line = f"ID <code>{uid}</code>"
    if uname:
        user_line += f" · @{html.escape(str(uname))}"
    if fname:
        user_line += f" · {html.escape(str(fname))}"
    lines = [
        f"<b>🎫 Тикет #{tid}</b>",
        f"Тема: <b>{subject}</b>",
        f"Статус: {status}",
        f"Пользователь: {user_line}",
        "",
    ]
    for msg in messages[-15:]:
        role = str(msg.get("author_role") or "")
        if role == "user":
            who = "Пользователь"
        elif role == "ai":
            who = "ИИ"
        else:
            who = "Админ"
        body = html.escape(str(msg.get("body") or ""))
        lines.append(f"<b>{who}:</b>\n{body}")
        lines.append("")
    return "\n".join(lines).strip()


async def notify_admins_new_ticket(
    bot: Bot,
    settings: Settings,
    *,
    ticket_id: int,
    telegram_id: int,
    subject: str,
    body: str,
    ai_handled: bool = False,
) -> None:
    prefix = "🤖 Новый тикет (ИИ ответил)" if ai_handled else "🆕 Новый тикет"
    text = (
        f"<b>{prefix} #{ticket_id}</b>\n"
        f"От: <code>{telegram_id}</code>\n"
        f"Тема: <b>{html.escape(subject)}</b>\n\n"
        f"{html.escape(body[:500])}"
    )
    from keyboards import admin_ticket_notify_inline

    kb = admin_ticket_notify_inline(ticket_id)
    for admin_id in settings.admin_ids:
        try:
            await bot.send_message(admin_id, text, reply_markup=kb, parse_mode="HTML")
        except Exception as exc:  # noqa: BLE001
            logger.warning("notify admin %s ticket: %s", admin_id, exc)


async def notify_admins_ticket_reply(
    bot: Bot,
    settings: Settings,
    *,
    ticket_id: int,
    telegram_id: int,
    body: str,
) -> None:
    text = (
        f"<b>💬 Ответ в тикете #{ticket_id}</b>\n"
        f"От: <code>{telegram_id}</code>\n\n"
        f"{html.escape(body[:500])}"
    )
    from keyboards import admin_ticket_notify_inline

    kb = admin_ticket_notify_inline(ticket_id)
    for admin_id in settings.admin_ids:
        try:
            await bot.send_message(admin_id, text, reply_markup=kb, parse_mode="HTML")
        except Exception as exc:  # noqa: BLE001
            logger.warning("notify admin %s reply: %s", admin_id, exc)


async def notify_user_ticket_reply(
    bot: Bot,
    *,
    telegram_id: int,
    ticket_id: int,
    subject: str,
    body: str,
    is_ai: bool = False,
) -> None:
    from keyboards import ticket_notify_inline

    title = "🤖 ИИ-поддержка" if is_ai else "💬 Ответ поддержки"
    text = (
        f"<b>{title} · тикет #{ticket_id}</b>\n"
        f"Тема: <b>{html.escape(subject)}</b>\n\n"
        f"{html.escape(body[:800])}"
    )
    try:
        await bot.send_message(
            telegram_id,
            text,
            reply_markup=ticket_notify_inline(ticket_id),
            parse_mode="HTML",
        )
    except Exception as exc:  # noqa: BLE001
        logger.warning("notify user %s ticket reply: %s", telegram_id, exc)