duxingchen/track
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: duxingchen:edd43fec292d23be72a4ae869dd63c7d43b72d79
Branches Tags
duxingchen:master duxingchen:feature/ai-audit-update duxingchen:1.0应用
..
compare: duxingchen:1.0应用
Branches Tags
duxingchen:feature/ai-audit-update duxingchen:1.0应用 duxingchen:master

1 Commits
edd43fec29 ... 1.0应用

Author SHA1 Message Date
duxingchen
d729a19b86 feat: MOM 撤回出库强制回滚通道 
MOM 把误点出库的设备物理回滚到仓库时，Track 被动跟随 MOM 的权威物理状态。

【撤回信号识别】
- action / event 里的 revoke / rollback / revert / cancel 子串匹配
  （MOM 侧字段命名尚未冻结，刻意宽松，避免对方改词就整条链路失联）
- 无显式标记但产品正处于「已出库」时，按"已发货设备收到入库回调 = 货回来了"
  隐式判定

【匹配放宽】
- 常规入库仍严格要求 current_location_id == virtual_warehouse
- 撤回信号、或 payload 带 serial 时才放行「已出库」产品 —— 出库回调已把
  location 置为 None，不放宽则撤回必然失配、静默返回 matched=False
- 刻意不给 sku 兜底也无条件放宽：同型号可能多台，放宽会误标到别的设备

【特权通道】
- 强制覆写 overall_status=已入库 + 位置回滚 virtual_warehouse，优先级高于
  task_service 的【绝对物理终态保护】。两者方向刻意相反：那套约束的是
  "车间内部流转不许用工序名抹掉物理终态"，而本接口是物理事实的权威来源。
  代码内已留醒目注释，防止后续维护者误加状态互斥校验
- 改用 sync_product_status 统一双字段同步（原先硬编码 status="ARCHIVED"
  绕过了 lifecycle.py 的约定），并把 status 变化一并计入 changed，
  避免"状态与位置本就正确时纠偏不提交"

【撤回留痕】
- 追加「撤回出库(重新入库)」主线节点。名称里的「入库」二字是必须保留的契约：
  product_service._has_warehouse_task 用子串判定仓库节点，若只有"出库"会让
  location==virtual_warehouse 的产品被注入假的「待仓库收货」虚拟节点，
  出现"已入库却在等收货"的自相矛盾

【重构】
- 抽出 _pick_warehouse_log_task / _match_inbound_product 复用，出库回调同步简化
 2026-09-17 17:42:06 +08:00
39 changed files with 40 additions and 3034 deletions
Expand all files Collapse all files

63
AGENTS.md
View File

@ -1,63 +0,0 @@
# AGENTS.md
本仓库（Track 生产流转系统）的工作笔记。仅在验证过之后才写入，避免传谣。
## 架构速览
- `backend/` FastAPI + SQLAlchemy 2.x(async) + Alembic，PostgreSQL。
- `frontend/` React 19 + Vite + antd + Tailwind。路由见 `src/App.tsx`，
管理端菜单见 `src/components/layout/AdminLayout.tsx`（`MENU` 数组）。
- 登录不走 Track 自己的用户表，而是**只读** MOM(KCGL) 的 `sys_user`：
`sys_user.username` 存 `"真实姓名/登录账号"`，`login()` 用
`WHERE username LIKE '%/<账号>'` 匹配，`display_name` 由 `/` 拆解得到。
MOM 连接配置在 `app/core/mom_database.py`（同步 psycopg2 引擎）。
## 本地起环境（关键，踩过的坑都在这）
1. **本机没有 Postgres 时需要先装**（容器内 `sudo` 可用）：
`sudo -n apt-get install -y --fix-missing postgresql postgresql-contrib`
然后 `sudo -n pg_ctlcluster 17 main start`。
本仓库不使用 pgvector，无需额外扩展。
2. **数据库端口与生产默认值不同**，必须用环境变量覆盖：
- `DATABASE_URL=postgresql+asyncpg://track:track_prod_2026@127.0.0.1:5432/track_production`
- `MOM_DB_HOST=127.0.0.1`、`MOM_DB_PORT=5432`
- `SECRET_KEY=<≥32 字符>`：`DEBUG=false` 时配置项会**拒绝**默认 SECRET_KEY
（见 `app/core/config.py` 的校验），不设会直接 import 失败。
3. 迁移：`cd backend && python3 -m alembic upgrade head`（没有全局 `alembic` 命令，
要用 `python3 -m alembic`）。校验纯 SQL 用 `alembic upgrade head --sql`。
4. 前端 proxy 指向 Docker 服务名 `backend:8000`。本机跑要么把
`127.0.0.1 backend` 写进 `/etc/hosts`，要么直接给
`VITE_API_BASE_URL=http://localhost:<port>/api/v1` 绕过 proxy。
注意 dev server 由 `basicSsl` 起 HTTPS，跨域需要后端
`CORS_ORIGINS` 加上 `https://localhost:1420`。
## 测试的坑（重要）
- **不要用 `starlette.testclient.TestClient` 测异步 SQLAlchemy 应用。**
它每个请求新建事件循环，而引擎是模块级单例、池里挂着 asyncpg 连接，
跨循环复用会报 `got Future attached to a different loop`，表现为随机 500。
正确做法：`httpx.AsyncClient(transport=httpx.ASGITransport(app=app))`
并在单个 `asyncio.run()` 里跑完全部请求。生产 uvicorn 单循环无此问题。
- 仓库目前**没有** pytest 基建，也没有前端测试脚本。
## 已知的待修问题（截至 1.0应用 分支）
- **读接口大面积未鉴权**（已实测，非推测）：无 token 直接 200 的包括
`/api/v1/users/`、全部 `/api/v1/dashboard/*`（含
`people-history/export` —— 匿名即可批量导出个人工时台账）、
`/api/v1/analytics/*`、`/api/v1/screen/*`、`/api/v1/orders/`。
写操作和 `/api/v1/tasks`、`/api/v1/products` 是有鉴权的。
新增接口请统一用 `app/core/deps.py` 的 `require_admin` / `require_roles`。
- 「管理员角色」这份规则此前散在 4 处（后端 `task_service`、`products.py` 内联、
前端 `constants/task.ts`、`AdminProductsPage` 内联），已因此出过事故。
**后端唯一事实来源是 `app/core/roles.py`，前端用 `constants/task.ts::isAdminRole`。**
新增判断不要手写 `===` 比较。
- `task_logs.task_id` 是 NOT NULL 外键，只能挂任务，不是通用审计。通用审计是
`audit_logs`（本轮新增，由 `app/core/audit_middleware.py` 自动采集）。
## 约定
- 时间统一北京时间（`app/core/time_utils.py`），库里存 timestamptz。
- 中文枚举标签尽量由服务端下发（如审计接口的 `module_label`/`action_label`），
避免前端再抄一份映射开始漂移。
- 本仓库的提交信息用中文，说明「为什么」而非「改了什么」。

8
backend/.env.example
View File

@ -3,11 +3,3 @@ SECRET_KEY=change-me-to-a-random-secret-key-in-production
ACCESS_TOKEN_EXPIRE_MINUTES=30
DEBUG=true
CORS_ORIGINS='["http://localhost:1420", "tauri://localhost"]'
# 日志：LOG_JSON=true 输出单行 JSON（便于采集），本地调试可设 false 换可读格式
LOG_LEVEL=INFO
LOG_JSON=true
# 错误追踪（可选）：填了 DSN 且已安装 sentry-sdk 才会启用，否则自动跳过
# SENTRY_DSN=
# SENTRY_TRACES_SAMPLE_RATE=0.1

81
backend/alembic/versions/j1k2l3m4n5o6_add_audit_logs.py
View File

@ -1,81 +0,0 @@
"""add_audit_logs
Revision ID: j1k2l3m4n5o6
Revises: i1j2k3l4m5n6
Create Date: 2026-09-21
操作审计日志表（audit_logs）
--------------------------
新增一张独立的审计表，用于记录 task_logs 覆盖不到的操作：
登录、导出、产品增删改、收口、权限/配置变更等与单个任务无关的动作。
为什么另起一张表而不复用 task_logs：
task_logs.task_id 是 NOT NULL 外键，只能挂在任务上，无法表达「张三导出了
产品清单」这类动作；且缺少来源 IP / UA / 结果状态等审计必需字段。
存量数据无需回填（本表从上线时刻开始记录）。
"""
from typing import Sequence, Union
import sqlalchemy as sa
from alembic import op
from sqlalchemy.dialects import postgresql
revision: str = "j1k2l3m4n5o6"
down_revision: Union[str, None] = "i1j2k3l4m5n6"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
"audit_logs",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, nullable=False),
# 操作人
sa.Column("user_id", sa.String(64), nullable=True, comment="操作人账号(逻辑外键→MOM)"),
sa.Column("display_name", sa.String(100), nullable=True, comment="操作人显示名"),
sa.Column("role", sa.String(50), nullable=True, comment="操作时角色快照"),
# 业务语义
sa.Column("action", sa.String(50), nullable=False, comment="动作"),
sa.Column("module", sa.String(50), nullable=False, comment="业务模块"),
sa.Column("target_type", sa.String(50), nullable=True),
sa.Column("target_id", sa.String(100), nullable=True),
sa.Column("target_name", sa.String(200), nullable=True),
sa.Column("details", postgresql.JSONB(), nullable=True, comment="变更详情"),
# 请求上下文
sa.Column("ip_address", sa.String(50), nullable=True),
sa.Column("user_agent", sa.String(500), nullable=True),
sa.Column("method", sa.String(10), nullable=True),
sa.Column("url", sa.String(500), nullable=True),
sa.Column("status_code", sa.Integer(), nullable=True),
sa.Column("error_message", sa.Text(), nullable=True),
# 与结构化日志对账
sa.Column("request_id", sa.String(64), nullable=True),
sa.Column(
"created_at",
sa.DateTime(timezone=True),
nullable=False,
server_default=sa.func.now(),
),
)
# 索引：按审计页最常用的检索维度建
op.create_index("ix_audit_logs_created_at", "audit_logs", ["created_at"])
op.create_index("ix_audit_logs_user_id", "audit_logs", ["user_id"])
op.create_index("ix_audit_logs_module", "audit_logs", ["module"])
op.create_index("ix_audit_logs_action", "audit_logs", ["action"])
op.create_index("ix_audit_logs_target_id", "audit_logs", ["target_id"])
op.create_index("ix_audit_logs_request_id", "audit_logs", ["request_id"])
# 组合索引：审计页默认「按时间倒序 + 按模块/动作过滤」
op.create_index("ix_audit_logs_module_created", "audit_logs", ["module", "created_at"])
def downgrade() -> None:
op.drop_index("ix_audit_logs_module_created", table_name="audit_logs")
op.drop_index("ix_audit_logs_request_id", table_name="audit_logs")
op.drop_index("ix_audit_logs_target_id", table_name="audit_logs")
op.drop_index("ix_audit_logs_action", table_name="audit_logs")
op.drop_index("ix_audit_logs_module", table_name="audit_logs")
op.drop_index("ix_audit_logs_user_id", table_name="audit_logs")
op.drop_index("ix_audit_logs_created_at", table_name="audit_logs")
op.drop_table("audit_logs")

53
backend/alembic/versions/k1l2m3n4o5p6_add_user_daily_seen.py
View File

@ -1,53 +0,0 @@
"""add_user_daily_seen
Revision ID: k1l2m3n4o5p6
Revises: j1k2l3m4n5o6
Create Date: 2026-09-21
每日用户活动表（user_daily_seen）
--------------------------------
一天一人一行，记录当天首次 / 末次活动时刻，供日活报表计算
「上线时间 / 下线时间」。
为什么不复用 audit_logs：
· 上线/下线时间不能取登录时间 —— Refresh Token 有效期 7 天，用户不必每天
重新登录，「登录次数 0 却操作 35 次」的报表没有意义。
· 也不能只取写操作时间 —— 审计中间件只记写操作，普通 GET 不入账，
当天只翻看的人会被漏掉。
· 更不能把活动写进审计表 —— 「末次活动」是需要不断 UPDATE 的状态，
而审计流水必须只增不改；能改的审计记录等于没有审计价值。
存量数据无需回填：本表从上线时刻开始记录；日活接口对更早的日期会自动
回退到审计表的写操作时间去推算。
"""
from typing import Sequence, Union
import sqlalchemy as sa
from alembic import op
revision: str = "k1l2m3n4o5p6"
down_revision: Union[str, None] = "j1k2l3m4n5o6"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
"user_daily_seen",
sa.Column("user_id", sa.String(64), primary_key=True,
comment="操作人账号(逻辑外键→MOM)"),
sa.Column("day", sa.Date(), primary_key=True,
comment="北京时间自然日"),
sa.Column("first_seen_at", sa.DateTime(timezone=True), nullable=False,
comment="当天首次活动时刻"),
sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=False,
comment="当天末次活动时刻"),
)
# 日活查询按日期区间扫，给 day 单独建索引。
# （主键是 (user_id, day)，前缀是 user_id，按 day 过滤用不上，故需补一条）
op.create_index("ix_user_daily_seen_day", "user_daily_seen", ["day"])
def downgrade() -> None:
op.drop_index("ix_user_daily_seen_day", table_name="user_daily_seen")
op.drop_table("user_daily_seen")

292
backend/app/api/v1/endpoints/audit.py
View File

