В режиме «Помощь» нажми «🆘 Поддержка» или отправь /support. Напиши текстом, чем помочь, — можно приложить фото, документ или голосовое.
Сообщение уходит оператору, он отвечает прямо в этом чате. Переписка продолжается в том же обращении: можно отвлечься на другие режимы, вернуться и дописать.
+ """В режиме «Помощь» нажми «🆘 Поддержка» или отправь /support. В ответ придёт приветствие оператора — напиши текстом, чем помочь; можно приложить фото, документ или голосовое.
Оператор видит твоё сообщение и отвечает прямо в этом чате. Переписка продолжается в том же обращении: можно отвлечься на другие режимы, вернуться и дописать.
Кнопка «Главное меню» завершает обращение. Оно не удаляется: при следующем нажатии «Поддержка» оператор увидит всю историю.
""", ), } diff --git a/support.py b/support.py index d2320be..5657085 100644 --- a/support.py +++ b/support.py @@ -18,8 +18,9 @@ from typing import Any, Awaitable, Callable from aiogram import BaseMiddleware, Bot, F, Router from aiogram.types import Message, TelegramObject, User +from activity_phrases import support_greeting from keyboards import MAIN_MENU_VALUE, SUPPORT_BUTTON, SUPPORT_VALUE, TRAINING_RESET_VALUE, resolve_button_text -from user_store import connect +from user_store import connect, support_profile_summary logger = logging.getLogger(__name__) @@ -30,8 +31,7 @@ TOPIC_TITLE_LIMIT = 128 SUPPORT_START_TEXT = ( "🆘 Поддержка\n\n" - "Что случилось? Опиши проблему прямо здесь — я передам сообщение оператору, " - "он ответит в этом чате.\n\n" + "{greeting}\n\n" "Можно писать текстом, присылать фото, документы и голосовые. " "Кнопка «Главное меню» завершает обращение: бот снова работает как обычно, " "а при следующем нажатии «Поддержка» переписка продолжится в том же топике." @@ -176,7 +176,7 @@ def topic_title(user: User) -> str: return title[:TOPIC_TITLE_LIMIT] -def profile_card(user: User) -> str: +def profile_card(user: User, summary: dict[str, object] | None = None) -> str: full_name = " ".join( part for part in (user.first_name, user.last_name) if part ).strip() or "Пользователь" @@ -187,17 +187,47 @@ def profile_card(user: User) -> str: if user.username: lines.append(f"Username: @{html.escape(user.username)}") lines.append(f"ID:{user.id}")
+ if summary:
+ lines.append("")
+ lines.append(f"📅 Пользуется ботом: {_format_days(summary.get('days_used'))}")
+ lines.append(f"🔁 Продлений подписки: {_as_int(summary.get('renewals'))}")
+ lines.append(f"⏳ Осталось дней: {_format_days_left(summary)}")
+ lines.append(f"🎓 Бот обучен: {'да' if summary.get('is_trained') else 'нет'}")
lines.append("")
lines.append("Ответь в этом топике — сообщение придёт пользователю в личку.")
lines.append(f"Ответ с префиксом {html.escape(note_prefix())} останется здесь.")
return "\n".join(lines)
+def _as_int(value: object) -> int:
+ """Приводит значение сводки к int; нечисловое или пустое → 0."""
+ try:
+ return int(value or 0) # type: ignore[arg-type]
+ except (TypeError, ValueError):
+ return 0
+
+
+def _format_days(value: object) -> str:
+ return f"{_as_int(value)} дн."
+
+
+def _format_days_left(summary: dict[str, object]) -> str:
+ days_left = _as_int(summary.get("days_left")) if summary.get("days_left") is not None else -1
+ if days_left < 0:
+ return "нет активного периода"
+ return f"{days_left} дн."
+
+
async def send_profile_card(bot: Bot, user: User, thread_id: int) -> None:
group_id = support_group_id()
if group_id is None:
return
- card = profile_card(user)
+ try:
+ summary = await asyncio.to_thread(support_profile_summary, user.id)
+ except Exception as exc:
+ logger.warning("Support profile summary unavailable (%s)", type(exc).__name__)
+ summary = None
+ card = profile_card(user, summary)
photo_file_id = None
try:
photos = await bot.get_user_profile_photos(user_id=user.id, limit=1)
@@ -310,7 +340,11 @@ async def start_support(message: Message) -> None:
await message.answer(SUPPORT_UNAVAILABLE_TEXT)
return
await asyncio.to_thread(set_active, user.id, True)
- await message.answer(SUPPORT_START_TEXT, parse_mode="HTML")
+ greeting = support_greeting()
+ await message.answer(
+ SUPPORT_START_TEXT.format(greeting=html.escape(greeting)),
+ parse_mode="HTML",
+ )
# --------------------------------------------------------------------------- #
diff --git a/tests/test_support.py b/tests/test_support.py
new file mode 100644
index 0000000..b160aeb
--- /dev/null
+++ b/tests/test_support.py
@@ -0,0 +1,292 @@
+import asyncio
+import datetime
+import sqlite3
+import tempfile
+import unittest
+from contextlib import closing
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any, cast
+from unittest.mock import AsyncMock, patch
+
+import activity_phrases
+import support
+import user_store
+
+
+class _TempDbCase(unittest.IsolatedAsyncioTestCase):
+ def setUp(self):
+ self.temp_dir = tempfile.TemporaryDirectory()
+ self.addCleanup(self.temp_dir.cleanup)
+ self.db_path = str(Path(self.temp_dir.name) / "support_test.sqlite3")
+ env = patch.dict("os.environ", {"BOT_DB_PATH": self.db_path}, clear=False)
+ env.start()
+ self.addCleanup(env.stop)
+ user_store.init_db(self.db_path)
+ support.init_support_table(self.db_path)
+ with closing(sqlite3.connect(self.db_path)) as conn, conn:
+ conn.execute(
+ "CREATE TABLE IF NOT EXISTS payments ("
+ "telegram_payment_charge_id TEXT PRIMARY KEY,"
+ "telegram_user_id INTEGER NOT NULL,"
+ "invoice_payload TEXT NOT NULL,"
+ "currency TEXT NOT NULL,"
+ "total_amount INTEGER NOT NULL,"
+ "created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP)"
+ )
+
+ def _user(self, user_id=42, trained=0, created_at=None):
+ with closing(sqlite3.connect(self.db_path)) as conn, conn:
+ conn.execute(
+ "INSERT OR IGNORE INTO users (telegram_user_id, is_trained, created_at) "
+ "VALUES (?, ?, COALESCE(?, CURRENT_TIMESTAMP))",
+ (user_id, trained, created_at),
+ )
+
+ def _payments(self, user_id, count):
+ with closing(sqlite3.connect(self.db_path)) as conn, conn:
+ for index in range(count):
+ conn.execute(
+ "INSERT OR IGNORE INTO payments "
+ "(telegram_payment_charge_id, telegram_user_id, invoice_payload, "
+ "currency, total_amount) VALUES (?, ?, 'p', 'XTR', 100)",
+ (f"charge-{user_id}-{index}", user_id),
+ )
+
+
+class SupportProfileSummaryTests(_TempDbCase):
+ def test_summary_reports_defaults_for_unknown_user(self):
+ summary = user_store.support_profile_summary(999, self.db_path)
+ self.assertEqual(summary["user_id"], 999)
+ self.assertEqual(summary["days_used"], 0)
+ self.assertEqual(summary["renewals"], 0)
+ self.assertEqual(summary["days_left"], -1)
+ self.assertEqual(summary["period"], "none")
+ self.assertFalse(summary["is_trained"])
+
+ def test_summary_reports_trial_days_left_and_no_renewals(self):
+ self._user(42)
+ user_store.ensure_subscription(42, self.db_path)
+ summary = user_store.support_profile_summary(42, self.db_path)
+ self.assertEqual(summary["period"], "trial")
+ self.assertGreater(cast(int, summary["days_left"]), 0)
+ self.assertEqual(summary["renewals"], 0)
+
+ def test_summary_counts_payments_as_renewals(self):
+ self._user(42)
+ user_store.ensure_subscription(42, self.db_path)
+ self._payments(42, 2)
+ summary = user_store.support_profile_summary(42, self.db_path)
+ self.assertEqual(summary["renewals"], 2)
+
+ def test_summary_reports_paid_period_after_activation(self):
+ self._user(42)
+ user_store.ensure_subscription(42, self.db_path)
+ self._payments(42, 1)
+ user_store.activate_subscription(42, months=1, db_path=self.db_path)
+ summary = user_store.support_profile_summary(42, self.db_path)
+ self.assertEqual(summary["period"], "paid")
+ self.assertGreater(cast(int, summary["days_left"]), 0)
+
+ def test_summary_reports_trained_flag(self):
+ self._user(42, trained=1)
+ summary = user_store.support_profile_summary(42, self.db_path)
+ self.assertTrue(summary["is_trained"])
+
+ def test_summary_reports_days_used_from_created_at(self):
+ created = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) - datetime.timedelta(days=9)
+ self._user(42, created_at=created.isoformat())
+ summary = user_store.support_profile_summary(42, self.db_path)
+ self.assertEqual(summary["days_used"], 9)
+
+ def test_summary_reports_expired_period_as_negative(self):
+ self._user(42)
+ user_store.ensure_subscription(42, self.db_path)
+ with closing(sqlite3.connect(self.db_path)) as conn, conn:
+ conn.execute(
+ "UPDATE subscriptions SET trial_started_at = ? WHERE telegram_user_id = 42",
+ ((datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None)
+ - datetime.timedelta(days=10)).isoformat(),),
+ )
+ summary = user_store.support_profile_summary(42, self.db_path)
+ self.assertEqual(summary["days_left"], -1)
+
+ def test_summary_tolerates_broken_created_at(self):
+ self._user(42, created_at="not-a-date")
+ summary = user_store.support_profile_summary(42, self.db_path)
+ self.assertEqual(summary["days_used"], 0)
+
+ def test_summary_survives_missing_payments_table(self):
+ self._user(42)
+ user_store.ensure_subscription(42, self.db_path)
+ with closing(sqlite3.connect(self.db_path)) as conn, conn:
+ conn.execute("DROP TABLE IF EXISTS payments")
+ summary = user_store.support_profile_summary(42, self.db_path)
+ self.assertEqual(summary["renewals"], 0)
+
+
+class SupportProfileCardTests(unittest.TestCase):
+ def _tg_user(self, **kwargs):
+ defaults = dict(id=42, first_name="Иван", last_name=None, username="ivan", is_bot=False)
+ defaults.update(kwargs)
+ return SimpleNamespace(**defaults)
+
+ def test_card_without_summary_keeps_identity_only(self):
+ card = support.profile_card(self._tg_user())
+ self.assertIn("Иван", card)
+ self.assertIn("@ivan", card)
+ self.assertIn("42", card)
+ self.assertNotIn("Пользуется ботом", card)
+
+ def test_card_with_summary_shows_all_profile_fields(self):
+ summary = {
+ "user_id": 42,
+ "days_used": 9,
+ "renewals": 2,
+ "days_left": 12,
+ "period": "paid",
+ "is_trained": True,
+ }
+ card = support.profile_card(self._tg_user(), summary)
+ self.assertIn("Пользуется ботом", card)
+ self.assertIn("9 дн.", card)
+ self.assertIn("Продлений подписки", card)
+ self.assertIn("Осталось дней", card)
+ self.assertIn("12 дн.", card)
+ self.assertIn("Бот обучен", card)
+ self.assertIn("да", card)
+
+ def test_card_marks_untrained_bot(self):
+ summary = {"days_used": 1, "renewals": 0, "days_left": 2, "is_trained": False}
+ card = support.profile_card(self._tg_user(), summary)
+ self.assertIn("Бот обучен", card)
+ self.assertIn("нет", card)
+
+ def test_card_marks_expired_period(self):
+ summary = {"days_used": 40, "renewals": 0, "days_left": -1, "is_trained": False}
+ card = support.profile_card(self._tg_user(), summary)
+ self.assertIn("нет активного периода", card)
+
+ def test_card_falls_back_to_id_when_no_username(self):
+ card = support.profile_card(self._tg_user(username=None))
+ self.assertNotIn("@", card)
+ self.assertIn("42", card)
+
+ def test_card_tolerates_garbage_summary_values(self):
+ summary = {"days_used": None, "renewals": "x", "days_left": None, "is_trained": None}
+ card = support.profile_card(self._tg_user(), summary)
+ self.assertIn("0 дн.", card)
+ self.assertIn("нет активного периода", card)
+
+
+class SupportGreetingTests(unittest.TestCase):
+ def test_greeting_comes_from_the_pool(self):
+ for _ in range(30):
+ self.assertIn(support.support_greeting(), activity_phrases.SUPPORT_GREETINGS)
+
+ def test_greeting_never_mentions_relaying(self):
+ import re
+
+ for phrase in activity_phrases.SUPPORT_GREETINGS:
+ lowered = phrase.lower()
+ # Слова целиком: простое вхождение "бот" ловит "добрый".
+ words = set(re.findall(r"[а-яё]+", lowered))
+ for banned in ("передам", "передаю", "передаёт", "оператор", "бот", "боту", "ботом"):
+ self.assertNotIn(banned, words, phrase)
+
+ def test_greeting_asks_to_describe_the_problem(self):
+ for phrase in activity_phrases.SUPPORT_GREETINGS:
+ self.assertTrue(
+ any(marker in phrase.lower() for marker in ("опиш", "опис", "расскаж", "напиш", "случилось")),
+ phrase,
+ )
+
+ def test_pool_has_enough_variety(self):
+ self.assertGreaterEqual(len(activity_phrases.SUPPORT_GREETINGS), 5)
+ self.assertEqual(len(set(activity_phrases.SUPPORT_GREETINGS)), len(activity_phrases.SUPPORT_GREETINGS))
+
+ def test_env_overrides_the_pool(self):
+ custom = "Первая фраза|Вторая фраза"
+ with patch.dict("os.environ", {"SUPPORT_GREETINGS": custom}, clear=False):
+ for _ in range(10):
+ self.assertIn(support.support_greeting(), ("Первая фраза", "Вторая фраза"))
+
+ def test_env_pool_ignores_empty_entries(self):
+ with patch.dict("os.environ", {"SUPPORT_GREETINGS": " |Одна фраза| "}, clear=False):
+ self.assertEqual(support.support_greeting(), "Одна фраза")
+
+ def test_env_pool_of_separators_only_falls_back(self):
+ with patch.dict("os.environ", {"SUPPORT_GREETINGS": "|||"}, clear=False):
+ self.assertIn(support.support_greeting(), activity_phrases.SUPPORT_GREETINGS)
+
+
+class SupportStartTextTests(unittest.TestCase):
+ def test_start_text_embeds_greeting(self):
+ rendered = support.SUPPORT_START_TEXT.format(greeting="Привет!")
+ self.assertIn("Привет!", rendered)
+ self.assertIn("Главное меню", rendered)
+
+ def test_start_text_has_no_relay_wording(self):
+ rendered = support.SUPPORT_START_TEXT.format(greeting="Привет!")
+ self.assertNotIn("передам", rendered.lower())
+ self.assertNotIn("оператор", rendered.lower())
+
+
+class SupportTopicTests(_TempDbCase):
+ def _tg_user(self, **kwargs):
+ defaults = dict(id=42, first_name="Иван", last_name=None, username="ivan", is_bot=False)
+ defaults.update(kwargs)
+ return SimpleNamespace(**defaults)
+
+ def test_topic_is_created_once_and_reused(self):
+ with patch.dict("os.environ", {"SUPPORT_GROUP_ID": "-100123"}, clear=False):
+ bot = AsyncMock()
+ topic = SimpleNamespace(message_thread_id=777)
+ bot.create_forum_topic = AsyncMock(return_value=topic)
+ bot.get_user_profile_photos = AsyncMock(side_effect=RuntimeError("no photos"))
+ user = self._tg_user()
+
+ first = asyncio.run(support.ensure_topic(bot, user))
+ second = asyncio.run(support.ensure_topic(bot, user))
+
+ self.assertEqual(first, 777)
+ self.assertEqual(second, 777)
+ bot.create_forum_topic.assert_awaited_once()
+
+ def test_profile_card_is_sent_only_on_topic_creation(self):
+ with patch.dict("os.environ", {"SUPPORT_GROUP_ID": "-100123"}, clear=False):
+ bot = AsyncMock()
+ bot.create_forum_topic = AsyncMock(
+ return_value=SimpleNamespace(message_thread_id=778)
+ )
+ bot.get_user_profile_photos = AsyncMock(side_effect=RuntimeError("no photos"))
+ user = self._tg_user()
+
+ asyncio.run(support.ensure_topic(bot, user))
+ sends_after_first = bot.send_message.await_count
+ asyncio.run(support.ensure_topic(bot, user))
+
+ self.assertEqual(sends_after_first, 1)
+ self.assertEqual(bot.send_message.await_count, 1)
+ self.assertIn("Пользуется ботом", bot.send_message.await_args.kwargs["text"])
+
+ def test_topic_title_uses_username_when_present(self):
+ user = self._tg_user(first_name="Иван", last_name="Петров", username="ivan")
+ self.assertEqual(support.topic_title(user), "Иван Петров · @ivan")
+
+ def test_topic_title_falls_back_to_id(self):
+ self.assertEqual(support.topic_title(self._tg_user(username=None)), "Иван · 42")
+
+ def test_topic_title_is_capped(self):
+ long_name = "А" * 200
+ title = support.topic_title(self._tg_user(first_name=long_name, username="ivan"))
+ self.assertLessEqual(len(title), support.TOPIC_TITLE_LIMIT)
+
+ def test_support_is_disabled_without_group_id(self):
+ with patch.dict("os.environ", {"SUPPORT_GROUP_ID": ""}, clear=False):
+ self.assertFalse(support.is_enabled())
+ self.assertFalse(support.is_support_active(42, self.db_path))
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/user_store.py b/user_store.py
index c5be691..e01bac0 100644
--- a/user_store.py
+++ b/user_store.py
@@ -1,10 +1,12 @@
from contextlib import closing
import datetime
import json
+import logging
import os
import sqlite3
from pathlib import Path
+logger = logging.getLogger(__name__)
DEFAULT_DB_PATH = "/app/data/social_writer.sqlite3"
API_LOG_RETENTION_DAYS = 30
@@ -438,6 +440,74 @@ def is_user_trained(user_id: int, db_path: str | None = None) -> bool:
return bool(row["is_trained"]) if row else False
+def support_profile_summary(user_id: int, db_path: str | None = None) -> dict[str, object]:
+ """Сводка профиля для карточки обращения в поддержку.
+
+ Возвращает словарь с полями: user_id, days_used, renewals, days_left,
+ period (trial|paid|none), is_trained.
+ """
+ summary: dict[str, object] = {
+ "user_id": int(user_id),
+ "days_used": 0,
+ "renewals": 0,
+ "days_left": -1,
+ "period": "none",
+ "is_trained": False,
+ }
+ with closing(connect(db_path)) as connection:
+ row = connection.execute(
+ "SELECT created_at, is_trained FROM users WHERE telegram_user_id = ?",
+ (user_id,),
+ ).fetchone()
+ sub = connection.execute(
+ "SELECT * FROM subscriptions WHERE telegram_user_id = ?", (user_id,)
+ ).fetchone()
+ try:
+ renewals = connection.execute(
+ "SELECT COUNT(*) AS total FROM payments WHERE telegram_user_id = ?",
+ (user_id,),
+ ).fetchone()
+ except sqlite3.OperationalError as exc:
+ # Таблица платежей создаётся payments.init_payments_table().
+ logger.warning("Payments table is unavailable: %s", exc)
+ renewals = None
+
+ if row is not None:
+ summary["is_trained"] = bool(row["is_trained"])
+ created = _parse_utc(row["created_at"])
+ if created is not None:
+ summary["days_used"] = max(
+ 0, (datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) - created).days
+ )
+ if sub is not None:
+ days_left = _days_until(sub)
+ summary["days_left"] = days_left
+ summary["period"] = "paid" if sub["last_paid_at"] else "trial"
+ if sub["last_paid_at"] is None:
+ started = _parse_utc(sub["trial_started_at"])
+ if started is not None and (row is None or row["created_at"] is None):
+ summary["days_used"] = max(
+ 0,
+ (datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) - started).days,
+ )
+ if renewals is not None:
+ summary["renewals"] = int(renewals["total"])
+ return summary
+
+
+def _parse_utc(value: str | None) -> datetime.datetime | None:
+ """Разбирает SQLite CURRENT_TIMESTAMP в naive UTC; None при пустом/битом значении."""
+ if not value:
+ return None
+ try:
+ parsed = datetime.datetime.fromisoformat(str(value).replace("Z", "+00:00"))
+ except ValueError:
+ return None
+ if parsed.tzinfo is not None:
+ parsed = parsed.astimezone(datetime.timezone.utc).replace(tzinfo=None)
+ return parsed
+
+
def get_deploy_notice_version(db_path: str | None = None) -> str:
with closing(connect(db_path)) as connection, connection:
connection.execute(