fix: 看板4项体验修复 — 时间/身份证/日期筛选/通知跳转

1. 完整16位产品身份证号显示(不再截断)
2. 日期筛选栏: 今天 | 近7天 | 近30天 | 全部
   - 后端 recent-activity 支持 since/until ISO参数
   - 默认展示今天动态,可按时间范围切换
3. 操作人兜底: TaskLog无operator_id时用任务assignee_id
4. 未读通知可点击跳转通知页面 + 快捷入口增加通知中心
This commit is contained in:
2026-08-12 13:35:23 +08:00
parent c3f3a5e291
commit 00fae01eed
4 changed files with 151 additions and 69 deletions

View File

@ -1,4 +1,5 @@
"""Dashboard 统计服务"""
from datetime import datetime
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from pydantic import BaseModel
@ -69,31 +70,43 @@ async def get_dashboard_stats(db: AsyncSession) -> DashboardStats:
)
async def get_recent_activity(db: AsyncSession, limit: int = 10) -> list[RecentActivity]:
async def get_recent_activity(
db: AsyncSession, limit: int = 10,
since: datetime | None = None, until: datetime | None = None,
) -> list[RecentActivity]:
from app.models.task_log import TaskLog
from app.models.task import Task
from app.models.product import Product
from app.core.time_utils import BEIJING_TZ
stmt = (
select(TaskLog, Task.task_name, Product.serial_number)
select(TaskLog, Task.task_name, Product.serial_number, Task.assignee_id)
.join(Task, TaskLog.task_id == Task.id)
.join(Product, Task.product_id == Product.id)
.order_by(TaskLog.created_at.desc())
.limit(limit)
)
if since:
stmt = stmt.where(TaskLog.created_at >= since)
if until:
stmt = stmt.where(TaskLog.created_at <= until)
stmt = stmt.order_by(TaskLog.created_at.desc()).limit(limit)
result = await db.execute(stmt)
rows = result.all()
# 收集 operator_id → 批量翻译中文姓名
operator_ids = list({row[0].operator_id for row in rows if row[0].operator_id})
# 收集 operator_id + 兜底 assignee_id → 批量翻译中文姓名
raw_ids: set[str] = set()
for row in rows:
op = row[0].operator_id
if op:
raw_ids.add(op)
elif row[3]: # assignee_id 兜底
raw_ids.add(row[3])
name_map: dict[str, str] = {}
if operator_ids:
if raw_ids:
from app.services.mom_cache import get_display_names
name_map = get_display_names(operator_ids)
name_map = get_display_names(list(raw_ids))
activities: list[RecentActivity] = []
for log, task_name, product_sn in rows:
for log, task_name, product_sn, task_assignee in rows:
action_label = _action_label(log.action_type)
# 强制转北京时间显示
t = log.created_at
@ -105,9 +118,9 @@ async def get_recent_activity(db: AsyncSession, limit: int = 10) -> list[RecentA
time_str = t.strftime("%m-%d %H:%M")
else:
time_str = ""
# operator_id: 优先显示中文姓名 → 英文用户名兜底 → 无记录时为
op = log.operator_id
op_display = name_map.get(op, op or "")
# operator_id: 优先显示中文姓名 → 英文用户名兜底 → 任务负责人兜底 →
op = log.operator_id or task_assignee or ""
op_display = name_map.get(op, op)
activities.append(RecentActivity(
action=action_label,
task_name=task_name or "",