@ -1,292 +0,0 @@
"""审计日志 API —— 查看系统操作审计记录
与 MOM(KCGL) /audit/logs 的接口保持同构的筛选维度（操作人/模块/动作/目标/
时间区间），便于两端运维习惯统一；额外提供 request_id 筛选，可凭它直接跳到
结构化日志里的那一次请求。
另提供两个 CSV 导出端点（审计明细 / 日活统计），均支持按列导出。
"""
from __future__ import annotations
import csv
import io
from datetime import datetime, time, timedelta
from typing import Any, Callable
from fastapi import APIRouter, Depends, Query, Response
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.core.deps import require_admin
from app.core.time_utils import BEIJING_TZ, get_beijing_time
from app.schemas.audit import (
AuditLogListResponse,
AuditLogResponse,
AuditOption,
AuditOptionsResponse,
DailyUsageResponse,
DailyUsageRow,
)
from app.services import audit_service
from app.services.audit_service import ACTION_LABELS, MODULE_LABELS
router = APIRouter(prefix="/audit", tags=["审计日志"])
def _parse_day(value: str | None, *, end_of_day: bool = False) -> datetime | None:
"""解析 YYYY-MM-DD 为北京时间。
结束日期取次日 00:00 作为上界（配合 < 判断）—— 直接取当天 23:59:59 会
漏掉该秒内的记录，是日期区间筛选最常见的差一错误。
"""
if not value:
return None
try:
day = datetime.strptime(value, "%Y-%m-%d").date()
except ValueError:
return None
if end_of_day:
return datetime.combine(day + timedelta(days=1), time.min, tzinfo=BEIJING_TZ)
return datetime.combine(day, time.min, tzinfo=BEIJING_TZ)
@router.get("/logs", response_model=AuditLogListResponse)
async def get_audit_logs(
user_id: str | None = Query(None, description="操作人账号（模糊匹配）"),
module: str | None = Query(None, description="业务模块"),
action: str | None = Query(None, description="动作类型"),
target_id: str | None = Query(None, description="目标ID"),
request_id: str | None = Query(None, description="请求ID（与接口日志对账）"),
status_code: int | None = Query(None, description="响应状态码"),
start_date: str | None = Query(None, description="起始日期 YYYY-MM-DD"),
end_date: str | None = Query(None, description="结束日期 YYYY-MM-DD（含当天）"),
page: int = Query(1, ge=1),
page_size: int = Query(50, ge=1, le=200),
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_admin),
) -> AuditLogListResponse:
"""审计日志分页查询（按时间倒序）"""
start = _parse_day(start_date)
# 结束日期用「次日 00:00」作为开区间上界，避免漏掉当天最后几条
end_exclusive = _parse_day(end_date, end_of_day=True)
rows, total = await audit_service.list_audit_logs(
db,
user_id=user_id,
module=module,
action=action,
target_id=target_id,
request_id=request_id,
status_code=status_code,
start=start,
end=end_exclusive - timedelta(microseconds=1) if end_exclusive else None,
skip=(page - 1) * page_size,
limit=page_size,
)
items = []
for row in rows:
item = AuditLogResponse.model_validate(row)
# 中文标签由服务端补，避免前端为每个枚举再维护一份映射
item.module_label = MODULE_LABELS.get(row.module, row.module)
item.action_label = ACTION_LABELS.get(row.action, row.action)
items.append(item)
return AuditLogListResponse(items=items, total=total)
@router.get("/options", response_model=AuditOptionsResponse)
async def get_audit_options(
current_user: dict = Depends(require_admin),
) -> AuditOptionsResponse:
"""筛选项：模块与动作的中文下拉；顺带下发导出可选列"""
return AuditOptionsResponse(
modules=[AuditOption(value=k, label=v) for k, v in MODULE_LABELS.items()],
actions=[AuditOption(value=k, label=v) for k, v in ACTION_LABELS.items()],
log_export_columns=[
AuditOption(value=k, label=v[0]) for k, v in _AUDIT_LOG_COLUMNS.items()
],
usage_export_columns=[
AuditOption(value=k, label=v[0]) for k, v in _DAILY_USAGE_COLUMNS.items()
],
)
# ============================================================
# CSV 导出
# ============================================================
def _bj(dt: datetime | None) -> str:
"""时间列统一按北京时间输出（与列表页、日活分日口径一致）。
直接输出 UTC 会让导出文件里 01:00 的操作显示成前一天 17:00，
与网页上看到的对不上 —— 导出与页面不一致是最容易被质疑的那种问题。
"""
if dt is None:
return ""
return dt.astimezone(BEIJING_TZ).strftime("%Y-%m-%d %H:%M:%S")
def _actor(log) -> str:
"""操作人：优先中文名，退化为账号（与列表页的展示规则一致）"""
if not log.user_id and not log.display_name:
return "未认证"
return f"{log.display_name}({log.user_id})" if log.display_name else (log.user_id or "")
# 列定义：key → (表头, 取值函数)。
# 前端只传 key 列表，中文表头与取值口径都由后端统一维护，
# 避免两端各写一份导致"导出的列和页面上的对不上"。
_AUDIT_LOG_COLUMNS: dict[str, tuple[str, Callable[[Any], Any]]] = {
"time": ("时间", lambda r: _bj(r.created_at)),
"user": ("操作人", _actor),
"role": ("角色", lambda r: r.role or ""),
"module": ("模块", lambda r: MODULE_LABELS.get(r.module, r.module)),
"action": ("动作", lambda r: ACTION_LABELS.get(r.action, r.action)),
"method": ("方法", lambda r: r.method or ""),
"url": ("请求路径", lambda r: r.url or ""),
"status": ("结果", lambda r: r.status_code if r.status_code is not None else ""),
"ip": ("来源IP", lambda r: r.ip_address or ""),
"target": ("目标", lambda r: f"{r.target_type or ''}:{r.target_id or ''}".strip(":")),
"error": ("错误信息", lambda r: r.error_message or ""),
"request_id": ("请求ID", lambda r: r.request_id or ""),
"user_agent": ("User-Agent", lambda r: r.user_agent or ""),
}
_DAILY_USAGE_COLUMNS: dict[str, tuple[str, Callable[[dict], Any]]] = {
"day": ("日期", lambda r: r["day"]),
"user": ("操作人", lambda r: f"{r['display_name']}({r['user_id']})" if r["display_name"] else (r["user_id"] or "")),
"role": ("角色", lambda r: r["role"] or ""),
# 上线/下线时间 = 当天首次/末次活动（非登录时间），
# 登录/登出次数单独成列，两者不再混为一谈
"first_active": ("上线时间", lambda r: _bj(r["first_active_at"])),
"last_active": ("下线时间", lambda r: _bj(r["last_active_at"])),
"login_count": ("登录次数", lambda r: r["login_count"]),
"logout_count": ("登出次数", lambda r: r["logout_count"]),
"op_count": ("操作次数", lambda r: r["op_count"]),
}
def _csv_response(
columns: dict[str, tuple[str, Callable]], keys: list[str], rows: list, filename: str,
) -> Response:
"""把行数据渲染成 CSV 响应。
⚠️ 必须带 UTF-8 BOM：Excel 靠它识别编码，否则中文表头与内容全是乱码。
这是 CSV 导出最常见、也最容易被忽略的坑。
"""
buf = io.StringIO()
writer = csv.writer(buf)
writer.writerow([columns[k][0] for k in keys])
for row in rows:
writer.writerow([columns[k][1](row) for k in keys])
return Response(
content=b"\xef\xbb\xbf" + buf.getvalue().encode("utf-8"),
media_type="text/csv; charset=utf-8",
# 文件名用纯 ASCII：中文文件名要走 RFC 5987，各浏览器行为不一致，
# 内部系统没必要为它引入兼容成本。
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
)
def _resolve_keys(raw: str | None, columns: dict) -> list[str]:
"""解析前端传来的列 key。缺省 = 全部列；未知 key 直接忽略（不报错）。"""
if not raw:
return list(columns)
keys = [k.strip() for k in raw.split(",") if k.strip() in columns]
return keys or list(columns)
@router.get("/logs/export")
async def export_audit_logs(
user_id: str | None = Query(None, description="操作人账号（模糊匹配）"),
module: str | None = Query(None, description="业务模块"),
action: str | None = Query(None, description="动作类型"),
target_id: str | None = Query(None, description="目标ID"),
request_id: str | None = Query(None, description="请求ID"),
status_code: int | None = Query(None, description="响应状态码"),
start_date: str | None = Query(None, description="起始日期 YYYY-MM-DD"),
end_date: str | None = Query(None, description="结束日期 YYYY-MM-DD（含当天）"),
columns: str | None = Query(None, description="导出列，逗号分隔；缺省=全部"),
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_admin),
) -> Response:
"""审计明细 CSV 导出 —— 筛选维度与 /logs 完全一致，保证"看到什么就能导出什么"。"""
start = _parse_day(start_date)
end_exclusive = _parse_day(end_date, end_of_day=True)
rows, truncated = await audit_service.export_audit_logs(
db,
user_id=user_id, module=module, action=action, target_id=target_id,
request_id=request_id, status_code=status_code,
start=start,
end=end_exclusive - timedelta(microseconds=1) if end_exclusive else None,
)
keys = _resolve_keys(columns, _AUDIT_LOG_COLUMNS)
resp = _csv_response(_AUDIT_LOG_COLUMNS, keys, rows, "audit_logs.csv")
if truncated:
# 用响应头传递"已截断"，前端据此提示用户收窄筛选条件
resp.headers["X-Export-Truncated"] = "1"
resp.headers["X-Export-Max-Rows"] = str(audit_service.EXPORT_MAX_ROWS)
resp.headers["Access-Control-Expose-Headers"] = "X-Export-Truncated, X-Export-Max-Rows"
return resp
@router.get("/daily-usage/export")
async def export_daily_usage(
start_date: str | None = Query(None, description="起始日期 YYYY-MM-DD（北京时间），默认今天"),
end_date: str | None = Query(None, description="结束日期 YYYY-MM-DD（北京时间），默认同起始日"),
columns: str | None = Query(None, description="导出列，逗号分隔；缺省=全部"),
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_admin),
) -> Response:
"""日活统计 CSV 导出 —— 每人一行：上线/下线次数与时间、操作次数。"""
start = _parse_day(start_date) or datetime.combine(
get_beijing_time().date(), time.min, tzinfo=BEIJING_TZ,
)
end = _parse_day(end_date, end_of_day=True) or (start + timedelta(days=1))
items = await audit_service.get_daily_usage(db, start=start, end=end)
keys = _resolve_keys(columns, _DAILY_USAGE_COLUMNS)
return _csv_response(_DAILY_USAGE_COLUMNS, keys, items, "daily_usage.csv")
@router.get("/daily-usage", response_model=DailyUsageResponse)
async def get_daily_usage(
start_date: str | None = Query(None, description="起始日期 YYYY-MM-DD（北京时间），默认今天"),
end_date: str | None = Query(None, description="结束日期 YYYY-MM-DD（北京时间），默认同起始日"),
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_admin),
) -> DailyUsageResponse:
"""日活 / 使用统计 —— 按【北京时间自然日 × 操作人】聚合。
回答的是「每天有哪些人用了系统、用了多少」：
· 上线时间 / 下线时间：当天**首次 / 末次活动**时间（任意审计记录）
· 操作次数：当天该用户的全部审计记录数（使用深度）
· 登录次数 / 登出次数：真实的手动登录 / 登出行为计数
⚠️ 上线时间【不取登录时间】：token 有效期内（refresh 7 天）用户不重新登录，
按登录算会让「周一登录、周二继续用」的周二变成"登录次数 0、上线时间空，
但操作次数 35"——报表自相矛盾。改用活动口径后，当天的第一次操作即上线时间。
⚠️ 登出次数天然小于登录次数：用户直接关浏览器、断网、token 过期都不会
产生登出记录。这是真实情况，不做任何"补齐"推算。
"""
# 起始日：未传则取北京的今天。_parse_day 返回的是北京时间当日 00:00。
start = _parse_day(start_date) or datetime.combine(
get_beijing_time().date(), time.min, tzinfo=BEIJING_TZ,
)
# 结束日：_parse_day(end_of_day=True) 已给出「次日 00:00」，正好当作半开上界。
# 未传则默认单日查询（= 起始日当天）。
end = _parse_day(end_date, end_of_day=True) or (start + timedelta(days=1))
items = await audit_service.get_daily_usage(db, start=start, end=end)
return DailyUsageResponse(
start_date=start.astimezone(BEIJING_TZ).strftime("%Y-%m-%d"),
end_date=(end - timedelta(days=1)).astimezone(BEIJING_TZ).strftime("%Y-%m-%d"),
items=[DailyUsageRow(**row) for row in items],
total=len(items),
)

53
backend/app/api/v1/endpoints/auth.py
View File

