320 lines
9.5 KiB
Python
320 lines
9.5 KiB
Python
"""Центр уведомлений зоотехника: хранение 30 дней, список по дням, прочитано."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import date, datetime, time, timedelta, timezone
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
from sqlalchemy import case, func, or_, select
|
||
|
||
from app import db
|
||
from app.models.zootech_notification import ZootechNotification
|
||
from app.timeutil import (
|
||
display_calendar_today,
|
||
display_timezone_name,
|
||
naive_utc_to_iso,
|
||
utc_now_naive,
|
||
zoneinfo_or_utc,
|
||
)
|
||
|
||
RETENTION_DAYS = 30
|
||
DEFAULT_LIST_LIMIT = 200
|
||
|
||
|
||
def format_detail_timestamp(dt: Optional[datetime] = None) -> str:
|
||
"""Человекочитаемая дата/время для detail (ru-RU, Москва)."""
|
||
value = dt or utc_now_naive()
|
||
if value.tzinfo is None:
|
||
value = value.replace(tzinfo=timezone.utc)
|
||
try:
|
||
local = value.astimezone(zoneinfo_or_utc("Europe/Moscow"))
|
||
except Exception:
|
||
local = value
|
||
months = (
|
||
"января",
|
||
"февраля",
|
||
"марта",
|
||
"апреля",
|
||
"мая",
|
||
"июня",
|
||
"июля",
|
||
"августа",
|
||
"сентября",
|
||
"октября",
|
||
"ноября",
|
||
"декабря",
|
||
)
|
||
month = months[local.month - 1]
|
||
return f"{local.day} {month} {local.year}, {local.strftime('%H:%M')}"
|
||
|
||
|
||
def purge_expired(retention_days: int = RETENTION_DAYS) -> int:
|
||
cutoff = utc_now_naive() - timedelta(days=retention_days)
|
||
rows = (
|
||
db.session.execute(
|
||
select(ZootechNotification).where(ZootechNotification.created_at < cutoff)
|
||
)
|
||
.scalars()
|
||
.all()
|
||
)
|
||
count = len(rows)
|
||
for row in rows:
|
||
db.session.delete(row)
|
||
if count:
|
||
db.session.commit()
|
||
return count
|
||
|
||
|
||
def _normalize_kind(kind: Optional[str]) -> str:
|
||
if kind in ("success", "error", "warning", "info"):
|
||
return kind
|
||
return "info"
|
||
|
||
|
||
def create_notification(
|
||
*,
|
||
title: str,
|
||
detail: str,
|
||
kind: str = "info",
|
||
category: str = "general",
|
||
page: Optional[str] = None,
|
||
link_kind: Optional[str] = None,
|
||
link_id: Optional[str] = None,
|
||
user_login: Optional[str] = None,
|
||
) -> ZootechNotification:
|
||
purge_expired()
|
||
row = ZootechNotification(
|
||
title=(title or "")[:200],
|
||
detail=detail or "",
|
||
kind=_normalize_kind(kind),
|
||
category=(category or "general")[:40],
|
||
page=(page or None),
|
||
link_kind=(link_kind or None),
|
||
link_id=(link_id or None),
|
||
user_login=(user_login or None),
|
||
created_at=utc_now_naive(),
|
||
)
|
||
db.session.add(row)
|
||
db.session.commit()
|
||
return row
|
||
|
||
|
||
def _serialize(row: ZootechNotification) -> Dict[str, Any]:
|
||
return {
|
||
"id": row.id,
|
||
"title": row.title,
|
||
"detail": row.detail,
|
||
"kind": row.kind,
|
||
"category": row.category,
|
||
"page": row.page,
|
||
"linkKind": row.link_kind,
|
||
"linkId": row.link_id,
|
||
"read": row.read_at is not None,
|
||
"readAt": naive_utc_to_iso(row.read_at) if row.read_at else None,
|
||
"createdAt": naive_utc_to_iso(row.created_at),
|
||
}
|
||
|
||
|
||
def _display_tz():
|
||
try:
|
||
return zoneinfo_or_utc(display_timezone_name())
|
||
except Exception:
|
||
return timezone.utc
|
||
|
||
|
||
def _parse_calendar_date(value: Optional[str]) -> date:
|
||
if value:
|
||
try:
|
||
return date.fromisoformat(str(value).strip()[:10])
|
||
except ValueError:
|
||
pass
|
||
return display_calendar_today()
|
||
|
||
|
||
def _day_bounds_utc(target: date) -> Tuple[datetime, datetime]:
|
||
tz = _display_tz()
|
||
start_local = datetime.combine(target, time.min, tzinfo=tz)
|
||
end_local = datetime.combine(target, time.max, tzinfo=tz)
|
||
start_utc = start_local.astimezone(timezone.utc).replace(tzinfo=None)
|
||
end_utc = end_local.astimezone(timezone.utc).replace(tzinfo=None)
|
||
return start_utc, end_utc
|
||
|
||
|
||
def _retention_cutoff() -> datetime:
|
||
return utc_now_naive() - timedelta(days=RETENTION_DAYS)
|
||
|
||
|
||
def _daily_plan_match_clause():
|
||
"""План на день: явная категория + старые toast-записи с category=recipe."""
|
||
text = (
|
||
"%из плана%",
|
||
"%в плане%",
|
||
"%замен%",
|
||
"%исключ%",
|
||
"%вернул%",
|
||
"%снова в план%",
|
||
)
|
||
|
||
def _text_matches(col):
|
||
return or_(*[col.ilike(p) for p in text])
|
||
|
||
return or_(
|
||
ZootechNotification.category == "daily_plan",
|
||
ZootechNotification.page == "daily_plan",
|
||
ZootechNotification.link_kind == "daily_plan",
|
||
_text_matches(ZootechNotification.title),
|
||
_text_matches(ZootechNotification.detail),
|
||
)
|
||
|
||
|
||
def _apply_category_filter(stmt, category: Optional[str]):
|
||
cat = (category or "").strip().lower()
|
||
if cat == "daily_plan":
|
||
return stmt.where(_daily_plan_match_clause())
|
||
if cat == "recipe":
|
||
return stmt.where(
|
||
ZootechNotification.category == "recipe",
|
||
~_daily_plan_match_clause(),
|
||
)
|
||
if cat and cat != "all":
|
||
return stmt.where(ZootechNotification.category == cat)
|
||
return stmt
|
||
|
||
|
||
def _apply_sort(stmt, sort: str):
|
||
sort_key = (sort or "newest").strip().lower()
|
||
if sort_key == "oldest":
|
||
return stmt.order_by(ZootechNotification.created_at.asc())
|
||
if sort_key == "unread_first":
|
||
return stmt.order_by(
|
||
ZootechNotification.read_at.is_(None).desc(),
|
||
ZootechNotification.created_at.desc(),
|
||
ZootechNotification.id.desc(),
|
||
)
|
||
if sort_key == "severity":
|
||
severity_rank = case(
|
||
(ZootechNotification.kind == "error", 0),
|
||
(ZootechNotification.kind == "warning", 1),
|
||
(ZootechNotification.kind == "info", 2),
|
||
(ZootechNotification.kind == "success", 3),
|
||
else_=4,
|
||
)
|
||
return stmt.order_by(
|
||
severity_rank,
|
||
ZootechNotification.created_at.desc(),
|
||
ZootechNotification.id.desc(),
|
||
)
|
||
return stmt.order_by(
|
||
ZootechNotification.created_at.desc(),
|
||
ZootechNotification.id.desc(),
|
||
)
|
||
|
||
|
||
def _count_unread(*, category: Optional[str] = None, unread_only: bool = False) -> int:
|
||
stmt = select(func.count()).select_from(ZootechNotification).where(
|
||
ZootechNotification.created_at >= _retention_cutoff()
|
||
)
|
||
if unread_only:
|
||
stmt = stmt.where(ZootechNotification.read_at.is_(None))
|
||
stmt = _apply_category_filter(stmt, category)
|
||
return int(db.session.execute(stmt).scalar() or 0)
|
||
|
||
|
||
def _has_notifications_on_date(target: date, *, category: Optional[str] = None) -> bool:
|
||
start_utc, end_utc = _day_bounds_utc(target)
|
||
stmt = select(func.count()).select_from(ZootechNotification).where(
|
||
ZootechNotification.created_at >= start_utc,
|
||
ZootechNotification.created_at <= end_utc,
|
||
ZootechNotification.created_at >= _retention_cutoff(),
|
||
)
|
||
stmt = _apply_category_filter(stmt, category)
|
||
return int(db.session.execute(stmt).scalar() or 0) > 0
|
||
|
||
|
||
def _find_prev_notification_date(
|
||
before: date,
|
||
*,
|
||
category: Optional[str] = None,
|
||
) -> Optional[date]:
|
||
retention_start = display_calendar_today() - timedelta(days=RETENTION_DAYS - 1)
|
||
cursor = before - timedelta(days=1)
|
||
while cursor >= retention_start:
|
||
if _has_notifications_on_date(cursor, category=category):
|
||
return cursor
|
||
cursor -= timedelta(days=1)
|
||
return None
|
||
|
||
|
||
def unread_count(*, category: Optional[str] = None) -> Dict[str, Any]:
|
||
purge_expired()
|
||
return {"items": [], "unreadCount": _count_unread(unread_only=True)}
|
||
|
||
|
||
def list_notifications(
|
||
*,
|
||
date: Optional[str] = None,
|
||
category: Optional[str] = None,
|
||
unread_only: bool = False,
|
||
sort: str = "newest",
|
||
summary: bool = False,
|
||
) -> Dict[str, Any]:
|
||
purge_expired()
|
||
target = _parse_calendar_date(date)
|
||
unread = _count_unread(unread_only=True)
|
||
|
||
if summary:
|
||
return {"items": [], "unreadCount": unread, "date": target.isoformat()}
|
||
|
||
start_utc, end_utc = _day_bounds_utc(target)
|
||
stmt = select(ZootechNotification).where(
|
||
ZootechNotification.created_at >= start_utc,
|
||
ZootechNotification.created_at <= end_utc,
|
||
ZootechNotification.created_at >= _retention_cutoff(),
|
||
)
|
||
if unread_only:
|
||
stmt = stmt.where(ZootechNotification.read_at.is_(None))
|
||
stmt = _apply_category_filter(stmt, category)
|
||
stmt = _apply_sort(stmt, sort)
|
||
rows = db.session.execute(stmt).scalars().all()
|
||
|
||
prev_date = _find_prev_notification_date(target, category=category)
|
||
return {
|
||
"items": [_serialize(r) for r in rows],
|
||
"unreadCount": unread,
|
||
"date": target.isoformat(),
|
||
"prevDate": prev_date.isoformat() if prev_date else None,
|
||
"hasMore": prev_date is not None,
|
||
}
|
||
|
||
|
||
def get_notification(notification_id: str) -> Optional[ZootechNotification]:
|
||
purge_expired()
|
||
return db.session.get(ZootechNotification, notification_id)
|
||
|
||
|
||
def mark_read(notification_id: str) -> Optional[ZootechNotification]:
|
||
row = get_notification(notification_id)
|
||
if row is None:
|
||
return None
|
||
if row.read_at is None:
|
||
row.read_at = utc_now_naive()
|
||
db.session.commit()
|
||
return row
|
||
|
||
|
||
def mark_all_read() -> int:
|
||
purge_expired()
|
||
rows = (
|
||
db.session.execute(
|
||
select(ZootechNotification).where(ZootechNotification.read_at.is_(None))
|
||
)
|
||
.scalars()
|
||
.all()
|
||
)
|
||
now = utc_now_naive()
|
||
for row in rows:
|
||
row.read_at = now
|
||
if rows:
|
||
db.session.commit()
|
||
return len(rows)
|