184 lines
7.2 KiB
Python
184 lines
7.2 KiB
Python
"""
|
||
Фоновый опрос Intradesk API для push-уведомлений о новых комментариях.
|
||
|
||
Алгоритм за один цикл:
|
||
1. Один запрос — все открытые задачи по всем клиентам сразу.
|
||
2. Группируем по clientid в Python — никакого цикла запросов по клиентам.
|
||
3. Для задач с изменившимся updatedat — запрашиваем историю (1 запрос на задачу).
|
||
4. Нашли новый комментарий → уведомляем всех пользователей организации.
|
||
5. Задачи, исчезнувшие из открытых — убираем из слежения.
|
||
|
||
Первая встреча с задачей: запоминаем updatedat как базовую линию, историю не смотрим.
|
||
Первое изменение updatedat: устанавливаем comment-baseline, не уведомляем.
|
||
Последующие изменения: проверяем и уведомляем.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
from collections import defaultdict
|
||
|
||
import config
|
||
from database.queries.bot_users import get_all_bot_users
|
||
from database.queries.task_watch import (
|
||
get_task_watch,
|
||
get_watched_clients,
|
||
remove_tasks_not_in,
|
||
upsert_task_watch,
|
||
)
|
||
from external.intradesk_api import (
|
||
fetch_all_open_tasks_lightweight,
|
||
get_task_comments,
|
||
invalidate_task_comment,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
async def _send_notification(
|
||
bot,
|
||
chat_id: int,
|
||
max_user_id: int,
|
||
task_id: int,
|
||
task_number: int,
|
||
comment: dict,
|
||
) -> None:
|
||
from bot.keyboards import notification_kb
|
||
from database.queries.reply_context import set_reply_context
|
||
|
||
ctext = comment["text"]
|
||
if len(ctext) > 200:
|
||
ctext = ctext[:200] + "…"
|
||
|
||
text = (
|
||
f"💬 Новый ответ по заявке #{task_number}\n\n"
|
||
f"— {comment['author']}, {comment['date']}"
|
||
f"«{ctext}»\n"
|
||
|
||
)
|
||
await bot.send_message(
|
||
chat_id=chat_id,
|
||
text=text,
|
||
attachments=[notification_kb(task_id=task_id, task_number=task_number)],
|
||
)
|
||
await set_reply_context(max_user_id, task_id, task_number)
|
||
|
||
|
||
async def _process_task(task: dict, org_id: int, bot) -> None:
|
||
"""Проверить одну задачу и при необходимости отправить уведомление."""
|
||
tid = task["id"]
|
||
task_number = task["task_number"]
|
||
client_id = task["client_id"]
|
||
updatedat = task["updatedat"]
|
||
|
||
watch = await get_task_watch(tid)
|
||
|
||
if watch is None:
|
||
# Новая задача — запоминаем updatedat, history не трогаем
|
||
await upsert_task_watch(tid, client_id, task_number, updatedat, "")
|
||
logger.debug("task_watch new: task=%s client=%s", tid, client_id)
|
||
return
|
||
|
||
if updatedat == watch["last_updatedat"]:
|
||
return # Ничего не изменилось — history не трогаем
|
||
|
||
# updatedat изменился — только теперь идём в history API
|
||
last_known = watch["last_comment_date"]
|
||
|
||
invalidate_task_comment(tid)
|
||
comments = await get_task_comments(tid, count=10)
|
||
|
||
if not last_known:
|
||
# Первое изменение с момента начала слежения — устанавливаем baseline
|
||
baseline = comments[0]["date"] if comments else ""
|
||
await upsert_task_watch(tid, client_id, task_number, updatedat, baseline)
|
||
return
|
||
|
||
new_comments = [c for c in comments if c["date"] > last_known]
|
||
|
||
if new_comments:
|
||
latest = new_comments[0]
|
||
users = await get_all_bot_users(is_authorized=1, is_blocked=0, org_id=org_id)
|
||
notified = 0
|
||
for user in users:
|
||
chat_id = user.get("max_chat_id")
|
||
max_user_id = user.get("max_user_id")
|
||
if not chat_id or not max_user_id:
|
||
continue
|
||
try:
|
||
await _send_notification(bot, chat_id, max_user_id, tid, task_number, latest)
|
||
notified += 1
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"notify failed user=%s task=%s: %s",
|
||
max_user_id, task_number, exc,
|
||
)
|
||
logger.info(
|
||
"notifier: task=%s (#%s) new comment → notified %d users",
|
||
tid, task_number, notified,
|
||
)
|
||
await upsert_task_watch(tid, client_id, task_number, updatedat, latest["date"])
|
||
else:
|
||
# updatedat изменился (смена статуса и т.п.), новых комментариев нет
|
||
await upsert_task_watch(tid, client_id, task_number, updatedat, last_known)
|
||
|
||
|
||
async def _poll_once(bot) -> None:
|
||
# Один запрос — все открытые задачи
|
||
all_tasks = await fetch_all_open_tasks_lightweight()
|
||
|
||
# Группируем по clientid прямо в Python
|
||
tasks_by_client: dict[int, list[dict]] = defaultdict(list)
|
||
for task in all_tasks:
|
||
tasks_by_client[task["client_id"]].append(task)
|
||
|
||
# Маппинг intradesk_id → org_id из нашей БД
|
||
clients = await get_watched_clients()
|
||
client_to_org = {c["intradesk_id"]: c["org_id"] for c in clients}
|
||
|
||
logger.debug(
|
||
"notifier poll: %d tasks across %d clients",
|
||
len(all_tasks), len(tasks_by_client),
|
||
)
|
||
|
||
# Обрабатываем задачи по клиентам
|
||
for client_id, tasks in tasks_by_client.items():
|
||
org_id = client_to_org.get(client_id)
|
||
if org_id is None:
|
||
continue # клиент есть в Intradesk, но не привязан к нашей БД
|
||
|
||
task_ids_seen = {t["id"] for t in tasks}
|
||
|
||
for task in tasks:
|
||
try:
|
||
await _process_task(task, org_id, bot)
|
||
except Exception as exc:
|
||
logger.exception(
|
||
"notifier error task=%s client=%s: %s", task["id"], client_id, exc
|
||
)
|
||
|
||
# Убираем из слежения задачи, пропавшие из открытых
|
||
await remove_tasks_not_in(client_id, task_ids_seen)
|
||
|
||
# Клиенты из нашей БД, у которых сейчас 0 открытых задач
|
||
for c in clients:
|
||
if c["intradesk_id"] not in tasks_by_client:
|
||
await remove_tasks_not_in(c["intradesk_id"], set())
|
||
|
||
|
||
async def run_notifier(bot) -> None:
|
||
"""Запустить фоновый опрос. Добавлен в asyncio.gather в main.py."""
|
||
interval = config.NOTIFICATION_POLL_INTERVAL
|
||
if not interval or interval <= 0:
|
||
logger.info("Notifier disabled (NOTIFICATION_POLL_INTERVAL=0)")
|
||
return
|
||
|
||
logger.info("Notifier started, interval=%ds", interval)
|
||
# Пауза перед первым прогоном — даём боту и БД время инициализироваться
|
||
await asyncio.sleep(30)
|
||
while True:
|
||
try:
|
||
await _poll_once(bot)
|
||
except Exception as exc:
|
||
logger.exception("notifier _poll_once failed: %s", exc)
|
||
await asyncio.sleep(interval)
|