@ -1,5 +1,5 @@
"""认证 API — 对接 MOM sys_user + 双 Token 刷新"""
from fastapi import APIRouter, Depends, Request
from fastapi import APIRouter, Depends
from app.schemas.user import (
LoginRequest,
LoginResponse,
@ -7,68 +7,23 @@ from app.schemas.user import (
RefreshResponse,
UserResponse,
)
from app.core.security import peek_token_identity
from app.services.auth_service import login, refresh_access_token, get_current_user
router = APIRouter(prefix="/auth", tags=["认证"])
@router.post("/login", response_model=LoginResponse)
def login_endpoint(data: LoginRequest, request: Request):
def login_endpoint(data: LoginRequest):
"""登录 — 验证 MOM sys_user 表，返回 Access + Refresh 双 Token"""
# 登录请求本身尚未认证，中间件拿不到操作人。但「谁在尝试登录、失败了多少次」
# 恰恰是审计里最该有的信息，所以在校验之前就把尝试的账号写进 state：
# 登录失败时同样留痕，且能按账号追踪暴力破解。
# 注意：绝不把 data.password 写进 state / 审计，密码不落库。
request.state.audit_user = data.username
result = login(data.username, data.password)
# 登录成功后补上显示名 / 角色 —— 否则审计里这条记录的「操作人」会退化成账号
# （前端按 display_name || user_id 渲染，见 AdminAuditLogPage）。
# 能在这里补的原因：中间件是在 call_next 返回【之后】才落库的，此刻写入
# request.state 依然会被采集到。
# 失败登录走不到这里，保持「只有账号可追责」——这正是想要的语义。
if result.user:
request.state.audit_display_name = result.user.display_name
request.state.audit_role = result.user.role
return result
return login(data.username, data.password)
@router.post("/refresh", response_model=RefreshResponse)
def refresh_endpoint(data: RefreshRequest, request: Request):
def refresh_endpoint(data: RefreshRequest):
"""刷新 Access Token — 使用 Refresh Token 换取新的 Access Token"""
# 本接口刻意不挂 get_current_user：能用到这里，正是因为 access token 已经
# 过期/缺失，请求里没有 Authorization 头，JWT 依赖不会执行 → 审计拿不到操作人，
# 记录只能显示「未认证」。
# 但 refresh token 里本来就带着完整身份（sub/username/display_name/role），
# 解出来写进 state，审计才能记到人 —— 而"谁在何时尝试刷新"正是要留痕的。
# 注意 peek 只用于审计标注，鉴权判断一律走 get_current_user。
identity = peek_token_identity(data.refresh_token)
if identity:
request.state.audit_user = identity.get("username") or identity.get("sub")
request.state.audit_display_name = identity.get("display_name") or ""
request.state.audit_role = identity.get("role") or ""
return refresh_access_token(data.refresh_token)
@router.post("/logout")
def logout_endpoint(current_user: dict = Depends(get_current_user)):
"""登出 —— 仅用于审计留痕。
JWT 是无状态的，服务端没有可吊销的会话，因此本接口**不做任何令牌失效**
（客户端清掉本地 token 即为登出），返回体也没有实际语义。
它存在的唯一目的：让审计中间件记下「谁在何时退出了系统」。
没有这个端点时，前端「退出」只清本地存储、不产生任何请求，
退出动作在审计里完全不可见 —— 而"谁在什么时候退掉了系统"
在追责场景下和"谁登录了"同等重要。
挂 Depends(get_current_user) 是为了让 JWT 依赖把操作人写进 request.state
（见 auth_service.get_current_user），记录到真实姓名而非「未认证」。
"""
return {"ok": True}
@router.get("/me", response_model=UserResponse)
def get_me(current_user: dict = Depends(get_current_user)):
"""获取当前用户信息（从 Access Token 解析）"""

1
backend/app/api/v1/endpoints/orders.py
View File

@ -19,7 +19,6 @@ async def list_orders(
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=200),
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
):
result = await db.execute(
select(ProductionOrder).offset(skip).limit(limit).order_by(ProductionOrder.created_at.desc())

16
backend/app/api/v1/endpoints/products.py
View File

@ -29,18 +29,6 @@ router = APIRouter(prefix="/products", tags=["产品管理"])
async def get_product_qrcode(serial_number: str):
"""
生成产品二维码（PNG 图片）。
⚠️ 本接口【刻意不加鉴权】：
前端以 `<img src="/api/v1/products/qrcode/{sn}">` 引用它，而 <img>
无法携带 Authorization 头 —— 加了鉴权会让所有二维码图片加载失败，
并在审计里刷出大量 401。
不加鉴权是安全的：本函数**不查数据库**，只校验长度并把这个字符串渲染成
二维码，没有任何业务数据泄露面（序列号本身就是调用方提供的）。
也刻意不支持 ?token= 兜底：把 JWT 放进 URL 会渗进访问日志、浏览器历史
与 Referer，比它想解决的问题更糟。
内容为 16 位序列号，扫描后可调用 /scan/{serial_number} 查询产品。
尺寸：300×300 px，用于 PC 端打印或嵌入标签。
"""
@ -62,7 +50,6 @@ async def get_product_qrcode(serial_number: str):
async def scan_product(
serial_number: str,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
):
"""
扫码接口：根据 16 位序列号查询产品及其当前进度。
@ -82,7 +69,6 @@ async def list_products(
keyword: str | None = Query(None, description="多维搜索: 产品身份证/订单号/规格型号"),
status: str | None = Query(None, description="产品状态筛选: PENDING/WIP/COMPLETED/ARCHIVED"),
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
):
"""获取产品列表 — 支持 keyword 搜索 + 状态筛选"""
return await product_service.get_all_products(
@ -94,7 +80,6 @@ async def list_products(
async def get_product(
product_id: str,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
):
"""获取单个产品详情"""
import uuid
@ -211,7 +196,6 @@ class MessageCreate(BaseModel):
async def get_product_messages(
product_id: str,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
):
"""获取某产品的所有留言（按时间正序）"""
result = await db.execute(

3
backend/app/api/v1/endpoints/tasks.py
View File

@ -38,7 +38,6 @@ async def list_tasks(
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=200),
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
):
"""获取任务列表，可按产品/负责人筛选（只返回顶层任务）"""
pid = uuid.UUID(product_id) if product_id else None
@ -49,7 +48,6 @@ async def list_tasks(
async def get_task(
task_id: str,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
):
"""
获取任务详情 — 递归包含所有层级的子任务。
@ -301,7 +299,6 @@ async def create_subtask_endpoint(
async def get_tasks_by_product(
product_id: str,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
):
"""获取指定产品的顶层任务列表（不含子任务嵌套）"""
return await task_service.get_top_level_tasks(db, uuid.UUID(product_id))

82
backend/app/api/v1/endpoints/webhooks.py
View File

@ -11,7 +11,7 @@ from __future__ import annotations
from datetime import datetime
from fastapi import APIRouter, Depends, Header, HTTPException, Request
from fastapi import APIRouter, Depends, Header, HTTPException
from pydantic import BaseModel
from sqlalchemy import or_, select
from sqlalchemy.ext.asyncio import AsyncSession
@ -36,7 +36,6 @@ class MomInboundPayload(BaseModel):
event: str | None = None # 事件名，如 inbound.created / outbound.revoked
action: str | None = None # 显式动作指令，如 revoke_outbound
source_table: str | None = None # stock_product / stock_semi
company_name: str | None = None # 目标公司（IRIS / LICA），MOM 据此分流到不同 Track 实例
# 「撤回出库」信号词 —— 只在 action / event 里做子串匹配。
@ -45,56 +44,6 @@ class MomInboundPayload(BaseModel):
_OUTBOUND_REVOKE_TOKENS = ("revoke", "rollback", "revert", "cancel")
# ── 公司归属分流 ──────────────────────────────────────────────────────────
# MOM 现在会在载荷里带 company_name，同一套物理库可能同时向多个 Track 实例
# （IRIS / LICA）回调。本实例服务的是 IRIS，故只放行 IRIS 与空白值。
#
# ⚠️ 判定刻意做成「只排除已知的外来公司」，而非「白名单只认 IRIS」：
# MOM 在无法确定公司归属时会回落到扁平配置，该配置指向本实例 —— 这类
# 消息的 company_name 会是空 / 缺失。若此处按白名单把空白也拒掉，它们
# 就彻底丢了：MOM 那边已收到 200、认为投递成功，不会再重推。
# 同理，未见过的新值（不是 IRIS 也不是 LICA）也一律照常处理。
_FOREIGN_COMPANIES = {"LICA"}
def _is_foreign_company(company_name: str | None) -> bool:
"""载荷是否属于本实例不该处理的其它公司。
返回 True 表示应原样忽略（仍回 200，避免 MOM 反复重推）。
"""
# 大小写 / 首尾空白都容忍：MOM 侧常量书写方式未必冻结，误判的代价是
# 一条消息被错误地当成本公司处理（有唯一匹配约束，最坏是 matched=False）。
return (company_name or "").strip().upper() in _FOREIGN_COMPANIES
def _attribute_audit_to_mom_operator(request: Request, operator: str | None) -> None:
"""把外部回调归因到 MOM 侧的实际操作人。
外部回调走 X-API-Key 鉴权、没有 JWT，所以 JWT 依赖不执行，
审计中间件读到的 request.state.audit_user 永远是空 ——
操作审计里就出现一堆没有归属的「外部系统对接」记录。
但 MOM 载荷里本来就带着实际操作人（operator，即 MOM 侧扫码的那位），
写进 request.state 即可让审计归因到人。
⚠️ 必须在 X-API-Key 校验【之后】调用：密钥不对说明载荷本身就不可信，
此时把 operator 写进审计等于允许伪造人。
"""
who = (operator or "").strip()
if not who:
# 取不到操作人时留一个明确的系统标记，而不是继续显示「未认证」——
# 「MOM系统」至少说明这是一次机器回调，不是"一个匿名的人"。
request.state.audit_user = "MOM系统"
return
request.state.audit_user = who
try:
# 尽力而为：查不到中文名也不影响审计（前端会回退显示账号）
from app.services.mom_cache import get_display_names
request.state.audit_display_name = get_display_names([who]).get(who) or ""
except Exception: # noqa: BLE001 —— 姓名解析失败绝不能影响回调处理
pass
def _is_outbound_revoke(payload: MomInboundPayload) -> bool:
"""payload 是否携带**显式**的撤回出库信号。
@ -178,7 +127,6 @@ async def _match_inbound_product(
@router.post("/mom-inbound")
async def mom_inbound_webhook(
payload: MomInboundPayload,
request: Request,
x_api_key: str | None = Header(default=None, alias="X-API-Key"),
db: AsyncSession = Depends(get_db),
) -> dict:
@ -188,23 +136,12 @@ async def mom_inbound_webhook(
- 常规入库：用 serial_number（优先）或 sku 匹配「当前位于 virtual_warehouse」
的产品，命中则标记"已实收"（overall_status=已入库 + 记录 task_logs）。
- 撤回出库：MOM 把误出库的设备物理回滚到仓库 → 本接口强制执行特权回滚。
- 公司归属：company_name 明确写着其它公司（LICA）时原样忽略；空白 / 缺失
一律照常处理（见 _is_foreign_company 的说明）。
- 未命中返回 200（MOM 可能操作了非 Track 生产的物料，直接忽略）。
"""
# ── 鉴权 ──
if not settings.TRACK_WEBHOOK_KEY or x_api_key != settings.TRACK_WEBHOOK_KEY:
raise HTTPException(status_code=401, detail="Unauthorized: invalid X-API-Key")
# ── 公司归属：不是本实例的消息原样忽略（仍回 200，避免 MOM 当作失败而重推） ──
# ⚠️ 键名 reason / 取值 "ignored_company" 与 LICA 实例（~/track-lica）保持一致：
# MOM 侧不解析它，但排查时两边日志对着看，字段名不一致会白白浪费时间。
if _is_foreign_company(payload.company_name):
return {"ok": True, "matched": False, "reason": "ignored_company"}
# 归因到 MOM 侧实际扫码的人（必须在鉴权通过之后，见函数注释）
_attribute_audit_to_mom_operator(request, payload.operator)
explicit_revoke = _is_outbound_revoke(payload)
# ── 匹配产品 ──
@ -392,17 +329,11 @@ class MomOutboundPayload(BaseModel):
sku: str | None = None # 规格型号 spec_model（serial 缺失时的兜底匹配）
operator: str | None = None # 出库操作人（写入 task_logs.operator_id）
outbound_time: datetime | None = None # 出库时间
company_name: str | None = None # 目标公司（IRIS / LICA），MOM 据此分流到不同 Track 实例
# ↓ 2026-09 新增：MOM 一直在发、此前被 Pydantic 静默丢弃。当前无人读取，
# 先接住是为了与 LICA 实例（~/track-lica）对同一载荷的解析结果保持一致 ——
# 否则将来谁写了读这个字段的代码，会在 LICA 拿到值、在本实例拿到 None。
outbound_type: str | None = None # SALES / USE / PRODUCTION
@router.post("/mom-outbound")
async def mom_outbound_webhook(
payload: MomOutboundPayload,
request: Request,
x_api_key: str | None = Header(default=None, alias="X-API-Key"),
db: AsyncSession = Depends(get_db),
) -> dict:
@ -411,23 +342,12 @@ async def mom_outbound_webhook(
- 鉴权：Header X-API-Key 必须等于环境变量 TRACK_WEBHOOK_KEY。
- 用 serial_number（优先）或 sku 匹配"在仓库/已入库"的产品；
命中则标记"已出库"（overall_status=已出库 + status=OUTBOUND + 记录 task_logs）。
- 公司归属：company_name 明确写着其它公司（LICA）时原样忽略；空白 / 缺失
一律照常处理（见 _is_foreign_company 的说明）。
- 未命中返回 200（MOM 出库的可能是非 Track 生产的物料，直接忽略）。
"""
# ── 鉴权 ──
if not settings.TRACK_WEBHOOK_KEY or x_api_key != settings.TRACK_WEBHOOK_KEY:
raise HTTPException(status_code=401, detail="Unauthorized: invalid X-API-Key")
# ── 公司归属：不是本实例的消息原样忽略（仍回 200，避免 MOM 当作失败而重推） ──
# ⚠️ 键名 reason / 取值 "ignored_company" 与 LICA 实例（~/track-lica）保持一致：
# MOM 侧不解析它，但排查时两边日志对着看，字段名不一致会白白浪费时间。
if _is_foreign_company(payload.company_name):
return {"ok": True, "matched": False, "reason": "ignored_company"}
# 归因到 MOM 侧实际出库的人（必须在鉴权通过之后，见函数注释）
_attribute_audit_to_mom_operator(request, payload.operator)
# ── 按 serial_number（优先）或 sku 匹配"在仓库/已入库"的产品 ──
product = None
where_cond = or_(

2
backend/app/api/v1/router.py
View File

@ -17,7 +17,6 @@ from app.api.v1.endpoints.holidays import router as holidays_router
from app.api.v1.endpoints.webhooks import router as webhooks_router
from app.api.v1.endpoints.external_products import router as external_products_router
from app.api.v1.endpoints.screen import router as screen_router
from app.api.v1.endpoints.audit import router as audit_router
api_router = APIRouter()
@ -38,4 +37,3 @@ api_router.include_router(holidays_router)
api_router.include_router(webhooks_router)
api_router.include_router(external_products_router)
api_router.include_router(screen_router)
api_router.include_router(audit_router)

243
backend/app/core/audit_middleware.py
View File

@ -1,243 +0,0 @@
"""审计采集中间件
在响应生成后，把「谁 / 何时 / 从哪来 / 调了哪个接口 / 做了什么 / 结果如何」
落进 audit_logs。
为什么用中间件自动采集，而不是在每个业务函数里手写 record_audit
------------------------------------------------------------------
1. 手写必然漏。新加的端点很容易忘记补审计，而审计的价值恰恰建立在「完整」上。
现状可佐证：task_logs 全项目只有 4 处写入点，凡是不挂在任务上的动作
（登录、导出、改产品）全都没有留痕。
2. 中间件能拿到业务函数拿不到的事实：真实来源 IP、UA、最终状态码、
以及与结构化日志对齐的 request_id。
3. 业务语义（module / target）由路径推导，不如手写精确，但对「谁动了什么」
的追责场景已经够用；关键动作后续可再调 record_audit 补 details 做增强。
采集范围
--------
- 所有写操作（POST/PUT/PATCH/DELETE）
- 少数**读但敏感**的操作：导出、下载、打印（本项目 GET /people-history/export
就是导出，只按方法过滤会漏掉）
明确不采集：GET /health*、/docs、/openapi.json —— 探针与文档的噪声没有审计价值。
"""
from __future__ import annotations
import logging
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.responses import Response
from app.services.audit_service import record_audit
logger = logging.getLogger("track.audit")
# 写操作一律采集
_MUTATING_METHODS = frozenset({"POST", "PUT", "PATCH", "DELETE"})
# 读操作里需要留痕的（导出/下载/打印属于「读」，但把数据带出了系统）
_SENSITIVE_READ_KEYWORDS = frozenset({"export", "download", "print"})
# 核心业务模块 —— 这些前缀下的「查看详情」GET 也采集，
# 用于回答「谁在什么时候看过哪条业务数据」，而不只是「谁改过」。
#
# ⚠️ 只覆盖【核心业务实体】：
# products —— 移动端扫码查询 GET /products/scan/{sn} 是车间最高频的读操作
# tasks —— 查看任务详情 /tasks/{id}
# records —— 任务记录
# notifications / orders —— 见下方 _is_bare_list 的说明
_TRACKED_READ_PREFIXES = (
"/api/v1/notifications",
"/api/v1/tasks",
"/api/v1/orders",
"/api/v1/products",
"/api/v1/records",
)
# 永久忽略的路径前缀
#
# 两类内容：
# 1. 探针与文档（/health、/docs…）—— 噪声没有审计价值
# 2. 图片类端点（/api/v1/products/qrcode）—— 走 <img src> 加载，
# 一次列表页渲染就会并发拉几十张图，逐条留痕会把审计日志塞满，
# 真正有价值的操作反而被淹没。它也不含业务数据（只渲染二维码图片）。
_IGNORED_PREFIXES = (
"/health", "/docs", "/redoc", "/openapi.json",
"/api/v1/products/qrcode",
)
# 路径段 → 审计模块
_PATH_MODULE: dict[str, str] = {
"products": "product",
"tasks": "task",
"orders": "order",
"records": "record",
"print": "print",
"materials": "material",
"users": "user",
"upload": "upload",
"notifications": "notification",
"app-version": "app",
"analytics": "analytics",
"dashboard": "dashboard",
"holidays": "holiday",
"screen": "screen",
"webhooks": "external",
"external": "external",
"audit": "audit",
"auth": "auth",
}
# 路径段 → 动作（优先于按 HTTP 方法推断）
_SEGMENT_ACTION: dict[str, str] = {
"login": "login",
"logout": "logout",
"refresh": "refresh",
"export": "export",
"download": "export",
"print": "print",
"upload": "upload",
"finalize": "finalize",
"receive": "receive",
"transfer": "transfer",
"reject": "reject",
"recall": "recall",
"spawn": "spawn",
"complete": "complete",
"end": "end",
# 消息已读：PUT /notifications/{id}/read。
# 没有这一条时会回退到 _METHOD_ACTION（PUT → update → "修改"），
# 把"点开一条通知"记成"修改了某样东西"，语义完全走样。
"read": "mark_read",
}
_METHOD_ACTION: dict[str, str] = {
"POST": "create",
"PUT": "update",
"PATCH": "update",
"DELETE": "delete",
"GET": "read",
}
# 不可能是业务 ID 的路径段，避免把动作词误当成 target_id
_NON_ID_SEGMENTS = frozenset(
set(_SEGMENT_ACTION) | {"api", "v1", "me", "options", "export", "lookup", "batch"}
)
def _is_bare_list(path: str) -> bool:
"""判断是否只是「拉整个列表」（如 GET /api/v1/tasks/）。
这类请求【不采集】，理由：
· 列表接口被前端高频轮询（消息、任务列表尤其明显），逐条留痕会让
audit_logs 迅速膨胀，真正有价值的操作反而被淹没；
· 「查看详情」(/tasks/{id}) 才代表用户真的点开了某条业务数据。
判定用「去掉末尾斜杠后是否恰好等于某个受跟踪前缀」，
比正则更直观，也天然把查询串排除在外（request.url.path 不含 ?query）。
"""
return path.rstrip("/") in _TRACKED_READ_PREFIXES
def _derive_module_and_action(path: str, method: str) -> tuple[str, str, str | None]:
"""由请求路径与 HTTP 方法推导 (module, action, target_id)"""
parts = [p for p in path.split("/") if p]
module = "other"
module_idx = -1
for i, seg in enumerate(parts):
if seg in _PATH_MODULE:
module = _PATH_MODULE[seg]
module_idx = i
break
action = None
for seg in reversed(parts):
if seg in _SEGMENT_ACTION:
action = _SEGMENT_ACTION[seg]
break
if action is None:
action = _METHOD_ACTION.get(method, method.lower())
target_id = None
if module_idx >= 0 and module_idx + 1 < len(parts):
candidate = parts[module_idx + 1]
if candidate not in _NON_ID_SEGMENTS:
target_id = candidate
return module, action, target_id
class AuditMiddleware(BaseHTTPMiddleware):
"""写操作审计采集。
必须注册在 RequestContextMiddleware **内层**，因为它依赖后者写入
request.state 的 request_id 才能与结构化日志对账。
"""
def _should_audit(self, request: Request) -> bool:
path = request.url.path
if path.startswith(_IGNORED_PREFIXES):
return False
if request.method in _MUTATING_METHODS:
return True
if request.method == "GET":
lowered = path.lower()
if any(kw in lowered for kw in _SENSITIVE_READ_KEYWORDS):
return True
# 核心业务数据的「查看详情」也留痕（证明用户在真的使用系统）
if path.startswith(_TRACKED_READ_PREFIXES):
return not _is_bare_list(path)
return False
return False
async def dispatch(
self, request: Request, call_next: RequestResponseEndpoint
) -> Response:
if not self._should_audit(request):
return await call_next(request)
status_code = 500
error_message: str | None = None
try:
response = await call_next(request)
status_code = response.status_code
return response
except Exception as exc:
# 异常最终由 ServerErrorMiddleware 转成 500；这里先标记，
# 保证「失败的操作也有审计」——这正是选用独立 session 的目的
error_message = f"{type(exc).__name__}: {exc}"[:1000]
raise
finally:
await self._write(request, status_code, error_message)
async def _write(
self, request: Request, status_code: int, error_message: str | None
) -> None:
try:
module, action, target_id = _derive_module_and_action(
request.url.path, request.method
)
client = request.client
await record_audit(
action=action,
module=module,
user_id=getattr(request.state, "audit_user", None),
display_name=getattr(request.state, "audit_display_name", None),
role=getattr(request.state, "audit_role", None),
target_type=module,
target_id=target_id,
# 对产品而言路径里的 ID 就是身份证号，本身即人可读的标识
target_name=target_id if module == "product" else None,
ip_address=client.host if client else None,
user_agent=request.headers.get("user-agent"),
method=request.method,
url=request.url.path,
status_code=status_code,
error_message=error_message,
request_id=getattr(request.state, "request_id", None),
)
except Exception:
# record_audit 内部已兜底；这里再兜一层，确保审计绝不冒泡成 500
logger.exception("审计采集失败（已忽略）")

11
backend/app/core/config.py
View File

@ -16,17 +16,6 @@ class Settings(BaseSettings):
# ---- 调试 ----
DEBUG: bool = True
# ---- 应用元信息 ----
APP_VERSION: str = "1.0.0"
# ---- 日志 ----
LOG_LEVEL: str = "INFO"
LOG_JSON: bool = True # 生产保持 True（便于采集）；本地调试可设 False 换可读格式
# ---- 错误追踪（可选，不装 sentry-sdk 则自动跳过）----
SENTRY_DSN: str | None = None
SENTRY_TRACES_SAMPLE_RATE: float = 0.0
# ---- CORS 跨域白名单（JSON 数组字符串，直接从 .env 的 CORS_ORIGINS 读取） ----
CORS_ORIGINS: str = '["http://localhost:1420", "tauri://localhost"]'

35
backend/app/core/deps.py
View File

@ -1,35 +0,0 @@
"""通用 FastAPI 依赖"""
from __future__ import annotations
from fastapi import Depends, HTTPException, status
from app.core.roles import ADMIN_ROLES
from app.services.auth_service import get_current_user
def require_roles(*roles: str):
"""生成「限定角色」依赖，避免同一个内联判断被复制到每个端点。
用法::
@router.get("/x")
async def x(current_user: dict = Depends(require_admin)):
...
失败一律 403 且不透露允许的角色集合（避免给探测者提供线索）。
"""
allowed = frozenset(roles)
async def _guard(current_user: dict = Depends(get_current_user)) -> dict:
if (current_user or {}).get("role") not in allowed:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="当前角色无权访问该接口",
)
return current_user
return _guard
# 审计日志等高权限接口复用同一实例
require_admin = require_roles(*ADMIN_ROLES)

93
backend/app/core/health.py
View File

@ -1,93 +0,0 @@
"""健康检查 — 存活探针与就绪探针分离
为什么必须拆开：
- 存活探针（liveness）只回答「进程还活着吗」，绝不能探测外部依赖。
否则数据库抖一下，编排系统会判定进程已死并反复重启容器，
把一次依赖故障放大成全站雪崩。
- 就绪探针（readiness）回答「现在能对外服务吗」。依赖不可用时返回 503，
由负载均衡把该实例摘掉，依赖恢复后自动回来。
"""
from __future__ import annotations
import logging
import anyio
from fastapi import APIRouter
from fastapi.responses import JSONResponse
from sqlalchemy import text
from app.core.config import settings
from app.core.database import AsyncSessionLocal
from app.core.mom_database import mom_engine
logger = logging.getLogger("track.health")
router = APIRouter(tags=["健康检查"])
async def _probe_primary_db() -> bool:
"""主库探活 — 业务强依赖，失败即不就绪"""
try:
async with AsyncSessionLocal() as session:
await session.execute(text("SELECT 1"))
return True
except Exception:
logger.exception("主库探活失败")
return False
def _probe_mom_db_sync() -> bool:
try:
with mom_engine.connect() as conn:
conn.execute(text("SELECT 1"))
return True
except Exception:
logger.exception("MOM 库探活失败")
return False
async def _probe_mom_db() -> bool:
# MOM 用的是同步引擎，放线程池执行，避免阻塞事件循环
return await anyio.to_thread.run_sync(_probe_mom_db_sync)
async def _collect() -> tuple[bool, dict[str, str]]:
primary_ok = await _probe_primary_db()
mom_ok = await _probe_mom_db()
checks = {
"database": "ok" if primary_ok else "fail",
# MOM 是外部只读依赖：挂掉时登录/选料降级，但扫码、流转、看板仍可用。
# 因此只标记 degraded、不摘流量 —— 否则 MOM 一抖就让在产车间全线停摆。
"mom_database": "ok" if mom_ok else "degraded",
}
return primary_ok, checks
@router.get("/health/live", include_in_schema=False)
async def liveness() -> dict:
"""存活探针：不触碰任何依赖，恒定快速返回"""
return {"status": "ok"}
@router.get("/health/ready", include_in_schema=False)
async def readiness() -> JSONResponse:
"""就绪探针：主库不可用时返回 503，让负载均衡摘流量"""
ready, checks = await _collect()
return JSONResponse(
{"status": "ready" if ready else "not_ready", "checks": checks},
status_code=200 if ready else 503,
)
@router.get("/health", include_in_schema=False)
async def health() -> JSONResponse:
"""兼容旧监控脚本：语义等同就绪探针，并附带版本号"""
ready, checks = await _collect()
return JSONResponse(
{
"status": "ok" if ready else "unavailable",
"version": settings.APP_VERSION,
"checks": checks,
},
status_code=200 if ready else 503,
)

89
backend/app/core/logging.py
View File

@ -1,89 +0,0 @@
"""结构化日志 — 单行 JSON 输出 + 请求上下文注入
设计要点：
1. 零第三方依赖，只用 stdlib（logging + json + contextvars）。
2. 业务代码通过 `extra={"extra_fields": {...}}` 附加结构化字段，
不要把可检索的字段拼进 msg 字符串 —— 拼进去就只能靠正则捞了。
3. request_id / user 走 contextvar。contextvar 在 asyncio 下按任务隔离，
并发请求之间不会串号；由 RequestContextMiddleware 与 get_current_user 写入。
"""
from __future__ import annotations
import json
import logging
import sys
from contextvars import ContextVar
from datetime import datetime, timezone
# 请求级上下文
request_id_var: ContextVar[str | None] = ContextVar("request_id", default=None)
user_var: ContextVar[str | None] = ContextVar("user", default=None)
class _ContextFilter(logging.Filter):
"""把 contextvar 注入每条 record，使 JSON 自带 request_id / user"""
def filter(self, record: logging.LogRecord) -> bool:
record.request_id = request_id_var.get()
record.user = user_var.get()
return True
class JsonFormatter(logging.Formatter):
"""单行 JSON — 便于 Loki / ELK / CloudWatch 直接解析，无需正则"""
def format(self, record: logging.LogRecord) -> str:
payload: dict = {
"ts": datetime.fromtimestamp(record.created, timezone.utc).isoformat(),
"level": record.levelname,
"logger": record.name,
"msg": record.getMessage(),
}
if getattr(record, "request_id", None):
payload["request_id"] = record.request_id
if getattr(record, "user", None):
payload["user"] = record.user
payload.update(getattr(record, "extra_fields", None) or {})
if record.exc_info:
payload["exc"] = self.formatException(record.exc_info)
return json.dumps(payload, ensure_ascii=False, default=str)
class TextFormatter(logging.Formatter):
"""本地开发可读格式（LOG_JSON=false 时启用）"""
def format(self, record: logging.LogRecord) -> str:
line = (
f"{self.formatTime(record, '%H:%M:%S')} "
f"{record.levelname:<5} {record.name} - {record.getMessage()}"
)
extras = getattr(record, "extra_fields", None)
if extras:
line += " | " + " ".join(f"{k}={v}" for k, v in extras.items())
if record.exc_info:
line += "\n" + self.formatException(record.exc_info)
return line
def setup_logging(level: str = "INFO", json_output: bool = True) -> None:
"""配置根 logger。必须在应用启动前调用一次。"""
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JsonFormatter() if json_output else TextFormatter())
handler.addFilter(_ContextFilter())
root = logging.getLogger()
# 清空既有 handler：uvicorn --reload / 多 worker 下模块可能被重复导入，
# 不清会看到每条日志打印 N 遍
root.handlers.clear()
root.addHandler(handler)
root.setLevel(level.upper())
# uvicorn 自带 handler 会绕过上面的 formatter，必须清掉并让它向根传播
for name in ("uvicorn", "uvicorn.error", "uvicorn.access"):
lg = logging.getLogger(name)
lg.handlers.clear()
lg.propagate = True
# 访问日志统一由 RequestContextMiddleware 输出（含耗时 / 用户 / request_id），
# 故关闭 uvicorn 自带的访问日志，避免重复
logging.getLogger("uvicorn.access").disabled = True

103
backend/app/core/middleware.py
View File

@ -1,103 +0,0 @@
"""请求上下文中间件 — request_id 生成/透传 + 结构化访问日志"""
from __future__ import annotations
import logging
import time
import uuid
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from app.core.logging import request_id_var, user_var
from app.services.audit_service import touch_daily_seen
access_log = logging.getLogger("track.access")
# 探针被高频轮询，降级为 DEBUG 避免把有价值的信息淹掉
_QUIET_PATHS = frozenset({"/health", "/health/live", "/health/ready"})
class RequestContextMiddleware(BaseHTTPMiddleware):
"""为每个请求建立可追踪上下文。
- request_id：优先沿用上游网关传来的 X-Request-ID，实现全链路追踪；
没有就生成一个。响应头回写该 ID，前端报错时可直接带上，
运维拿 ID 就能在日志里精确定位到这一次请求。
- 访问日志：method / path / status / duration_ms / client / user。
"""
async def dispatch(self, request: Request, call_next):
request_id = request.headers.get("X-Request-ID") or uuid.uuid4().hex
# 同时写入 request.state：它由 ASGI scope 承载，作用域比 contextvar 更长。
# FastAPI 把 Exception 处理器交给 ServerErrorMiddleware（位于本中间件外层），
# 异常传播到那里时 contextvar 已在 finally 中被重置，只有 state 还留着 ID。
request.state.request_id = request_id
rid_token = request_id_var.set(request_id)
user_token = user_var.set(None)
started = time.perf_counter()
logged = False
status_code = 500
try:
response = await call_next(request)
status_code = response.status_code
response.headers["X-Request-ID"] = request_id
self._log_access(request, status_code, started)
logged = True
await self._touch_activity(request)
return response
finally:
# 异常路径也要留下访问记录，否则接口 500 时日志里反而没有痕迹
if not logged:
self._log_access(request, status_code, started)
await self._touch_activity(request)
request_id_var.reset(rid_token)
user_var.reset(user_token)
async def _touch_activity(self, request: Request) -> None:
"""记录「该用户今天活动过」，供日活报表算上线/下线时间。
为什么挂在这一层：本中间件是最外层，能覆盖**所有**请求 ——
包括不被审计的普通 GET。而审计中间件只记写操作，当天只翻看、
没做写操作的人会被日活完全漏掉。
user 同样只能从 request.state 取：本中间件在独立 task 中执行，
路由内写的 contextvar 不会回流（详见 _log_access 的说明）。
未认证请求取不到 user，自然跳过。
"""
await touch_daily_seen(getattr(request.state, "audit_user", None))
def _log_access(self, request: Request, status_code: int, started: float) -> None:
path = request.url.path
duration_ms = round((time.perf_counter() - started) * 1000, 1)
# user 必须从 request.state 取：本中间件在独立 task 中执行，路由内
# 写入的 contextvar 不会回流到这里（详见 get_current_user 的说明）。
user = getattr(request.state, "audit_user", None) or user_var.get()
if status_code >= 500:
level = logging.ERROR
elif status_code >= 400:
level = logging.WARNING
elif path in _QUIET_PATHS:
level = logging.DEBUG
else:
level = logging.INFO
access_log.log(
level,
"%s %s -> %s (%.1fms)",
request.method,
path,
status_code,
duration_ms,
extra={
"extra_fields": {
"method": request.method,
"path": path,
"status": status_code,
"duration_ms": duration_ms,
"client": request.client.host if request.client else None,
"user": user,
}
},
)

31
backend/app/core/roles.py
View File

@ -1,31 +0,0 @@
"""角色定义与管理员判定 —— 单一事实来源
背景：角色字符串此前散落在至少三处 —— task_service.ADMIN_ROLES、
products.py 的内联判断、以及前端 constants/task.ts。同一份规则抄多份的后果
已经发生过：前端 constants/task.ts:233 的注释记录了一次「移动端只判了
SUPER_ADMIN、漏了 SUPERVISOR，导致主管被误挡」的事故。
本模块把**角色常量与管理员判定**先收敛到一处，供后端统一引用。
完整的「角色 × 权限点」可配置矩阵是后续工作；但任何推进都应从这里出发，
不要再新增第四份副本。
"""
from __future__ import annotations
SUPER_ADMIN = "SUPER_ADMIN"
SUPERVISOR = "SUPERVISOR"
# 注意：MOM 登录返回的默认角色是小写 operator（见 auth_service.login）
OPERATOR = "OPERATOR"
# 管理员角色：可执行收口、审计查看等高权限动作
ADMIN_ROLES: frozenset[str] = frozenset({SUPER_ADMIN, SUPERVISOR})
ROLE_LABELS: dict[str, str] = {
SUPER_ADMIN: "超级管理员",
SUPERVISOR: "主管",
OPERATOR: "操作员",
}
def is_admin(role: str | None) -> bool:
"""role 为 None / 未知值一律视为无权限（fail-closed，不做兜底放行）"""
return role in ADMIN_ROLES

22
backend/app/core/security.py
View File

@ -39,28 +39,6 @@ def decode_token(token: str) -> dict:
return jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM])
def peek_token_identity(token: str) -> dict | None:
"""读出令牌里的用户身份 —— **仅供审计标注，绝不可用于授权**。
与 decode_token 的唯一区别：**关闭过期校验**。
为什么需要它：刷新令牌接口正是"access token 过期了才来"的场景，
请求里不带 Authorization 头，JWT 依赖根本不执行，审计只能记成
「未认证」—— 而"谁在什么时候尝试刷新"恰恰是该留痕的信息。
签名校验照常进行，伪造的令牌解不出任何东西。
⚠️ 返回值只允许写进 request.state 的审计字段；
任何鉴权判断一律走 get_current_user，不要用本函数。
"""
try:
return jwt.decode(
token, settings.SECRET_KEY, algorithms=[ALGORITHM],
options={"verify_exp": False},
)
except JWTError:
return None
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""验证明文密码 vs 哈希密码"""
return pwd_context.verify(plain_password, hashed_password)

93
backend/app/main.py
View File

@ -1,64 +1,22 @@
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from app.core.config import settings
from app.core.audit_middleware import AuditMiddleware
from app.core.health import router as health_router
from app.core.logging import request_id_var, setup_logging
from app.core.middleware import RequestContextMiddleware
from app.api.v1.router import api_router
# 日志必须在任何模块开始产日志之前配置好，故放模块顶层而非 lifespan 内
setup_logging(level=settings.LOG_LEVEL, json_output=settings.LOG_JSON)
logger = logging.getLogger("track.main")
def _init_error_tracking() -> None:
"""可选错误追踪：未配置 DSN，或未安装 sentry-sdk 时静默跳过"""
if not settings.SENTRY_DSN:
return
try:
import sentry_sdk
except ImportError:
logger.warning(
"已配置 SENTRY_DSN 但未安装 sentry-sdk，错误追踪未启用；"
"需要时执行 pip install sentry-sdk"
)
return
sentry_sdk.init(
dsn=settings.SENTRY_DSN,
traces_sample_rate=settings.SENTRY_TRACES_SAMPLE_RATE,
environment="production" if not settings.DEBUG else "development",
release=settings.APP_VERSION,
)
logger.info("错误追踪已启用 (Sentry)")
@asynccontextmanager
async def lifespan(app: FastAPI):
"""应用生命周期：启动时初始化连接，关闭时释放资源"""
_init_error_tracking()
logger.info(
"服务启动",
extra={
"extra_fields": {
"version": settings.APP_VERSION,
"debug": settings.DEBUG,
"cors_origins": settings.CORS_ORIGINS_LIST,
}
},
)
# 启动：验证数据库连接等
yield
logger.info("服务关闭")
# 关闭：清理资源
app = FastAPI(
title="Track Production API",
description="工厂生产流转管理系统 API",
version=settings.APP_VERSION,
version="0.1.0",
lifespan=lifespan,
)
@ -69,45 +27,12 @@ app.add_middleware(
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
# 暴露给浏览器 JS 读取：前端报错时才能把 request_id 一起带上便于对账
expose_headers=["X-Request-ID"],
)
# Starlette 的 add_middleware 是「后添加者在外层」。执行顺序（由外到内）：
# RequestContextMiddleware -> AuditMiddleware -> CORS -> 路由
# AuditMiddleware 必须在 RequestContext 内层，才能读到后者写入 request.state
# 的 request_id，从而把审计记录与结构化日志对上。
app.add_middleware(AuditMiddleware)
app.add_middleware(RequestContextMiddleware)
@app.exception_handler(Exception)
async def unhandled_exception_handler(request: Request, exc: Exception) -> JSONResponse:
"""兜底异常处理。
完整堆栈只进日志；响应体仅返回 request_id —— 既不把内部实现泄露给客户端，
又让用户报障时能凭这个 ID 在日志里精确定位到本次失败。
"""
# 优先取 request.state（见 RequestContextMiddleware 的说明）：
# 本处理器由 ServerErrorMiddleware 调用，此时 contextvar 已被重置
request_id = getattr(request.state, "request_id", None) or request_id_var.get()
logger.exception(
"未处理异常: %s %s",
request.method,
request.url.path,
extra={"extra_fields": {"method": request.method, "path": request.url.path}},
)
return JSONResponse(
status_code=500,
content={"detail": "服务器内部错误", "request_id": request_id},
# 该响应由 ServerErrorMiddleware（位于 RequestContextMiddleware 外层）
# 生成，中间件没机会再往响应头写 X-Request-ID，故在此显式补上，
# 保证报障时前端从响应头就能拿到可对账的 ID。
headers={"X-Request-ID": request_id} if request_id else None,
)
# ---- 注册路由 ----
app.include_router(api_router, prefix="/api/v1")
# 健康检查挂在根路径（/health*），运维探针不经过 /api/v1
app.include_router(health_router)
@app.get("/health")
async def health_check():
return {"status": "ok", "version": "0.1.0"}

4
backend/app/models/__init__.py
View File

@ -8,8 +8,6 @@ from app.models.notification import Notification
from app.models.app_version import AppVersion
from app.models.message import ProductMessage
from app.models.holiday import Holiday
from app.models.audit_log import AuditLog
from app.models.user_daily_seen import UserDailySeen
__all__ = [
"Base",
"ProductionOrder",
@ -21,6 +19,4 @@ __all__ = [
"AppVersion",
"ProductMessage",
"Holiday",
"AuditLog",
"UserDailySeen",
]

83
backend/app/models/audit_log.py
View File

@ -1,83 +0,0 @@
"""操作审计日志模型
设计参考 MOM(KCGL) 的 audit_logs，但按 Track 的技术栈与诉求做了取舍：
- 主键用 UUID（与库内其它表一致），而非 MOM 的自增 int。
- 增加 request_id：与 core/logging.py 的结构化日志打通 —— 凭一个 ID 就能把
「接口访问日志」和「审计记录」对上，排障时不用再猜。MOM 无此字段。
- 保留 module / action / target_* 的业务语义，使审计能按业务维度检索，
而不是只能按时间翻。
- 绝不记录请求体：登录等接口 body 含明文密码，一旦落库就成了长期泄露面。
与既有 task_logs 的分工：task_logs 是「任务流转轨迹」（有 task_id 非空约束，
只能挂在任务上，供流转树渲染）；本表是「操作审计」，覆盖登录、导出、
产品增删改、权限变更等与单个任务无关的动作，且额外记录来源 IP / UA / 耗时结果。
"""
import uuid
from datetime import datetime
from sqlalchemy import DateTime, Integer, String, Text
from sqlalchemy.dialects.postgresql import JSONB, UUID
from sqlalchemy.orm import Mapped, mapped_column
from app.models.base import Base
from app.core.time_utils import get_beijing_time
class AuditLog(Base):
__tablename__ = "audit_logs"
id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), primary_key=True, default=uuid.uuid4,
)
# ---- 操作人（逻辑外键 → MOM sys_user，仅存账号，无物理约束）----
user_id: Mapped[str | None] = mapped_column(
String(64), nullable=True, index=True, comment="操作人账号(逻辑外键→MOM)",
)
display_name: Mapped[str | None] = mapped_column(
String(100), nullable=True, comment="操作人显示名",
)
role: Mapped[str | None] = mapped_column(
String(50), nullable=True, comment="操作时角色快照",
)
# ---- 业务语义 ----
action: Mapped[str] = mapped_column(
String(50), nullable=False, index=True, comment="动作: create/update/delete/export/login/...",
)
module: Mapped[str] = mapped_column(
String(50), nullable=False, index=True, comment="业务模块: product/task/order/auth/print/...",
)
target_type: Mapped[str | None] = mapped_column(
String(50), nullable=True, comment="目标类型(表名或实体名)",
)
target_id: Mapped[str | None] = mapped_column(
String(100), nullable=True, index=True, comment="目标ID",
)
target_name: Mapped[str | None] = mapped_column(
String(200), nullable=True, comment="目标显示名(如产品身份证/工单号)",
)
details: Mapped[dict | None] = mapped_column(
JSONB, nullable=True, comment="变更详情 {old:{}, new:{}}；禁止写入密码等敏感字段",
)
# ---- 请求上下文（由中间件自动填充）----
ip_address: Mapped[str | None] = mapped_column(String(50), nullable=True, comment="来源IP")
user_agent: Mapped[str | None] = mapped_column(String(500), nullable=True, comment="浏览器UA")
method: Mapped[str | None] = mapped_column(String(10), nullable=True, comment="HTTP方法")
url: Mapped[str | None] = mapped_column(String(500), nullable=True, comment="请求路径")
status_code: Mapped[int | None] = mapped_column(Integer, nullable=True, comment="响应状态码")
error_message: Mapped[str | None] = mapped_column(Text, nullable=True, comment="错误信息(如有)")
# ---- 与结构化日志对账用 ----
request_id: Mapped[str | None] = mapped_column(
String(64), nullable=True, index=True, comment="关联 core/logging 的 request_id",
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=get_beijing_time, index=True, comment="操作时间",
)
def __repr__(self) -> str:
return f"<AuditLog {self.action} {self.module} by {self.user_id}>"

47
backend/app/models/user_daily_seen.py
View File

@ -1,47 +0,0 @@
"""每日用户活动表 —— 一天一人一行，只记录"今天来过"这件事。
为什么需要它（而不是复用 audit_logs）
--------------------------------------
日活报表要的「上线时间 / 下线时间」，两个都不能从审计表直接得出：
1. **上线/下线时间不能取登录时间**：Refresh Token 有效期 7 天，用户不必每天
重新登录。按登录算会出现「登录次数 0、上线时间空，但操作次数 35」的
自相矛盾报表。
2. **也不能只取写操作时间**：审计中间件只记录写操作（及导出/打印这类敏感读），
普通 GET 不入账。当天只翻看、没做写操作的人会被整条漏掉。
3. **更不能把活动记录写进 audit_logs**：
· 「上线时间」是**事件**（INSERT 一次即可），但「下线时间」是**状态**
（每次活动都要刷新同一个值）。往审计流水里做 UPDATE，等于承认审计记录
可以被改写 —— 那审计本身就失去可信度了。
· 若改为每个请求 INSERT 一条，表会随访问量线性膨胀。
于是单开一张"可变的小状态表"：一人一天一行，首见 INSERT、其后只
UPDATE last_seen_at。50 人 × 365 天 ≈ 1.8 万行/年，可忽略。
"""
from __future__ import annotations
from datetime import date, datetime
from sqlalchemy import Date, DateTime, String
from sqlalchemy.orm import Mapped, mapped_column
from app.models.base import Base
class UserDailySeen(Base):
"""用户在某个北京时间自然日的首末活动时刻"""
__tablename__ = "user_daily_seen"
# 联合主键即 UPSERT 的冲突目标，也是"一天一人一行"的保证
user_id: Mapped[str] = mapped_column(String(64), primary_key=True)
day: Mapped[date] = mapped_column(Date, primary_key=True, comment="北京时间自然日")
first_seen_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, comment="当天首次活动时刻",
)
last_seen_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, comment="当天末次活动时刻",
)

