|
|
# backend/app/services/triage.py
|
|
|
# 알림 트리아지: 발신자 규칙 + 가드 → 버킷 결정 + 이유(why/tone). '나중에' 반복 패턴 탐지.
|
|
|
from sqlmodel import Session, select
|
|
|
|
|
|
from ..models import Notification, NotificationGuard, NotifyBucket, SenderRule
|
|
|
|
|
|
|
|
|
def classify_notification(s: Session, *, src: str, sender: str, title: str) -> dict:
|
|
|
"""새 알림 1건의 버킷/이유 결정. 반환 {bucket, why, tone, held_by_rule}."""
|
|
|
# 1) 발신자 규칙 우선
|
|
|
for r in s.exec(select(SenderRule).order_by(SenderRule.sort_order)).all():
|
|
|
if sender and (sender in r.who or r.who in sender):
|
|
|
why = {
|
|
|
"now": f"규칙: {r.rule}",
|
|
|
"later": "다이제스트",
|
|
|
"held": f"규칙: {r.rule}",
|
|
|
}[r.bucket.value]
|
|
|
return {
|
|
|
"bucket": r.bucket.value,
|
|
|
"why": why,
|
|
|
"tone": r.tone,
|
|
|
"held_by_rule": r.id if r.bucket == NotifyBucket.held else "",
|
|
|
}
|
|
|
# 2) 가드(딥 워크) 중이면 긴급 외 보류 → later (캘린더 임박 알림은 통과)
|
|
|
guard = s.get(NotificationGuard, 1)
|
|
|
if guard and guard.on and src not in ("cal",):
|
|
|
return {"bucket": "later", "why": "집중 보호 중", "tone": "muted", "held_by_rule": ""}
|
|
|
# 3) 기본 — 직접 봐야 하는 src 는 now
|
|
|
if src in ("mail", "chat", "cal", "fin"):
|
|
|
return {"bucket": "now", "why": "기본", "tone": "blue", "held_by_rule": ""}
|
|
|
return {"bucket": "later", "why": "낮은 우선순위", "tone": "muted", "held_by_rule": ""}
|
|
|
|
|
|
|
|
|
def detect_defer_pattern(s: Session, threshold: int = 3) -> list[dict]:
|
|
|
"""'나중에'로 반복 미룬 발신자/앱 패턴 탐지 → 자동화 제안 후보."""
|
|
|
counts: dict[str, int] = {}
|
|
|
for n in s.exec(select(Notification).where(Notification.bucket == NotifyBucket.later)).all():
|
|
|
if n.why == "내가 미룸": # 사용자가 직접 미룬 것만 카운트
|
|
|
counts[n.app] = counts.get(n.app, 0) + 1
|
|
|
return [
|
|
|
{"app": app, "count": c, "offer": f"‘{app}’ 알림을 항상 저녁 다이제스트로 묶을까요?"}
|
|
|
for app, c in counts.items()
|
|
|
if c >= threshold
|
|
|
]
|