From 4454047ce3756bd9d18c76fd102d6b90563fdf2f Mon Sep 17 00:00:00 2001 From: openhands Date: Mon, 21 Sep 2026 01:57:20 +0000 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E4=BB=BB=E5=8A=A1?= =?UTF-8?q?=E5=88=86=E9=A1=B5=E6=80=BB=E6=95=B0=E9=94=99=E8=AF=AF=20+=20?= =?UTF-8?q?=E8=A1=A5=E9=BD=90=E5=8F=AF=E8=A7=82=E6=B5=8B=E6=80=A7=E5=9F=BA?= =?UTF-8?q?=E5=BB=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 分页总数（#5）： - get_all_tasks 的 total 原为 len(flat_tasks)（当前页条数），移动端 「我的任务」用 tasks.length < total 判断 hasMore，首页满员时恒为 false，列表永远停在第一页 20 条。改为独立 COUNT 查询（与 notifications.py 已有写法保持一致）。 可观测性（#9）： - 新增 core/logging.py：单行 JSON 结构化日志 + request_id/user 上下文 注入；零第三方依赖；接管 uvicorn 自带 handler 避免格式绕过。 - 新增 core/middleware.py：RequestContextMiddleware 生成/透传 X-Request-ID 并回写响应头，输出含耗时/用户的结构化访问日志。 - 新增 core/health.py：拆分存活/就绪探针。/health/live 不触依赖； /health/ready 探主库，不可用返回 503；MOM 挂掉仅降级不摘流量。 - main.py：全局异常处理器只把堆栈写日志，响应体仅回 request_id； 接入可选 Sentry（未装 SDK 时静默跳过）。 - auth_service：解析 Token 后写入 user 上下文，日志自动带操作人。 注：异常处理器由 ServerErrorMiddleware 调用，此时 contextvar 已被重置， 故 request_id 同时写入 request.state（由 ASGI scope 承载）再读取。 已用 TestClient 验证：探针状态码/检查项、X-Request-ID 透传与生成、 500 响应携带可对账的 request_id 且不泄露堆栈。 --- backend/.env.example | 8 +++ backend/app/core/config.py | 11 ++++ backend/app/core/health.py | 93 ++++++++++++++++++++++++++++ backend/app/core/logging.py | 89 ++++++++++++++++++++++++++ backend/app/core/middleware.py | 83 +++++++++++++++++++++++++ backend/app/main.py | 85 ++++++++++++++++++++++--- backend/app/services/auth_service.py | 5 ++ backend/app/services/task_service.py | 33 +++++++--- 8 files changed, 389 insertions(+), 18 deletions(-) create mode 100644 backend/app/core/health.py create mode 100644 backend/app/core/logging.py create mode 100644 backend/app/core/middleware.py diff --git a/backend/.env.example b/backend/.env.example index 7d5a240..e501671 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -3,3 +3,11 @@ 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 diff --git a/backend/app/core/config.py b/backend/app/core/config.py index b4eb988..4fef81d 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -16,6 +16,17 @@ 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"]' diff --git a/backend/app/core/health.py b/backend/app/core/health.py new file mode 100644 index 0000000..b535b00 --- /dev/null +++ b/backend/app/core/health.py @@ -0,0 +1,93 @@ +"""健康检查 — 存活探针与就绪探针分离 + +为什么必须拆开： +- 存活探针（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, + ) diff --git a/backend/app/core/logging.py b/backend/app/core/logging.py new file mode 100644 index 0000000..e49c06f --- /dev/null +++ b/backend/app/core/logging.py @@ -0,0 +1,89 @@ +"""结构化日志 — 单行 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 diff --git a/backend/app/core/middleware.py b/backend/app/core/middleware.py new file mode 100644 index 0000000..c530619 --- /dev/null +++ b/backend/app/core/middleware.py @@ -0,0 +1,83 @@ +"""请求上下文中间件 — 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 + +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 + return response + finally: + # 异常路径也要留下访问记录，否则接口 500 时日志里反而没有痕迹 + if not logged: + self._log_access(request, status_code, started) + request_id_var.reset(rid_token) + user_var.reset(user_token) + + 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) + + 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_var.get(), + } + }, + ) diff --git a/backend/app/main.py b/backend/app/main.py index 1f82a63..f672de9 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -1,22 +1,63 @@ +import logging from contextlib import asynccontextmanager -from fastapi import FastAPI +from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import JSONResponse from app.core.config import settings +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="0.1.0", + version=settings.APP_VERSION, lifespan=lifespan, ) @@ -27,12 +68,38 @@ app.add_middleware( allow_credentials=True, allow_methods=["*"], allow_headers=["*"], + # 暴露给浏览器 JS 读取：前端报错时才能把 request_id 一起带上便于对账 + expose_headers=["X-Request-ID"], ) +# Starlette 的 add_middleware 是「后添加者在外层」，故 RequestContextMiddleware +# 最后注册 → 最先进入请求，才能覆盖包括 CORS 预检在内的全部请求并回写 X-Request-ID +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}, + ) + + # ---- 注册路由 ---- app.include_router(api_router, prefix="/api/v1") - - -@app.get("/health") -async def health_check(): - return {"status": "ok", "version": "0.1.0"} +# 健康检查挂在根路径（/health*），运维探针不经过 /api/v1 +app.include_router(health_router) diff --git a/backend/app/services/auth_service.py b/backend/app/services/auth_service.py index 170d614..33f2bb5 100644 --- a/backend/app/services/auth_service.py +++ b/backend/app/services/auth_service.py @@ -14,6 +14,7 @@ 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() @@ -129,6 +130,10 @@ async def get_current_user( detail="请使用 Access Token 访问 API，Refresh Token 仅用于刷新", ) + # 注入日志上下文：此后本请求的所有日志都会自动带上操作人。 + # username 即 assignee_id 口径，排查「谁干了什么」时比数字 id 直观得多。 + user_var.set(payload.get("username") or user_id) + return payload except JWTError: raise HTTPException(status_code=401, detail="无效的 Token") diff --git a/backend/app/services/task_service.py b/backend/app/services/task_service.py index be9b6f3..a2bf8d1 100644 --- a/backend/app/services/task_service.py +++ b/backend/app/services/task_service.py @@ -3,7 +3,7 @@ from __future__ import annotations import json import uuid from fastapi import HTTPException, status -from sqlalchemy import select, delete +from sqlalchemy import select, delete, func from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload @@ -457,22 +457,37 @@ async def get_all_tasks( assignee_id: str | None = None, skip: int = 0, limit: int = 50 ) -> TaskListResponse: """获取任务列表，可按产品/负责人筛选""" - stmt = select(Task).options( - selectinload(Task.records), - selectinload(Task.product), - ) + filters = [] if product_id: - stmt = stmt.where(Task.product_id == product_id) + filters.append(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()) + 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()) + ) 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=len(flat_tasks)) + return TaskListResponse(tasks=flat_tasks, total=total) # ============================================================