88
backend/app/schemas/audit.py
View File

@ -1,88 +0,0 @@
"""审计日志 Pydantic Schema"""
from __future__ import annotations
import uuid
from datetime import datetime
from pydantic import BaseModel
class AuditLogResponse(BaseModel):
"""单条审计记录"""
id: uuid.UUID
user_id: str | None = None
display_name: str | None = None
role: str | None = None
action: str
action_label: str | None = None # 服务端补的中文标签，避免前端各处硬编码
module: str
module_label: str | None = None
target_type: str | None = None
target_id: str | None = None
target_name: str | None = None
details: dict | None = None
ip_address: str | None = None
user_agent: str | None = None
method: str | None = None
url: str | None = None
status_code: int | None = None
error_message: str | None = None
# 与结构化日志对账用：拿着它就能捞到对应的接口日志
request_id: str | None = None
created_at: datetime
model_config = {"from_attributes": True}
class AuditLogListResponse(BaseModel):
"""审计日志分页列表"""
items: list[AuditLogResponse]
total: int
class DailyUsageRow(BaseModel):
"""某个操作人在某一天的用量汇总（北京时间自然日）"""
day: str # YYYY-MM-DD（北京时间）
user_id: str | None = None
display_name: str | None = None
role: str | None = None
login_count: int = 0 # 登录次数（当天成功登录）
logout_count: int = 0 # 登出次数（当天成功登出）
op_count: int = 0 # 操作次数（当天全部审计记录数）
# ⚠️ 上线/下线时间取【当天首次/末次活动】，不是登录/登出时间：
# token 有效期内（refresh 7 天）用户不会重新登录，按登录算会导致
# 「登录次数 0 但操作 35 次」这种自相矛盾。
first_active_at: datetime | None = None # 上线时间（当天首次活动）
last_active_at: datetime | None = None # 下线时间（当天末次活动）
class DailyUsageResponse(BaseModel):
"""日活 / 使用统计"""
start_date: str
end_date: str
items: list[DailyUsageRow]
total: int # 行数（= 天数 × 人数），不是审计记录数
class AuditOption(BaseModel):
"""筛选项（value/label 结构，直接喂给前端下拉）"""
value: str
label: str
class AuditOptionsResponse(BaseModel):
"""筛选项集合"""
modules: list[AuditOption]
actions: list[AuditOption]
# 导出可选的列（value=后端列 key，label=中文表头）。
# 由后端下发而非前端硬编码：列的中文名与取值口径都在后端，
# 两端各写一份迟早会出现"导出的列和页面上的对不上"。
log_export_columns: list[AuditOption] = []
usage_export_columns: list[AuditOption] = []

