da28d0ac1c
- Playwright держит открытыми веб-версии Outlook и Teams (постоянный профиль, Xvfb) - watcher.js перехватывает уведомления клиента и изменения карточек в реальном времени - pipeline: дедуп, baseline, VIP/MUTE, оценка важности через Haiku (OpenRouter) - aiogram-бот: уведомления, дайджест, вопросы по журналу, /login с 2FA-кодом, ручное управление браузером - Docker/compose, systemd-юнит, e2e-тест на поддельных страницах Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0194buWJjpDjU52JPX7PjsMs
265 lines
11 KiB
Python
265 lines
11 KiB
Python
"""События из браузера → дедуп → правила → LLM-оценка → уведомление в Telegram."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import html
|
|
import logging
|
|
import re
|
|
import time
|
|
from collections import deque
|
|
from datetime import datetime, timedelta
|
|
from typing import Awaitable, Callable
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from .config import Settings
|
|
from .db import DB
|
|
from .llm import LLM
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
TIME_RE = re.compile(
|
|
r"^(\d{1,2}[:.]\d{2}(\s?[AP]M)?|\d{1,2}[./]\d{1,2}([./]\d{2,4})?|\d+\s?\S{0,5}|now|teraz|сейчас)$", re.I
|
|
)
|
|
ICON = {3: "🔴", 2: "🟡", 1: "⚪️", 0: "·"}
|
|
SRC_NAME = {"mail": "📧 Почта", "teams": "💬 Teams"}
|
|
BASELINE_SECONDS = 90
|
|
|
|
Say = Callable[..., Awaitable[None]]
|
|
|
|
|
|
def norm(s: str) -> str:
|
|
return re.sub(r"\s+", " ", s or "").strip().lower()
|
|
|
|
|
|
def sha(s: str) -> str:
|
|
return hashlib.sha1(s.encode()).hexdigest()[:16]
|
|
|
|
|
|
INLINE_TIME_RE = re.compile(r"\b\d{1,2}[:.]\d{2}(\s?[AP]M)?\b", re.I)
|
|
|
|
|
|
def stable_lines(lines: list[str]) -> list[str]:
|
|
"""Строки карточки без «тикающих» частей (время, «5 мин»)."""
|
|
cand = [INLINE_TIME_RE.sub("", l).strip() for l in lines if not TIME_RE.match(l.strip())]
|
|
cand = [l for l in cand if l]
|
|
long = [l for l in cand if len(l) > 15]
|
|
return long or cand
|
|
|
|
|
|
class Pipeline:
|
|
def __init__(self, s: Settings, db: DB, llm: LLM, say: Say):
|
|
self.s = s
|
|
self.db = db
|
|
self.llm = llm
|
|
self.say = say
|
|
self.q: asyncio.Queue[dict] = asyncio.Queue()
|
|
self.recent_texts: deque[tuple[float, str]] = deque(maxlen=300)
|
|
self.baseline_until: dict[str, float] = {}
|
|
self.tz = ZoneInfo(s.tz)
|
|
|
|
# ---------- вход ----------
|
|
async def submit(self, ev: dict) -> None:
|
|
await self.q.put(ev)
|
|
|
|
def _rules(self) -> str:
|
|
try:
|
|
return self.s.rules_path.read_text(encoding="utf-8")
|
|
except FileNotFoundError:
|
|
return ""
|
|
|
|
def _lists(self) -> tuple[list[str], list[str]]:
|
|
vip = [*self.s.vip, *self.db.get("vip", [])]
|
|
mute = [*self.s.mute, *self.db.get("mute", [])]
|
|
return [v.lower() for v in vip], [m.lower() for m in mute]
|
|
|
|
# ---------- нормализация ----------
|
|
def _to_item(self, ev: dict) -> dict | None:
|
|
src = ev.get("source")
|
|
if src not in ("mail", "teams"):
|
|
return None
|
|
kind = ev.get("kind")
|
|
if kind == "notification":
|
|
title, text = ev.get("title", ""), ev.get("text", "")
|
|
if not (title or text):
|
|
return None
|
|
bucket = int(time.time() // 600) # одинаковые уведомления в разные 10-минутки — разные события
|
|
return {"source": src, "kind": kind, "key": f"n:{sha(norm(title) + '|' + norm(text))}:{bucket}",
|
|
"sender": title, "title": "", "body": text, "fp": norm(text)[:80]}
|
|
if kind == "row":
|
|
lines = [l for l in ev.get("lines", []) if l]
|
|
if not lines and not ev.get("aria"):
|
|
return None
|
|
st = stable_lines(lines) or [ev.get("aria", "")]
|
|
key = f"r:{ev.get('id', '')}:{sha('|'.join(norm(l) for l in st))}"
|
|
sender = lines[0] if lines else ""
|
|
title = lines[1] if src == "mail" and len(lines) > 1 else ""
|
|
rest = [l for l in (lines[2:] if title else lines[1:]) if not TIME_RE.match(l)]
|
|
body = " | ".join(rest)
|
|
if ev.get("aria") and norm(ev["aria"]) not in norm(" ".join(lines)):
|
|
body += f"\n[aria] {ev['aria']}"
|
|
return {"source": src, "kind": kind, "key": key, "sender": sender, "title": title,
|
|
"body": body, "fp": norm(" ".join(lines))}
|
|
return None
|
|
|
|
def _is_dup(self, it: dict) -> bool:
|
|
now = time.time()
|
|
while self.recent_texts and now - self.recent_texts[0][0] > 900:
|
|
self.recent_texts.popleft()
|
|
fp = it["fp"]
|
|
for _, other in self.recent_texts:
|
|
if not other or not fp:
|
|
continue
|
|
short, long_ = (fp, other) if len(fp) <= len(other) else (other, fp)
|
|
if len(short) >= 12 and short[:60] in long_:
|
|
return True
|
|
return False
|
|
|
|
def _in_baseline(self, source: str) -> bool:
|
|
"""Первый запуск: всё, что уже было в списках, запоминаем молча."""
|
|
if self.db.get(f"baseline:{source}"):
|
|
return False
|
|
until = self.baseline_until.get(source)
|
|
if until is None:
|
|
self.baseline_until[source] = time.time() + BASELINE_SECONDS
|
|
log.info("Baseline для %s: %s сек молча запоминаю текущие списки", source, BASELINE_SECONDS)
|
|
return True
|
|
if time.time() < until:
|
|
return True
|
|
self.db.set(f"baseline:{source}", True)
|
|
return False
|
|
|
|
# ---------- основной цикл ----------
|
|
async def run(self) -> None:
|
|
while True:
|
|
first = await self.q.get()
|
|
batch = [first]
|
|
deadline = time.monotonic() + self.s.batch_window
|
|
while len(batch) < 20:
|
|
left = deadline - time.monotonic()
|
|
if left <= 0:
|
|
break
|
|
try:
|
|
batch.append(await asyncio.wait_for(self.q.get(), left))
|
|
except asyncio.TimeoutError:
|
|
break
|
|
try:
|
|
await self.process(batch)
|
|
except Exception:
|
|
log.exception("Ошибка обработки пачки")
|
|
|
|
async def process(self, events: list[dict]) -> None:
|
|
# уведомления раньше карточек — тогда карточка того же сообщения отсеется как дубль
|
|
events = sorted(events, key=lambda e: 0 if e.get("kind") == "notification" else 1)
|
|
fresh: list[tuple[int, dict]] = []
|
|
for ev in events:
|
|
it = self._to_item(ev)
|
|
if not it:
|
|
continue
|
|
if it["kind"] == "row" and self._in_baseline(it["source"]):
|
|
self.db.insert(it["source"], it["kind"], it["key"], it["sender"], it["title"], it["body"], status="baseline")
|
|
continue
|
|
dup = self._is_dup(it)
|
|
item_id = self.db.insert(it["source"], it["kind"], it["key"], it["sender"], it["title"], it["body"],
|
|
status="dup" if dup else "new")
|
|
if item_id is None:
|
|
continue # уже видели
|
|
self.recent_texts.append((time.time(), it["fp"]))
|
|
if not dup:
|
|
fresh.append((item_id, it))
|
|
if not fresh:
|
|
return
|
|
|
|
vip, mute = self._lists()
|
|
to_llm: list[tuple[int, dict]] = []
|
|
forced: dict[int, int] = {}
|
|
for item_id, it in fresh:
|
|
hay = norm(f"{it['sender']} {it['title']} {it['body']}")
|
|
if any(m in hay for m in mute):
|
|
self.db.update(item_id, level=0, status="skipped", summary="muted")
|
|
continue
|
|
if any(v in hay for v in vip):
|
|
forced[item_id] = 3
|
|
to_llm.append((item_id, it))
|
|
if not to_llm:
|
|
return
|
|
|
|
try:
|
|
verdicts = await self.llm.triage([it for _, it in to_llm], self._rules(), self.s.my_name)
|
|
by_i = {v["i"]: v for v in verdicts}
|
|
except Exception as e:
|
|
log.warning("LLM недоступна (%s) — шлю без фильтра", e)
|
|
by_i = {}
|
|
|
|
for i, (item_id, it) in enumerate(to_llm):
|
|
v = by_i.get(i, {"level": 2, "summary": "", "sender": ""})
|
|
level = max(v["level"], forced.get(item_id, 0))
|
|
summary = v.get("summary") or ""
|
|
sender = v.get("sender") or it["sender"]
|
|
if level >= self.s.notify_min:
|
|
await self._notify(it, level, summary, sender)
|
|
status = "sent"
|
|
elif level >= 1:
|
|
status = "digest"
|
|
else:
|
|
status = "skipped"
|
|
self.db.update(item_id, level=level, summary=summary, status=status)
|
|
|
|
# ---------- отправка ----------
|
|
def _quiet_now(self) -> bool:
|
|
if not self.s.quiet_hours:
|
|
return False
|
|
a, b = self.s.quiet_hours
|
|
h = datetime.now(self.tz).hour
|
|
return (a <= h or h < b) if a > b else (a <= h < b)
|
|
|
|
async def _notify(self, it: dict, level: int, summary: str, sender: str) -> None:
|
|
e = html.escape
|
|
head = f"{ICON.get(level, '')} <b>{SRC_NAME.get(it['source'], it['source'])}</b> · {e(sender[:80])}"
|
|
orig = " — ".join(x for x in (it["title"], it["body"].split("\n[aria]")[0]) if x)
|
|
if len(orig) > 600:
|
|
orig = orig[:599] + "…"
|
|
text = head
|
|
if summary:
|
|
text += f"\n{e(summary)}"
|
|
if orig and norm(orig) != norm(summary):
|
|
text += f"\n\n<blockquote>{e(orig)}</blockquote>"
|
|
silent = level < 3 and self._quiet_now()
|
|
await self.say(text, silent=silent)
|
|
|
|
# ---------- дайджест ----------
|
|
async def send_digest(self, manual: bool = False) -> None:
|
|
items = self.db.digest_pending()
|
|
if not items:
|
|
if manual:
|
|
await self.say("Дайджест пуст — ничего неважного не накопилось.")
|
|
return
|
|
e = html.escape
|
|
parts = [f"🗞 <b>Дайджест</b> ({len(items)})"]
|
|
for src in ("mail", "teams"):
|
|
group = [i for i in items if i.source == src]
|
|
if not group:
|
|
continue
|
|
parts.append(f"\n<b>{SRC_NAME[src]}</b>")
|
|
for it in group[-40:]:
|
|
t = datetime.fromtimestamp(it.created_at, self.tz).strftime("%d.%m %H:%M")
|
|
parts.append(f"• {t} {e((it.sender or '')[:40])}: {e(it.summary or it.short(150))}")
|
|
await self.say("\n".join(parts), silent=True)
|
|
for it in items:
|
|
self.db.update(it.id, status="digested")
|
|
|
|
async def digest_loop(self) -> None:
|
|
if not self.s.digest_time:
|
|
return
|
|
hh, mm = (int(x) for x in self.s.digest_time.split(":"))
|
|
while True:
|
|
now = datetime.now(self.tz)
|
|
target = now.replace(hour=hh, minute=mm, second=0, microsecond=0)
|
|
if target <= now:
|
|
target += timedelta(days=1)
|
|
await asyncio.sleep(max(5, (target - now).total_seconds()))
|
|
try:
|
|
await self.send_digest()
|
|
except Exception:
|
|
log.exception("digest")
|