Files
DadaoTask/backend/app/services/notify.py
T
2026-11-10 13:47:56 +08:00

126 lines
3.6 KiB
Python

"""站内通知(M2 起用,M4 会接入外部推送)。
三条原则:
1. **不通知自己**。绝大多数通知都是由自己的操作触发的,
不排除自己会让"未读"迅速变成一个没人看的数字。
2. **通知里写清"谁、做了什么、在哪件事上"**。只写"状态已更新"等于没写,
用户还得点进去猜。
3. **通知只是提示,不是凭证**。点开后仍要按 `item_id` 重新拉数据,
因为通知生成之后事情可能又被改过。
"""
from __future__ import annotations
import uuid
from collections.abc import Iterable
from sqlalchemy import select
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession
from ..models import Member
from ..models_ticket import Notification
TICKET_STATUS_CN = {
"open": "待接收",
"accepted": "待办理",
"in_progress": "办理中",
"pending_verify": "待验收",
"done": "已完成",
"canceled": "已取消",
}
TASK_STATUS_CN = {
"inbox": "收件箱",
"todo": "待办",
"doing": "进行中",
"blocked": "被阻塞",
"done": "已完成",
"canceled": "已取消",
}
async def notify(
session: AsyncSession,
*,
ws_id: uuid.UUID,
user_ids: Iterable[uuid.UUID | None],
kind: str,
item_id: uuid.UUID | None,
title: str,
body: str = "",
actor_id: uuid.UUID | None = None,
dedupe_key: str | None = None,
) -> int:
"""给若干人各写一条通知,返回实际写入条数。
去重与"跳过自己"在这里统一做一次——每个调用点都手写一遍迟早漏一个。
"""
seen: set[uuid.UUID] = set()
count = 0
for user_id in user_ids:
if user_id is None or user_id == actor_id or user_id in seen:
continue
seen.add(user_id)
session.add(
Notification(
workspace_id=ws_id,
user_id=user_id,
kind=kind,
item_id=item_id,
title=title[:120],
body=body[:300],
dedupe_key=dedupe_key,
)
)
count += 1
return count
async def notify_once(
session: AsyncSession,
*,
ws_id: uuid.UUID,
user_id: uuid.UUID | None,
kind: str,
item_id: uuid.UUID | None,
title: str,
body: str = "",
dedupe_key: str,
actor_id: uuid.UUID | None = None,
) -> bool:
"""写一条带幂等键的通知;已经写过就跳过,返回是否真的写入。
后台提醒循环会反复扫同一批任务,靠 `dedupe_key` 的唯一索引把重复挡在数据库层,
比"先查再插"少一次竞态。提醒本身不是业务事实,丢了不致命,重复才烦人。
"""
if user_id is None or user_id == actor_id:
return False
result = await session.execute(
pg_insert(Notification)
.values(
workspace_id=ws_id,
user_id=user_id,
kind=kind,
item_id=item_id,
title=title[:120],
body=body[:300],
dedupe_key=dedupe_key,
)
.on_conflict_do_nothing()
)
return result.rowcount > 0
async def workspace_admins(session: AsyncSession, ws_id: uuid.UUID) -> list[uuid.UUID]:
"""工作区里能"发布/把关"的人:owner + admin,只算正常成员。"""
rows = await session.execute(
select(Member.user_id).where(
Member.workspace_id == ws_id,
Member.status == "active",
Member.role.in_(("owner", "admin")),
)
)
return [row[0] for row in rows]