458
backend/app/services/audit_service.py
View File

@ -1,458 +0,0 @@
"""审计服务 — 写入与检索
写入方案的取舍（与 MOM/KCGL 不同，理由如下）
--------------------------------------------------
MOM 用 SQLAlchemy event listener + **同事务**写入：优点是全自动、业务代码零改动；
缺点是业务事务回滚时审计记录一起被回滚掉 —— 而失败/被拒的操作恰恰是最需要
留痕的（比如越权尝试、参数错误导致的 4xx）。
Track 改为：响应生成后，用**独立 session** 写入审计。
- 业务回滚不影响审计，失败操作照样留痕
- 审计写入失败也不影响业务（全包裹 try/except，仅记日志）
- 代价：审计与业务不是原子提交，极端情况（响应后进程立即被 kill）可能丢一条。
对内部系统的操作审计，这个取舍划算。
"""
from __future__ import annotations
import logging
import time
import uuid
from datetime import datetime
from sqlalchemy import and_, func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import AsyncSessionLocal
from app.core.time_utils import get_beijing_time
from app.models.audit_log import AuditLog
from app.models.user_daily_seen import UserDailySeen
logger = logging.getLogger("track.audit")
# 绝不落库的敏感字段名（命中即替换为 ***）
# 登录请求体含明文密码，一旦进审计表就成了长期泄露面
_SENSITIVE_KEYS = frozenset(
{"password", "passwd", "pwd", "token", "access_token", "refresh_token",
"secret", "api_key", "authorization", "password_hash"}
)
# 模块 / 动作 的中文标签（前端下拉与列表展示用）
MODULE_LABELS: dict[str, str] = {
"auth": "认证登录",
"product": "产品管理",
"task": "任务流转",
"order": "订单管理",
"record": "任务记录",
"print": "标签打印",
"material": "物料",
"user": "用户",
"notification": "消息通知",
"upload": "文件上传",
"dashboard": "看板统计",
"analytics": "效能分析",
"screen": "数据大屏",
"holiday": "节假日配置",
"app": "App版本",
"external": "外部系统对接",
"audit": "审计日志",
"other": "其它",
}
ACTION_LABELS: dict[str, str] = {
"create": "新增",
"update": "修改",
"delete": "删除",
# 只用于被采集的 GET（核心业务详情 / 敏感读）。
# 叫「查看详情」而不是「查询」：前者说明用户确实点开了某条业务数据，
# 后者容易被误解成"随便搜了一下"。
"read": "查看详情",
"export": "导出",
"login": "登录",
"logout": "登出",
# 刷新令牌 = 用户重新开始使用系统（token 2 小时一换，7 天免登录），
# 业务上视作一次「上线」，比"刷新令牌"这种技术词更贴近车间口径
"refresh": "上线",
"print": "打印",
"upload": "上传",
"finalize": "收口",
"receive": "接收",
"transfer": "转交",
"reject": "驳回",
"recall": "撤回",
"spawn": "派发",
"end": "结束分支",
"complete": "完结",
"mark_read": "标为已读",
}
def sanitize_details(details: dict | None) -> dict | None:
"""递归剔除敏感字段，避免密码/令牌落库"""
if not details:
return details
def _clean(value):
if isinstance(value, dict):
return {
k: ("***" if str(k).lower() in _SENSITIVE_KEYS else _clean(v))
for k, v in value.items()
}
if isinstance(value, list):
return [_clean(v) for v in value]
return value
return _clean(details)
async def record_audit(
*,
action: str,
module: str,
user_id: str | None = None,
display_name: str | None = None,
role: str | None = None,
target_type: str | None = None,
target_id: str | None = None,
target_name: str | None = None,
details: dict | None = None,
ip_address: str | None = None,
user_agent: str | None = None,
method: str | None = None,
url: str | None = None,
status_code: int | None = None,
error_message: str | None = None,
request_id: str | None = None,
) -> None:
"""写入一条审计记录。**绝不抛异常**：审计失败不能影响业务。"""
try:
async with AsyncSessionLocal() as session:
session.add(
AuditLog(
id=uuid.uuid4(),
user_id=user_id,
display_name=display_name,
role=role,
action=action,
module=module,
target_type=target_type,
target_id=str(target_id) if target_id is not None else None,
target_name=target_name,
details=sanitize_details(details),
ip_address=ip_address,
user_agent=user_agent[:500] if user_agent else None,
method=method,
url=url[:500] if url else None,
status_code=status_code,
error_message=error_message,
request_id=request_id,
)
)
await session.commit()
except Exception:
# 用 exception 级别但吞掉异常：保证调用方业务流程不受影响
logger.exception(
"审计写入失败（已忽略，不影响业务）",
extra={"extra_fields": {"action": action, "module": module, "url": url}},
)
async def list_audit_logs(
db: AsyncSession,
*,
user_id: str | None = None,
module: str | None = None,
action: str | None = None,
target_id: str | None = None,
request_id: str | None = None,
status_code: int | None = None,
start: datetime | None = None,
end: datetime | None = None,
skip: int = 0,
limit: int = 50,
) -> tuple[list[AuditLog], int]:
"""审计日志检索（按时间倒序）。返回 (当前页, 真实总数)。
真实总数走独立 COUNT —— 前端分页器依赖它，不能用 len(当前页)。
"""
filters = _log_filters(
user_id=user_id, module=module, action=action, target_id=target_id,
request_id=request_id, status_code=status_code, start=start, end=end,
)
total = await db.scalar(
select(func.count()).select_from(AuditLog).where(*filters)
) or 0
rows = (
await db.execute(
select(AuditLog)
.where(*filters)
.order_by(AuditLog.created_at.desc())
.offset(skip)
.limit(limit)
)
).scalars().all()
return list(rows), total
# ============================================================
# 导出
# ============================================================
# 单次导出的行数上限。审计表只增不减，全量导出迟早会撑爆内存与浏览器，
# 故设硬上限；超出时向上层返回 truncated=True，由前端明确提示「已截断」——
# 静默截断会让使用者以为导全了，比报错更危险。
EXPORT_MAX_ROWS = 50000
def _log_filters(
*,
user_id: str | None = None,
module: str | None = None,
action: str | None = None,
target_id: str | None = None,
request_id: str | None = None,
status_code: int | None = None,
start: datetime | None = None,
end: datetime | None = None,
) -> list:
"""审计日志的筛选条件 —— list_audit_logs 与 export_audit_logs 共用。
抽出来的唯一目的：保证「列表看到的」和「导出出去的」永远是同一批数据。
两处各写一份迟早会漂移，而导出与列表不一致是最让人不信任的那种 bug。
"""
filters = []
if user_id:
filters.append(AuditLog.user_id.ilike(f"%{user_id}%"))
if module:
filters.append(AuditLog.module == module)
if action:
filters.append(AuditLog.action == action)
if target_id:
filters.append(AuditLog.target_id == target_id)
if request_id:
filters.append(AuditLog.request_id == request_id)
if status_code is not None:
filters.append(AuditLog.status_code == status_code)
if start:
filters.append(AuditLog.created_at >= start)
if end:
filters.append(AuditLog.created_at <= end)
return filters
async def export_audit_logs(
db: AsyncSession, *, limit: int = EXPORT_MAX_ROWS, **kwargs,
) -> tuple[list[AuditLog], bool]:
"""导出用：按筛选条件取全部记录（不分页）。返回 (rows, truncated)。
多取一行来判断是否被截断 —— 比再跑一次 COUNT 便宜。
"""
rows = (
await db.execute(
select(AuditLog)
.where(*_log_filters(**kwargs))
.order_by(AuditLog.created_at.desc())
.limit(limit + 1)
)
).scalars().all()
truncated = len(rows) > limit
return list(rows[:limit]), truncated
# ============================================================
# 每日活动打点（日活报表的「上线时间 / 下线时间」来源）
# ============================================================
# 同一用户两次落盘之间的最小间隔（秒）。
#
# 打点挂在「每个请求」上，但不希望每个请求都写一次数据库 —— 那会把
# user_daily_seen 变成热点。这里用进程内缓存做节流：同一用户 2 分钟内
# 只落盘一次。代价是「末次活动时间」最多落后真实值 2 分钟，
# 对"日活统计"这个精度要求完全够用。
#
# 多 worker 部署时每个进程各持一份缓存，实际写库频率最多放大到 worker 数倍
# （4 worker × 每人每 2 分钟 1 次），依然可忽略。
_TOUCH_INTERVAL_S = 120.0
_touch_cache: dict[str, float] = {}
# 缓存只增不减会缓慢泄漏（键是 user_id，量级 = 用户数，实际很小）。
# 超过阈值就整体清空 —— 代价只是多写几次库，换来内存有界。
_TOUCH_CACHE_MAX = 5000
async def touch_daily_seen(user_id: str | None) -> None:
"""记录「该用户此刻活动过」。首次 INSERT、其后只刷新 last_seen_at。
唯一的消费方是日活报表的上线/下线时间（见 get_daily_usage）。
刻意不写进 audit_logs：那是只增不改的审计流水，而本表是需要不断
UPDATE 的状态（详见 UserDailySeen 模型注释）。
任何异常都吞掉 —— 活动打点失败绝不能影响业务请求本身。
"""
if not user_id:
return
now_mono = time.monotonic()
last = _touch_cache.get(user_id)
if last is not None and now_mono - last < _TOUCH_INTERVAL_S:
return # 节流窗口内，跳过
if len(_touch_cache) > _TOUCH_CACHE_MAX:
_touch_cache.clear()
# 先占位再写库：同一用户的并发请求不会同时打进来
_touch_cache[user_id] = now_mono
try:
from sqlalchemy.dialects.postgresql import insert as pg_insert
now = get_beijing_time()
day = now.date() # 北京时间自然日（与报表分日口径一致）
async with AsyncSessionLocal() as db:
await db.execute(
pg_insert(UserDailySeen)
.values(user_id=user_id, day=day, first_seen_at=now, last_seen_at=now)
# 冲突时只刷新 last_seen_at，first_seen_at 保持当天首次值不变
.on_conflict_do_update(
index_elements=["user_id", "day"],
set_={"last_seen_at": now},
)
)
await db.commit()
except Exception: # noqa: BLE001 —— 打点失败不影响业务
logger.exception("记录每日活动失败（已忽略）")
# ============================================================
# 日活 / 使用统计
# ============================================================
# 成功 = 2xx/3xx。登录失败（401）也要留痕，但不应计入"上线次数"。
_OK_STATUS_UPPER = 400
async def get_daily_usage(
db: AsyncSession, *, start: datetime, end: datetime,
) -> list[dict]:
"""按【北京时间自然日 × 操作人】聚合用量 —— 日活报表的数据源。
start/end 为半开区间 [start, end)，调用方按北京时间日界传入。
全部指标由**一个 GROUP BY 查询**算出，不用窗口函数：
· 登录/登出次数 = 成功登录 / 成功登出数（最终凭证是 login_count，不是"上线次数"）
· 操作频次 = 当天该用户的全部审计记录数（代表系统使用深度）
· 登录/登出次数 = 成功登录 / 成功登出数
· 上线/下线时间 = 当天**首次 / 末次活动**（优先取 user_daily_seen）
⚠️ 上线/下线时间【不能】取登录/登出时间。
Access/Refresh Token 有效期内（refresh 7 天）用户无需重新登录，
于是"周一登录、周二到周日继续用"会导致周二~周日：
登录次数=0、登录时间=空，但操作次数却是几十 —— 报表自相矛盾。
⚠️ 也不能只取审计表的写操作时间：审计中间件只记写操作，普通 GET 不入账，
当天只翻看、没做写操作的人会被整条漏掉。
故上线/下线时间优先取 user_daily_seen（挂在每个请求上打点），
仅对本表上线前的历史数据回退到审计表的写操作时间。
为什么用 `count(*) FILTER (WHERE ...)`：分组内一次扫描同时算出多个条件计数，
比多次子查询或 UNION 简单得多，且语义一眼可读。Postgres 原生支持。
⚠️ 按【北京时间】分日：created_at 是 timestamptz（实存 UTC），
直接按 UTC 分日会让 00:00~08:00 的早班操作掉到前一天。
"""
day_col = func.date(func.timezone("Asia/Shanghai", AuditLog.created_at))
login_ok = and_(
AuditLog.action == "login", AuditLog.status_code < _OK_STATUS_UPPER,
)
logout_ok = and_(
AuditLog.action == "logout", AuditLog.status_code < _OK_STATUS_UPPER,
)
stmt = (
select(
day_col.label("day"),
AuditLog.user_id.label("user_id"),
# 同一用户的 display_name / role 是一致的，取 max 只是为了
# 在 GROUP BY 下拿到一个非空代表值（避免再套一层 DISTINCT ON）
func.max(AuditLog.display_name).label("display_name"),
func.max(AuditLog.role).label("role"),
func.count().filter(login_ok).label("login_count"),
func.count().filter(logout_ok).label("logout_count"),
func.count().label("op_count"),
# 上线/下线时间取「任意记录」的首末，而不是登录/登出的首末（原因见 docstring）
func.min(AuditLog.created_at).label("first_active_at"),
func.max(AuditLog.created_at).label("last_active_at"),
)
.where(
AuditLog.created_at >= start,
AuditLog.created_at < end,
# 只统计"人"：未认证请求（如登录前的探测、refresh）没有操作人，
# 混进来会让"日活人数"虚高。若要排查匿名异常流量，走日志列表页按
# 结果/来源 IP 过滤更合适。
AuditLog.user_id.isnot(None),
)
.group_by(day_col, AuditLog.user_id)
.order_by(day_col.desc(), func.count().desc())
)
rows = (await db.execute(stmt)).all()
# ── 活动表：当天首次/末次活动（覆盖"只翻看不操作"的人）──
# day 列是北京时间 DATE，与上面的 day_col 口径一致，可直接按 (user_id, day) 对齐。
# 取 start.date() ~ end.date()（end 是次日 00:00 的半开上界，故用 <）。
seen_rows = (
await db.execute(
select(
UserDailySeen.user_id, UserDailySeen.day,
UserDailySeen.first_seen_at, UserDailySeen.last_seen_at,
).where(
UserDailySeen.day >= start.date(),
UserDailySeen.day < end.date(),
)
)
).all()
seen = {
(s.user_id, s.day.strftime("%Y-%m-%d")): (s.first_seen_at, s.last_seen_at)
for s in seen_rows
}
audit = {
(r.user_id, r.day.strftime("%Y-%m-%d") if hasattr(r.day, "strftime") else str(r.day)): r
for r in rows
}
# ── 合并 ──
# 并集：只有审计记录的人（本表上线前的历史数据）和只有活动记录的人
# （当天只翻看、没做写操作）都要出现，各自缺的部分留空/计 0。
items: list[dict] = []
for key in set(audit) | set(seen):
user_id, day = key
a = audit.get(key)
first_seen, last_seen = seen.get(key, (None, None))
# 取「两者的最早/最晚」，而不是简单地"活动表优先"：
# 活动表靠请求触发且有 2 分钟节流，极端情况（跨零点被节流、
# 打点写库失败被吞掉）可能晚于当天第一次写操作。
# 取 min/max 后，结果永远不会比任一来源更差，也不需要为兜底写分支逻辑。
audit_first = a.first_active_at if a else None
audit_last = a.last_active_at if a else None
first_candidates = [t for t in (first_seen, audit_first) if t is not None]
last_candidates = [t for t in (last_seen, audit_last) if t is not None]
items.append({
"day": day,
"user_id": user_id,
# 姓名字段只有审计记录里有（活动表为了轻量刻意不冗余存）
"display_name": a.display_name if a else None,
"role": a.role if a else None,
"login_count": (a.login_count or 0) if a else 0,
"logout_count": (a.logout_count or 0) if a else 0,
"op_count": (a.op_count or 0) if a else 0,
"first_active_at": min(first_candidates) if first_candidates else None,
"last_active_at": max(last_candidates) if last_candidates else None,
})
# 与 SQL 里的排序保持一致：日期倒序 → 操作次数倒序
items.sort(key=lambda x: (x["day"], x["op_count"]), reverse=True)
return items

17
backend/app/services/auth_service.py
View File

@ -1,5 +1,5 @@
"""认证服务 — 对接 MOM 系统 sys_user 表 + Track 自有 JWT（双 Token 架构）"""
from fastapi import HTTPException, status, Depends, Request
from fastapi import HTTPException, status, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError, jwt
from werkzeug.security import check_password_hash
@ -14,7 +14,6 @@ from app.core.security import (
TOKEN_TYPE_REFRESH,
)
from app.core.mom_database import MomSessionLocal
from app.core.logging import user_var
from app.schemas.user import LoginResponse, UserResponse
security = HTTPBearer()
@ -113,7 +112,6 @@ def refresh_access_token(refresh_token: str) -> dict:
async def get_current_user(
request: Request,
credentials: HTTPAuthorizationCredentials = Depends(security),
) -> dict:
"""从 Bearer Token 解析当前用户（仅接受 Access Token）"""
@ -131,19 +129,6 @@ async def get_current_user(
detail="请使用 Access Token 访问 API，Refresh Token 仅用于刷新",
)
# 操作人身份要写两处，用途不同，缺一不可：
# 1) contextvar —— 供本请求任务内的业务/service 日志使用；
# 2) request.state —— 中间件在独立 task 中执行（Starlette 的
# BaseHTTPMiddleware 用 anyio start_soon 起新 task，而 asyncio
# 每个 Task 会复制 context），因此中间件读不到路由内改的
# contextvar，只能通过 ASGI scope 承载的 state 拿到。
# username 即 assignee_id 口径，比数字 id 直观得多。
user_label = payload.get("username") or user_id
user_var.set(user_label)
request.state.audit_user = user_label
request.state.audit_display_name = payload.get("display_name") or ""
request.state.audit_role = payload.get("role") or ""
return payload
except JWTError:
raise HTTPException(status_code=401, detail="无效的 Token")

39
backend/app/services/task_service.py
View File

@ -3,7 +3,7 @@ from __future__ import annotations
import json
import uuid
from fastapi import HTTPException, status
from sqlalchemy import select, delete, func
from sqlalchemy import select, delete
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
@ -22,7 +22,6 @@ from app.core.lifecycle import (
)
from app.models.product import Product
from app.models.task_log import TaskLog
from app.core.roles import ADMIN_ROLES
from app.schemas.task import (
TaskCreate,
TaskUpdate,
@ -43,8 +42,7 @@ from app.schemas.task import (
VIRTUAL_WAREHOUSE = "virtual_warehouse"
# 管理员/主管角色白名单 — 拥有上帝视角操作权限
# 定义已收敛到 app.core.roles（单一事实来源）；本模块继续以同名导出，
# 兼容 products.py 等处 `from app.services.task_service import ADMIN_ROLES` 的既有引用
ADMIN_ROLES = {"SUPER_ADMIN", "SUPERVISOR"}
async def _recalc_product_location(
@ -459,37 +457,22 @@ async def get_all_tasks(
assignee_id: str | None = None, skip: int = 0, limit: int = 50
) -> TaskListResponse:
"""获取任务列表，可按产品/负责人筛选"""
filters = []
if product_id:
filters.append(Task.product_id == product_id)
if assignee_id:
filters.append(Task.assignee_id == assignee_id)
# 总数必须独立 COUNT：移动端「我的任务」用 total 判断 hasMore
# （tasks.length < total），若 total 取当前页条数，首页满员时
# hasMore 恒为 false，列表永远停在第一页。
total = await db.scalar(
select(func.count()).select_from(Task).where(*filters)
) or 0
stmt = (
select(Task)
.options(
selectinload(Task.records),
selectinload(Task.product),
)
.where(*filters)
.offset(skip)
.limit(limit)
.order_by(Task.created_at.desc())
stmt = select(Task).options(
selectinload(Task.records),
selectinload(Task.product),
)
if product_id:
stmt = stmt.where(Task.product_id == product_id)
if assignee_id:
stmt = stmt.where(Task.assignee_id == assignee_id)
stmt = stmt.offset(skip).limit(limit).order_by(Task.created_at.desc())
result = await db.execute(stmt)
tasks = result.scalars().all()
# 返回扁平列表（不递归 children，避免 MissingGreenlet）
flat_tasks = [_to_flat_response(t) for t in tasks]
return TaskListResponse(tasks=flat_tasks, total=total)
return TaskListResponse(tasks=flat_tasks, total=len(flat_tasks))
# ============================================================

2
frontend/src/App.tsx
View File

@ -41,7 +41,6 @@ const AdminProductsPage = lazy(() => import("./pages/admin/AdminProductsPage"));
const AdminTasksPage = lazy(() => import("./pages/admin/AdminTasksPage"));
const AdminPeoplePage = lazy(() => import("./pages/admin/AdminPeoplePage"));
const AdminPrintConfigPage = lazy(() => import("./pages/admin/AdminPrintConfigPage"));
const AdminAuditLogPage = lazy(() => import("./pages/admin/AdminAuditLogPage"));
const AnalyticsDashboard = lazy(() => import("./pages/admin/AnalyticsDashboard"));
const MatrixBoard = lazy(() => import("./pages/MatrixBoard"));
const ScreenDashboard = lazy(() => import("./pages/admin/ScreenDashboard"));
@ -80,7 +79,6 @@ export default function App() {
<Route path="/admin/print-config" element={<AdminPrintConfigPage />} />
<Route path="/admin/analytics" element={<AnalyticsDashboard />} />
<Route path="/admin/matrix" element={<MatrixBoard />} />
<Route path="/admin/audit" element={<AdminAuditLogPage />} />
</Route>
</Routes>
</Suspense>

75
frontend/src/components/admin/ExportColumnsModal.tsx
View File

@ -1,75 +0,0 @@
/**
* 导出列选择弹窗 —— 勾选要写进 CSV 的列。
*
* 列清单由后端 /audit/options 下发（value=后端列 key，label=中文表头），
* 前端不硬编码表头：否则两端各维护一份，迟早出现「导出的列和页面对不上」。
*
* 默认全选 —— 大多数人只是想"全部导出来"，不该逼他们先勾一遍。
*/
import { useEffect, useState } from "react";
import { Modal, Checkbox, Button } from "antd";
import type { AuditOption } from "../../services/auditApi";
export default function ExportColumnsModal({
open,
columns,
submitting,
onCancel,
onConfirm,
}: {
open: boolean;
columns: AuditOption[];
submitting?: boolean;
onCancel: () => void;
/** 传出当前勾选的列 key（顺序 = 后端下发顺序，保证表头稳定） */
onConfirm: (keys: string[]) => void;
}) {
const [checked, setChecked] = useState<string[]>([]);
// 每次打开都重置为全选：上一次的勾选残留会让用户莫名少导几列
useEffect(() => {
if (open) setChecked(columns.map((c) => c.value));
}, [open, columns]);
return (
<Modal
open={open}
title="选择导出列"
onCancel={onCancel}
width={520}
footer={
<div className="flex items-center justify-between">
<div className="flex gap-2">
<Button size="small" onClick={() => setChecked(columns.map((c) => c.value))}>
全选
</Button>
<Button size="small" onClick={() => setChecked([])}>
全不选
</Button>
</div>
<div className="flex gap-2">
<Button onClick={onCancel}>取消</Button>
<Button
type="primary"
loading={submitting}
disabled={checked.length === 0}
onClick={() => onConfirm(checked)}
>
导出 ({checked.length} 列)
</Button>
</div>
</div>
}
>
{checked.length === 0 && (
<p className="mb-2 text-xs text-amber-600">至少勾选一列才能导出。</p>
)}
<Checkbox.Group
value={checked}
onChange={(v) => setChecked(v as string[])}
className="grid grid-cols-3 gap-y-2"
options={columns.map((c) => ({ value: c.value, label: c.label }))}
/>
</Modal>
);
}

14
frontend/src/components/layout/AdminLayout.tsx
View File

@ -1,5 +1,5 @@
import { NavLink, Outlet, useLocation, useNavigate, Navigate } from "react-router-dom";
import { QrCode, Package, ArrowLeft, LayoutDashboard, Smartphone, GitBranch, LogOut, User, Users, BarChart3, Table2, Tv, ScrollText } from "lucide-react";
import { QrCode, Package, ArrowLeft, LayoutDashboard, Smartphone, GitBranch, LogOut, User, Users, BarChart3, Table2, Tv } from "lucide-react";
import { useAuth } from "../../contexts/AuthContext";
const MENU = [
@ -39,12 +39,6 @@ const MENU = [
icon: Table2,
description: "规格型号 × 人员/工序 在制品透视表",
},
{
title: "操作审计",
path: "/admin/audit",
icon: ScrollText,
description: "谁在何时操作了什么 · 含失败与被拒请求",
},
{
title: "管理层大屏",
path: "/admin/screen",
@ -72,10 +66,8 @@ export default function AdminLayout() {
return <Navigate to="/admin/login" replace />;
}
async function handleLogout() {
// 必须 await：logout() 要先完成审计上报再清 token，
// 提前 navigate 会把请求掐断，退出就留不下痕
await logout();
function handleLogout() {
logout();
navigate("/admin/login", { replace: true });
}

3
frontend/src/constants/task.ts
View File

@ -227,8 +227,7 @@ export function overallOptionsFor(
/** 列表筛选枚举 — 两阶段并集（用于筛选，不是录入项） */
/**
* 管理角色 — 必须与后端 app/core/roles.py 的 ADMIN_ROLES 保持一致
* （后端那份已从 task_service 收敛到 core.roles，是全项目唯一事实来源）。
* 管理角色 — 必须与后端 task_service.ADMIN_ROLES 保持一致。
*
* ⚠️ 收敛到这里的理由：此前这段判断散落在多处（TaskFlowView / AdminProductsPage），
* 而移动端那份只判了 SUPER_ADMIN、漏了 SUPERVISOR，导致主管被前端误挡。

19
frontend/src/contexts/AuthContext.tsx
View File

@ -7,7 +7,7 @@ import {
useEffect,
type ReactNode,
} from "react";
import { login as loginApi, getMe, logout as logoutApi } from "../services/authApi";
import { login as loginApi, getMe } from "../services/authApi";
// ============================================================
// 类型
@ -28,8 +28,7 @@ interface AuthState {
interface AuthContextValue extends AuthState {
login: (username: string, password: string) => Promise<void>;
/** 登出。async 是因为必须先 await 审计上报、再清 token —— 顺序反了会丢日志 */
logout: () => Promise<void>;
logout: () => void;
isAuthenticated: boolean;
}
@ -114,19 +113,7 @@ export function AuthProvider({ children }: { children: ReactNode }) {
setState({ user, token: accessToken, loading: false });
}, []);
const logout = useCallback(async () => {
// ⚠️ 必须【先 await 上报、再清 token】。两边顺序反了或不等，退出就留不下痕：
// 1) axios 的请求拦截器是在微任务里执行的，它去 localStorage 读 token 时，
// 同步的 logoutInternal() 早已把 token 清掉 → 请求不带 Authorization
// → 后端只能记成「未认证」，退出归因不到人；
// 2) 调用方点完退出还会立刻 navigate 到登录页，进一步压缩执行窗口。
// 所以这里(async) + 调用方(await) 两处都得改，只改一处等于没改。
// 失败绝不影响退出：JWT 无状态，服务端本就不需要它成功。
try {
await logoutApi();
} catch {
/* 静默：断网/超时也照退不误 */
}
const logout = useCallback(() => {
logoutInternal();
}, []);

6
frontend/src/pages/ProfilePage.tsx
View File

@ -15,10 +15,8 @@ export default function ProfilePage() {
const displayName = user?.display_name || user?.username || "未知用户";
const avatarChar = displayName.charAt(0);
async function handleLogout() {
// 必须 await：logout() 要先完成审计上报再清 token，
// 提前 navigate 会把请求掐断，退出就留不下痕
await logout();
function handleLogout() {
logout();
navigate("/admin/login", { replace: true });
}

424
frontend/src/pages/admin/AdminAuditLogPage.tsx
View File

@ -1,424 +0,0 @@
/** 操作审计日志 — 谁 / 何时 / 从哪 / 对什么 / 做了什么事 / 结果如何 */
import { useCallback, useEffect, useState } from "react";
import { ScrollText, Loader2, AlertCircle, RefreshCw, Search, X, Download, BarChart3 } from "lucide-react";
import { Table, Tag, Input, Select, DatePicker, Button, Tooltip, Drawer, Descriptions } from "antd";
import type { ColumnsType } from "antd/es/table";
import dayjs, { type Dayjs } from "dayjs";
import {
fetchAuditLogs, fetchAuditOptions, exportAuditLogsCsv,
type AuditLogItem, type AuditOption,
} from "../../services/auditApi";
import { extractErrorMessage } from "../../utils/errorMessage";
import { useToast } from "../../components/ui/Toast";
import AuditUsagePanel from "./AuditUsagePanel";
import ExportColumnsModal from "../../components/admin/ExportColumnsModal";
const { RangePicker } = DatePicker;
/** HTTP 方法配色 —— 让「这是读还是写」一眼可辨 */
const METHOD_CLS: Record<string, string> = {
GET: "bg-slate-100 text-slate-600",
POST: "bg-emerald-100 text-emerald-700",
PUT: "bg-amber-100 text-amber-700",
PATCH: "bg-amber-100 text-amber-700",
DELETE: "bg-red-100 text-red-700",
};
/** 结果状态：2xx 正常 / 4xx 被拒 / 5xx 服务异常 */
function statusCls(code: number | null): string {
if (code === null) return "bg-slate-100 text-slate-500";
if (code >= 500) return "bg-red-100 text-red-700";
if (code >= 400) return "bg-orange-100 text-orange-700";
return "bg-emerald-100 text-emerald-700";
}
const PAGE_SIZE = 50;
export default function AdminAuditLogPage() {
const [rows, setRows] = useState<AuditLogItem[]>([]);
const [total, setTotal] = useState(0);
const [page, setPage] = useState(1);
const [loading, setLoading] = useState(true);
const [error, setError] = useState<string | null>(null);
const [detail, setDetail] = useState<AuditLogItem | null>(null);
const { toast } = useToast();
const [modules, setModules] = useState<AuditOption[]>([]);
const [actions, setActions] = useState<AuditOption[]>([]);
/** 导出可选列 —— 由后端下发，前端不硬编码表头 */
const [logColumns, setLogColumns] = useState<AuditOption[]>([]);
const [usageOpen, setUsageOpen] = useState(false); // 人员统计抽屉
const [exportPickerOpen, setExportPickerOpen] = useState(false);
const [exporting, setExporting] = useState(false);
// 筛选条件（user_id 用受控输入，其余即时生效）
const [userInput, setUserInput] = useState("");
const [userId, setUserId] = useState("");
const [module, setModule] = useState<string | undefined>();
const [action, setAction] = useState<string | undefined>();
const [statusCode, setStatusCode] = useState<number | undefined>();
const [range, setRange] = useState<[Dayjs, Dayjs] | null>(null);
const load = useCallback(async () => {
setLoading(true);
setError(null);
try {
const res = await fetchAuditLogs({
user_id: userId || undefined,
module,
action,
status_code: statusCode,
start_date: range?.[0]?.format("YYYY-MM-DD"),
// 后端按「含当天」处理结束日期，这里直接传所选日期即可
end_date: range?.[1]?.format("YYYY-MM-DD"),
page,
page_size: PAGE_SIZE,
});
setRows(res.items);
// total 取自后端 count 查询的真实总数，而非当前页条数
setTotal(res.total);
} catch (e) {
setError(extractErrorMessage(e));
} finally {
setLoading(false);
}
}, [userId, module, action, statusCode, range, page]);
useEffect(() => {
void load();
}, [load]);
useEffect(() => {
fetchAuditOptions()
.then((o) => {
setModules(o.modules);
setActions(o.actions);
setLogColumns(o.log_export_columns || []);
})
.catch(() => {
/* 筛选项拉取失败不影响列表本身 */
});
}, []);
/**
* 导出当前筛选条件下的明细。
* 刻意复用与列表完全相同的筛选参数 —— 导出与"看到的"必须是同一批数据，
* 否则使用者会怀疑到底哪份才是真的。
*/
async function handleExport(columns: string[]) {
setExporting(true);
try {
const truncated = await exportAuditLogsCsv({
user_id: userId || undefined,
module,
action,
status_code: statusCode,
start_date: range?.[0]?.format("YYYY-MM-DD"),
end_date: range?.[1]?.format("YYYY-MM-DD"),
columns,
});
setExportPickerOpen(false);
// 截断必须显式告知：静默少几万行比报错更危险
toast(
truncated ? "已导出，但数据超上限已被截断，请收窄筛选条件" : "已导出 CSV",
truncated ? "error" : "success",
);
} catch (e) {
toast(extractErrorMessage(e, "导出失败"), "error");
} finally {
setExporting(false);
}
}
const hasFilter = !!(userId || module || action || statusCode || range);
const resetFilters = () => {
setUserInput("");
setUserId("");
setModule(undefined);
setAction(undefined);
setStatusCode(undefined);
setRange(null);
setPage(1);
};
const columns: ColumnsType<AuditLogItem> = [
{
title: "时间",
dataIndex: "created_at",
width: 165,
render: (v: string) => (
<span className="whitespace-nowrap text-gray-600">{dayjs(v).format("YYYY-MM-DD HH:mm:ss")}</span>
),
},
{
title: "操作人",
dataIndex: "user_id",
width: 140,
render: (_, r) =>
r.user_id ? (
<div className="leading-tight">
<div className="text-gray-900">{r.display_name || r.user_id}</div>
<div className="text-xs text-gray-400">{r.user_id}</div>
</div>
) : (
<span className="text-gray-400">未认证</span>
),
},
{
title: "模块",
dataIndex: "module_label",
width: 110,
render: (v, r) => <span>{v || r.module}</span>,
},
{
title: "动作",
dataIndex: "action_label",
width: 100,
render: (v, r) => <Tag color="blue">{v || r.action}</Tag>,
},
{
title: "请求",
dataIndex: "method",
width: 210,
render: (_, r) => (
<div className="flex items-center gap-1.5">
<span className={`rounded px-1.5 py-0.5 text-xs font-mono ${METHOD_CLS[r.method || ""] || "bg-slate-100 text-slate-600"}`}>
{r.method}
</span>
<span className="truncate font-mono text-xs text-gray-500" title={r.url || ""}>
{r.url}
</span>
</div>
),
},
{
title: "结果",
dataIndex: "status_code",
width: 80,
render: (v: number | null) => <span className={`rounded px-2 py-0.5 text-xs font-mono ${statusCls(v)}`}>{v ?? "-"}</span>,
},
{
title: "来源 IP",
dataIndex: "ip_address",
width: 130,
render: (v: string | null) => <span className="font-mono text-xs text-gray-500">{v || "-"}</span>,
},
{
title: "",
key: "op",
width: 70,
render: (_, r) => (
<Button type="link" size="small" onClick={() => setDetail(r)}>
详情
</Button>
),
},
];
return (
<div className="p-6">
<div className="mb-4 flex items-center justify-between">
<div>
<h1 className="flex items-center gap-2 text-xl font-semibold text-gray-900">
<ScrollText className="h-5 w-5 text-blue-600" />
操作审计
</h1>
<p className="mt-1 text-sm text-gray-500">
所有写操作（含被拒绝的请求）自动留痕，共 {total} 条
</p>
</div>
<div className="flex items-center gap-2">
{/* 人员统计刻意做成抽屉而不是标签页：两个视图的粒度不同
（一行一次操作 vs 一人一天一行），并列成 Tab 会让筛选状态互相干扰 */}
<Button icon={<BarChart3 className="h-4 w-4" />} onClick={() => setUsageOpen(true)}>
人员统计
</Button>
<Button
icon={<Download className="h-4 w-4" />}
disabled={total === 0}
onClick={() => setExportPickerOpen(true)}
>
导出 CSV
</Button>
<Button icon={<RefreshCw className="h-4 w-4" />} onClick={() => void load()} loading={loading}>
刷新
</Button>
</div>
</div>
{/* 筛选区 */}
<div className="mb-4 flex flex-wrap items-center gap-2">
<Input
placeholder="操作人账号"
prefix={<Search className="h-4 w-4 text-gray-400" />}
value={userInput}
allowClear
style={{ width: 180 }}
onChange={(e) => setUserInput(e.target.value)}
onPressEnter={() => {
setPage(1);
setUserId(userInput.trim());
}}
onBlur={() => {
setPage(1);
setUserId(userInput.trim());
}}
/>
<Select
placeholder="模块"
allowClear
style={{ width: 140 }}
value={module}
options={modules}
onChange={(v) => {
setPage(1);
setModule(v);
}}
/>
<Select
placeholder="动作"
allowClear
style={{ width: 130 }}
value={action}
options={actions}
onChange={(v) => {
setPage(1);
setAction(v);
}}
/>
<Select
placeholder="结果"
allowClear
style={{ width: 130 }}
value={statusCode}
options={[
{ value: 200, label: "成功 (2xx/3xx)" },
{ value: 401, label: "未认证 401" },
{ value: 403, label: "无权限 403" },
{ value: 422, label: "参数错误 422" },
{ value: 500, label: "服务异常 500" },
]}
onChange={(v) => {
setPage(1);
setStatusCode(v);
}}
/>
<RangePicker
value={range}
onChange={(v) => {
setPage(1);
setRange(v as [Dayjs, Dayjs] | null);
}}
/>
{hasFilter && (
<Button icon={<X className="h-4 w-4" />} onClick={resetFilters}>
清空
</Button>
)}
</div>
{error && (
<div className="mb-4 flex items-start gap-2 rounded-lg border border-red-200 bg-red-50 p-3 text-sm text-red-700">
<AlertCircle className="mt-0.5 h-4 w-4 shrink-0" />
<span>{error}</span>
</div>
)}
<Table<AuditLogItem>
rowKey="id"
columns={columns}
dataSource={rows}
loading={loading && { indicator: <Loader2 className="h-5 w-5 animate-spin text-blue-600" /> }}
size="small"
scroll={{ x: 1000 }}
pagination={{
current: page,
pageSize: PAGE_SIZE,
total,
showSizeChanger: false,
showTotal: (t) => `共 ${t} 条`,
onChange: setPage,
}}
/>
{/* 详情抽屉：完整 URL / UA / request_id / error_message 都在这里 */}
<Drawer
title="审计详情"
width={560}
open={!!detail}
onClose={() => setDetail(null)}
>
{detail && (
<Descriptions column={1} size="small" bordered>
<Descriptions.Item label="时间">
{dayjs(detail.created_at).format("YYYY-MM-DD HH:mm:ss")}
</Descriptions.Item>
<Descriptions.Item label="操作人">
{detail.user_id ? `${detail.display_name || ""} (${detail.user_id})` : "未认证"}
</Descriptions.Item>
<Descriptions.Item label="角色">{detail.role || "-"}</Descriptions.Item>
<Descriptions.Item label="模块 / 动作">
{detail.module_label || detail.module} / {detail.action_label || detail.action}
</Descriptions.Item>
<Descriptions.Item label="目标">
{detail.target_id ? (
<>
{detail.target_name || detail.target_id}
<span className="ml-1 text-xs text-gray-400">({detail.target_type})</span>
</>
) : (
"-"
)}
</Descriptions.Item>
<Descriptions.Item label="请求">
<span className="font-mono text-xs">
{detail.method} {detail.url}
</span>
</Descriptions.Item>
<Descriptions.Item label="结果">{detail.status_code ?? "-"}</Descriptions.Item>
{detail.error_message && (
<Descriptions.Item label="错误">
<span className="break-all font-mono text-xs text-red-600">{detail.error_message}</span>
</Descriptions.Item>
)}
<Descriptions.Item label="来源 IP">
<span className="font-mono text-xs">{detail.ip_address || "-"}</span>
</Descriptions.Item>
<Descriptions.Item label="User-Agent">
<span className="break-all text-xs text-gray-500">{detail.user_agent || "-"}</span>
</Descriptions.Item>
<Descriptions.Item label="Request ID">
{detail.request_id ? (
<Tooltip title="可在后端结构化日志中用它定位同一次请求">
<span className="break-all font-mono text-xs">{detail.request_id}</span>
</Tooltip>
) : (
"-"
)}
</Descriptions.Item>
{detail.details && (
<Descriptions.Item label="变更详情">
<pre className="max-h-60 overflow-auto rounded bg-gray-50 p-2 text-xs">
{JSON.stringify(detail.details, null, 2)}
</pre>
</Descriptions.Item>
)}
</Descriptions>
)}
</Drawer>
{/* 人员统计：独立抽屉，本页表格与筛选完全不受影响 */}
<AuditUsagePanel open={usageOpen} onClose={() => setUsageOpen(false)} />
<ExportColumnsModal
open={exportPickerOpen}
columns={logColumns}
submitting={exporting}
onCancel={() => setExportPickerOpen(false)}
onConfirm={handleExport}
/>
</div>
);
}

4
frontend/src/pages/admin/AdminProductsPage.tsx
View File

@ -13,7 +13,7 @@ import {
} from "../../services/printApi";
import { useToast } from "../../components/ui/Toast";
import { useAuth } from "../../contexts/AuthContext";
import { getStatusConfig, lifecycleBadge, isAdminRole } from "../../constants/task";
import { getStatusConfig, lifecycleBadge } from "../../constants/task";
import { extractErrorMessage } from "../../utils/errorMessage";
const QR_BASE = "/api/v1/products/qrcode";
@ -31,7 +31,7 @@ interface ProductGroup { groupKey: string; products: ProductResponse[]; allInWar
export default function AdminProductsPage() {
const { toast } = useToast();
const { user: authUser } = useAuth();
const isAdmin = isAdminRole(authUser?.role);
const isAdmin = authUser?.role === "SUPER_ADMIN" || authUser?.role === "SUPERVISOR";
const [products, setProducts] = useState<ProductResponse[]>([]);
const [loading, setLoading] = useState(true);
const [error, setError] = useState<string | null>(null);

207
frontend/src/pages/admin/AuditUsagePanel.tsx
View File

@ -1,207 +0,0 @@
/**
* 人员统计（日活报表）—— 以抽屉形式挂在操作审计页旁边。
*
* 回答的是「每天有哪些人用了系统、用了多少」：
* 上线次数 / 上线时间、下线次数 / 下线时间、操作次数。
*
* 刻意不复用审计明细页的表格：两者的粒度不同（一个是一行一次操作，
* 一个是一人一天一行），合在一起筛选状态会互相干扰。
*/
import { useCallback, useEffect, useState } from "react";
import { Drawer, Table, DatePicker, Button, Alert, Tag, Empty } from "antd";
import type { ColumnsType } from "antd/es/table";
import { Download, Loader2, RefreshCw } from "lucide-react";
import dayjs, { type Dayjs } from "dayjs";
import utc from "dayjs/plugin/utc";
import {
fetchDailyUsage, fetchAuditOptions, exportDailyUsageCsv,
type DailyUsageRow, type AuditOption,
} from "../../services/auditApi";
import { extractErrorMessage } from "../../utils/errorMessage";
import { useToast } from "../../components/ui/Toast";
import ExportColumnsModal from "../../components/admin/ExportColumnsModal";
// 后端返回的是 UTC，而统计按【北京时间自然日】分组。
// 必须显式按 +08:00 渲染 —— 依赖浏览器本地时区的话，一旦有人机器不在东八区，
// 时间就会和「日期」列对不上（比如显示 17:00 而日期是次日）。
dayjs.extend(utc);
const BJ_OFFSET_MIN = 8 * 60;
function bjTime(v: string | null): string {
if (!v) return "—";
return dayjs.utc(v).utcOffset(BJ_OFFSET_MIN).format("HH:mm");
}
/** 「上线 vs 下线」次数配色：有记录就显眼，0 就淡化 */
function countTag(n: number, cls: string) {
if (!n) return <span className="text-gray-300">0</span>;
return <Tag className={`${cls} border-0 font-semibold`}>{n}</Tag>;
}
export default function AuditUsagePanel({
open,
onClose,
}: {
open: boolean;
onClose: () => void;
}) {
const { toast } = useToast();
const [rows, setRows] = useState<DailyUsageRow[]>([]);
const [loading, setLoading] = useState(false);
const [error, setError] = useState<string | null>(null);
const [range, setRange] = useState<[Dayjs, Dayjs]>([dayjs(), dayjs()]);
const [usageColumns, setUsageColumns] = useState<AuditOption[]>([]);
const [pickerOpen, setPickerOpen] = useState(false);
const [exporting, setExporting] = useState(false);
const load = useCallback(async () => {
setLoading(true);
setError(null);
try {
const res = await fetchDailyUsage({
start_date: range[0].format("YYYY-MM-DD"),
end_date: range[1].format("YYYY-MM-DD"),
});
setRows(res.items);
} catch (err: unknown) {
setError(extractErrorMessage(err, "加载使用统计失败"));
setRows([]);
} finally {
setLoading(false);
}
}, [range]);
useEffect(() => {
if (open) load();
}, [open, load]);
// 列清单只需拉一次；失败不阻断表格本身
useEffect(() => {
if (!open || usageColumns.length) return;
fetchAuditOptions()
.then((o) => setUsageColumns(o.usage_export_columns || []))
.catch(() => { /* 拉不到列清单只影响导出，不影响查看 */ });
}, [open, usageColumns.length]);
async function handleExport(columns: string[]) {
setExporting(true);
try {
const truncated = await exportDailyUsageCsv({
start_date: range[0].format("YYYY-MM-DD"),
end_date: range[1].format("YYYY-MM-DD"),
columns,
});
setPickerOpen(false);
toast(truncated ? "已导出（数据超上限，已截断）" : "已导出 CSV", truncated ? "error" : "success");
} catch (err: unknown) {
toast(extractErrorMessage(err, "导出失败"), "error");
} finally {
setExporting(false);
}
}
const multiDay = range[0].format("YYYY-MM-DD") !== range[1].format("YYYY-MM-DD");
const columns: ColumnsType<DailyUsageRow> = [
// 单日查询时日期列是冗余的，自动隐藏，少一列噪音
...(multiDay
? [{ title: "日期", dataIndex: "day", width: 110,
sorter: (a: DailyUsageRow, b: DailyUsageRow) => a.day.localeCompare(b.day) }]
: []),
{
title: "操作人", dataIndex: "display_name", width: 160,
render: (_: unknown, r: DailyUsageRow) => (
<div className="leading-tight">
<div className="text-gray-900">{r.display_name || r.user_id || "—"}</div>
{r.display_name && <div className="text-xs text-gray-400">{r.user_id}</div>}
</div>
),
},
// 上线/下线时间 = 当天首次/末次【活动】。token 有效期内用户不重新登录，
// 若取登录时间会得出"登录 0 次却操作 35 次"的矛盾数据（见后端 docstring）
{ title: "上线时间", dataIndex: "first_active_at", width: 100, align: "center",
render: (v: string | null) => <span className="font-mono text-gray-700">{bjTime(v)}</span> },
{ title: "下线时间", dataIndex: "last_active_at", width: 100, align: "center",
render: (v: string | null) => <span className="font-mono text-gray-700">{bjTime(v)}</span> },
{ title: "操作次数", dataIndex: "op_count", width: 110, align: "center",
render: (v: number) => <span className="font-bold text-blue-600">{v}</span>,
sorter: (a: DailyUsageRow, b: DailyUsageRow) => a.op_count - b.op_count,
defaultSortOrder: "descend" as const },
// 登录/登出次数是真实的手动行为计数，与上面的活动时间并列展示，不混为一谈
{ title: "登录次数", dataIndex: "login_count", width: 100, align: "center",
render: (v: number) => countTag(v, "bg-emerald-100 text-emerald-700"),
sorter: (a: DailyUsageRow, b: DailyUsageRow) => a.login_count - b.login_count },
{ title: "登出次数", dataIndex: "logout_count", width: 100, align: "center",
render: (v: number) => countTag(v, "bg-blue-100 text-blue-700"),
sorter: (a: DailyUsageRow, b: DailyUsageRow) => a.logout_count - b.logout_count },
];
return (
<Drawer
open={open}
onClose={onClose}
width={1000}
title="📊 人员统计（日活）"
extra={
<Button icon={<RefreshCw className="h-3.5 w-3.5" />} onClick={load} disabled={loading}>
刷新
</Button>
}
>
<div className="mb-4 flex flex-wrap items-center gap-2">
<DatePicker.RangePicker
value={range}
allowClear={false}
onChange={(v) => { if (v?.[0] && v?.[1]) setRange([v[0], v[1]]); }}
presets={[
{ label: "今天", value: [dayjs(), dayjs()] },
{ label: "昨天", value: [dayjs().subtract(1, "day"), dayjs().subtract(1, "day")] },
{ label: "近 7 天", value: [dayjs().subtract(6, "day"), dayjs()] },
{ label: "本月", value: [dayjs().startOf("month"), dayjs()] },
]}
/>
<Button
type="primary"
icon={<Download className="h-3.5 w-3.5" />}
disabled={rows.length === 0}
onClick={() => setPickerOpen(true)}
>
导出 CSV
</Button>
<span className="text-xs text-gray-400">
{rows.length > 0 && `共 ${rows.length} 人·天`} ｜ 时间均为北京时间
</span>
</div>
{error && (
<Alert type="error" showIcon className="mb-3" message={error} />
)}
{/* 两处口径容易被误读，直接写在表格上方 */}
<p className="mb-3 text-xs text-gray-400">
ⓘ 「上线/下线时间」= 当天首次/末次<strong>活动</strong>时间，不是登录时间 ——
登录状态可保持 7 天，当天不登录也会正常统计。
「登录/登出次数」是真实的手动登录行为计数，登出通常少于登录（关浏览器、断网不产生登出记录）。
</p>
<Table<DailyUsageRow>
rowKey={(r) => `${r.day}|${r.user_id ?? ""}`}
size="small"
columns={columns}
dataSource={rows}
loading={{ spinning: loading, indicator: <Loader2 className="h-5 w-5 animate-spin text-blue-500" /> }}
pagination={{ pageSize: 20, showSizeChanger: true, showTotal: (t) => `共 ${t} 条` }}
locale={{ emptyText: <Empty description="该时段没有使用记录" /> }}
/>
<ExportColumnsModal
open={pickerOpen}
columns={usageColumns}
submitting={exporting}
onCancel={() => setPickerOpen(false)}
onConfirm={handleExport}
/>
</Drawer>
);
}

171
frontend/src/services/auditApi.ts
View File

@ -1,171 +0,0 @@
/** 操作审计日志 API */
import api from "./api";
export interface AuditLogItem {
id: string;
user_id: string | null;
display_name: string | null;
role: string | null;
action: string;
/** 服务端补的中文标签，前端不再各自维护枚举映射 */
action_label: string | null;
module: string;
module_label: string | null;
target_type: string | null;
target_id: string | null;
target_name: string | null;
details: Record<string, unknown> | null;
ip_address: string | null;
user_agent: string | null;
method: string | null;
url: string | null;
status_code: number | null;
error_message: string | null;
/** 拿着它可在后端结构化日志中定位同一次请求 */
request_id: string | null;
created_at: string;
}
export interface AuditLogListResponse {
items: AuditLogItem[];
total: number;
}
export interface AuditOption {
value: string;
label: string;
}
export interface AuditOptionsResponse {
modules: AuditOption[];
actions: AuditOption[];
/** 导出可选列（value=后端列 key，label=中文表头）—— 由后端下发，前端不再硬编码 */
log_export_columns: AuditOption[];
usage_export_columns: AuditOption[];
}
/** 日活统计的单行（某人在某一天的用量） */
export interface DailyUsageRow {
day: string;
user_id: string | null;
display_name: string | null;
role: string | null;
login_count: number;
logout_count: number;
op_count: number;
/**
* 上线 / 下线时间 = 当天首次 / 末次【活动】时间（不是登录时间）。
* token 有效期内用户不重新登录，按登录算会得出"登录 0 次却操作 35 次"的矛盾数据。
* ISO(UTC)，展示前必须转北京时间，否则会和 day 列对不上。
*/
first_active_at: string | null;
last_active_at: string | null;
}
export interface DailyUsageResponse {
start_date: string;
end_date: string;
items: DailyUsageRow[];
total: number;
}
export interface AuditLogQuery {
user_id?: string;
module?: string;
action?: string;
target_id?: string;
request_id?: string;
status_code?: number;
/** YYYY-MM-DD */
start_date?: string;
/** YYYY-MM-DD（含当天） */
end_date?: string;
page?: number;
page_size?: number;
}
/** 分页查询审计日志（按时间倒序） */
export async function fetchAuditLogs(q: AuditLogQuery = {}): Promise<AuditLogListResponse> {
const params = Object.fromEntries(
Object.entries(q).filter(([, v]) => v !== undefined && v !== null && v !== "")
);
const { data } = await api.get<AuditLogListResponse>("/audit/logs", { params });
return data;
}
/** 获取模块/动作筛选项（含导出可选列） */
export async function fetchAuditOptions(): Promise<AuditOptionsResponse> {
const { data } = await api.get<AuditOptionsResponse>("/audit/options");
return data;
}
/** 日活 / 使用统计 —— 按【北京时间自然日 × 操作人】聚合 */
export async function fetchDailyUsage(params: {
start_date?: string;
end_date?: string;
} = {}): Promise<DailyUsageResponse> {
const { data } = await api.get<DailyUsageResponse>("/audit/daily-usage", { params });
return data;
}
/**
* 触发浏览器下载一个 CSV。
*
* ⚠️ 不能直接用 <a href="/api/..."> 或 window.open：本项目是 Bearer Token 鉴权
* （token 在 localStorage，不在 Cookie），普通链接带不上 Authorization 头，
* 后端会直接 401。必须先经 axios 取回 blob 再本地落盘。
*
* @returns 是否因超出后端行数上限而被截断（调用方据此提示用户，不要静默）
*/
async function downloadCsv(
path: string,
params: Record<string, unknown>,
filename: string,
): Promise<boolean> {
const resp = await api.get(path, { params, responseType: "blob" });
const url = URL.createObjectURL(resp.data as Blob);
const a = document.createElement("a");
a.href = url;
a.download = filename;
document.body.appendChild(a);
a.click();
a.remove();
URL.revokeObjectURL(url);
return resp.headers["x-export-truncated"] === "1";
}
/** 导出审计明细（列可自定义，columns 为后端列 key 数组；不传=全部列） */
export function exportAuditLogsCsv(
q: AuditLogQuery & { columns?: string[] },
): Promise<boolean> {
const { columns, ...rest } = q;
return downloadCsv(
"/audit/logs/export",
{ ...clean(rest), columns: columns?.join(",") },
"audit_logs.csv",
);
}
/** 导出日活统计（每人一行，列可自定义） */
export function exportDailyUsageCsv(
params: { start_date?: string; end_date?: string; columns?: string[] },
): Promise<boolean> {
const { columns, ...rest } = params;
return downloadCsv(
"/audit/daily-usage/export",
{ ...clean(rest), columns: columns?.join(",") },
"daily_usage.csv",
);
}
/** 去掉 undefined / null / 空串，避免拼出 ?a=&b= 这类空参数 */
function clean(o: Record<string, unknown>): Record<string, unknown> {
return Object.fromEntries(
Object.entries(o).filter(([, v]) => v !== undefined && v !== null && v !== "")
);
}

18
frontend/src/services/authApi.ts
View File

@ -1,5 +1,4 @@
/** 认证 API — 登录、刷新 Token、获取用户信息、登出留痕 */
import api from "./api";
/** 认证 API — 登录、刷新 Token、获取用户信息 */
import type { UserInfo } from "../contexts/AuthContext";
import { extractErrorMessage } from "../utils/errorMessage";
@ -53,18 +52,3 @@ export async function getMe(token: string): Promise<UserInfo> {
}
return res.json();
}
/**
* 登出上报 —— 唯一目的是【审计留痕】。
*
* JWT 无状态，服务端不会（也无法）吊销令牌，本地清 token 就是登出。
* 但没有这个请求，前端的「退出」动作在审计里完全不可见，所以必须上报一次。
*
* ⚠️ 调用方必须 **await 本函数之后**才清 localStorage 与跳转：
* axios 的请求拦截器在微任务里执行、需要现读 localStorage 取 token。
* 若不等就同步清空并 navigate，请求会不带 Authorization（或直接被掐断），
* 后端只能记成「未认证」，退出归因不到人 —— 实测审计里 logout 记录为 0。
*/
export async function logout(): Promise<void> {
await api.post("/auth/logout");
}

21
track-uniapp/src/pages/profile/settings.vue
View File

@ -38,7 +38,6 @@
<script setup>
import { ref, computed, onMounted } from "vue";
import { checkAppUpdate } from "../../utils/ota";
import { post } from "../../utils/request";
// ============================================================
// 缓存清理策略 —— 黑名单式：只删「明确登记过的业务缓存」
@ -168,31 +167,15 @@ async function handleCheckUpdate() {
await checkAppUpdate({ manual: true });
}
async function handleLogout() {
function handleLogout() {
// 退出是不可逆的（要重新输账号密码），按车间使用场景加一道确认防误触
uni.showModal({
title: "退出登录",
content: "退出后需要重新输入账号密码，确定退出吗？",
confirmText: "退出",
cancelText: "取消",
success: async (res) => {
success: (res) => {
if (!res.confirm) return;
// 🔴 必须【先 await 上报、再清 token】——两者顺序反了或不等，退出就留不下痕：
// 1) uni.reLaunch 会销毁页面上下文，直接掐断尚未发出的 uni.request；
// 2) 而 request.js 是在发送时才从 storage 读 access_token，
// 先清 storage 的话请求会不带 Authorization，后端只能记成「未认证」。
// 这里刻意 try/catch 兜住：上报失败（断网/超时）也绝不能挡住用户退出。
uni.showLoading({ title: "退出中...", mask: true });
try {
await post("/auth/logout");
} catch (e) {
// 静默：JWT 无状态，服务端本就不需要它成功
console.warn("[logout] 上报失败（不影响退出）", e);
} finally {
uni.hideLoading();
}
try {
uni.removeStorageSync("token");
uni.removeStorageSync("access_token");