lofyerandfactory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com>
71db82393a
Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com>
254 lines
8.3 KiB
Python
254 lines
8.3 KiB
Python
"""
|
||
FastAPI 主应用入口
|
||
报告管理系统：API + 前端静态文件，单端口服务
|
||
增加 JWT 认证保护
|
||
"""
|
||
import logging
|
||
from contextlib import asynccontextmanager
|
||
from pathlib import Path
|
||
|
||
from fastapi import FastAPI, Request, Depends, Response
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
from fastapi.staticfiles import StaticFiles
|
||
from fastapi.responses import FileResponse, JSONResponse
|
||
|
||
from .config import OUTPUT_DIR, STATIC_DIR, PROJECT_ROOT
|
||
from .api.era2_routes import router as era2_router
|
||
from .api.era2_state import era2_state
|
||
from .auth import (
|
||
LoginRequest, TokenResponse,
|
||
authenticate_user, create_access_token, get_current_user, verify_token,
|
||
)
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s [%(levelname)s] %(name)s - %(message)s",
|
||
datefmt="%H:%M:%S",
|
||
)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 前端构建产物目录
|
||
FRONTEND_DIST = PROJECT_ROOT / "frontend" / "dist"
|
||
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(app: FastAPI):
|
||
"""应用启动/关闭生命周期"""
|
||
logger.info("🚀 正在初始化全市数据引擎...")
|
||
era2_state.initialize()
|
||
logger.info(f"✅ 数据引擎就绪: {len(era2_state.schools)}所学校, {len(era2_state.districts)}个区")
|
||
|
||
if FRONTEND_DIST.exists():
|
||
logger.info(f"📦 前端静态文件: {FRONTEND_DIST}")
|
||
else:
|
||
logger.warning(f"⚠️ 前端静态文件不存在: {FRONTEND_DIST}，请先 cd frontend && npm run build")
|
||
|
||
logger.info("🔐 认证已启用，所有 API 需要登录访问")
|
||
yield
|
||
logger.info("👋 应用关闭")
|
||
|
||
|
||
import os
|
||
|
||
# 生产环境关闭 API 文档（设 DOCS_ENABLED=1 可临时打开）
|
||
_docs_enabled = os.environ.get("DOCS_ENABLED", "0") == "1"
|
||
|
||
app = FastAPI(
|
||
title="课程实施监测报告管理系统",
|
||
description="上海市高中课程实施监测数据分析与报告生成 API",
|
||
version="2.0.0",
|
||
lifespan=lifespan,
|
||
docs_url="/docs" if _docs_enabled else None,
|
||
redoc_url="/redoc" if _docs_enabled else None,
|
||
openapi_url="/openapi.json" if _docs_enabled else None,
|
||
)
|
||
|
||
# CORS — 收紧配置（部署时按需修改 allow_origins）
|
||
ALLOWED_ORIGINS = [
|
||
"http://localhost:5173", # 前端开发服务器
|
||
"http://localhost:7777", # 后端自身
|
||
"http://127.0.0.1:5173",
|
||
"http://127.0.0.1:7777",
|
||
]
|
||
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=ALLOWED_ORIGINS,
|
||
allow_credentials=True,
|
||
allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
|
||
allow_headers=["*"],
|
||
)
|
||
|
||
|
||
# ==================== 无需认证的路由 ====================
|
||
|
||
# 登录接口
|
||
@app.post("/api/auth/login", response_model=TokenResponse)
|
||
async def login(req: LoginRequest, response: Response):
|
||
"""用户登录，返回 JWT token"""
|
||
username = authenticate_user(req.username, req.password)
|
||
if not username:
|
||
return JSONResponse(
|
||
status_code=401,
|
||
content={"detail": "用户名或密码错误"},
|
||
)
|
||
|
||
token, expires_in = create_access_token(username)
|
||
|
||
# 同时设置 Cookie（方便浏览器直接访问静态资源）
|
||
response.set_cookie(
|
||
key="access_token",
|
||
value=token,
|
||
max_age=expires_in,
|
||
httponly=True,
|
||
samesite="lax",
|
||
# secure=True, # 生产环境使用 HTTPS 时取消注释
|
||
)
|
||
|
||
logger.info(f"🔑 用户 '{username}' 登录成功")
|
||
return TokenResponse(access_token=token, expires_in=expires_in)
|
||
|
||
|
||
# 验证 token 是否有效
|
||
@app.get("/api/auth/verify")
|
||
async def verify_auth(request: Request):
|
||
"""验证当前 token 是否有效（前端刷新页面时调用）"""
|
||
# 从 header 或 cookie 中提取 token
|
||
token = None
|
||
auth_header = request.headers.get("authorization", "")
|
||
if auth_header.startswith("Bearer "):
|
||
token = auth_header[7:]
|
||
if not token:
|
||
token = request.cookies.get("access_token")
|
||
if not token:
|
||
return JSONResponse(status_code=401, content={"valid": False})
|
||
|
||
username = verify_token(token)
|
||
if not username:
|
||
return JSONResponse(status_code=401, content={"valid": False})
|
||
|
||
return {"valid": True, "username": username}
|
||
|
||
|
||
# 登出
|
||
@app.post("/api/auth/logout")
|
||
async def logout(response: Response):
|
||
"""清除认证 Cookie"""
|
||
response.delete_cookie("access_token")
|
||
return {"message": "已退出登录"}
|
||
|
||
|
||
# ==================== 认证中间件 ====================
|
||
|
||
# 不需要认证的路径前缀（API 文档不对外暴露）
|
||
PUBLIC_PATHS = {
|
||
"/api/auth/login",
|
||
"/api/auth/logout",
|
||
"/api/auth/verify",
|
||
}
|
||
|
||
# 前端静态资源路径前缀（不需要 API 级别认证，前端自己处理路由守卫）
|
||
STATIC_PREFIXES = ("/assets/", "/favicon", "/vite.svg")
|
||
|
||
|
||
@app.middleware("http")
|
||
async def auth_middleware(request: Request, call_next):
|
||
"""
|
||
全局认证中间件：
|
||
- 公开路径（登录、静态资源）直接放行
|
||
- API 路径需要有效的 JWT token
|
||
- 前端 SPA 页面路径放行（由前端路由守卫处理）
|
||
"""
|
||
path = request.url.path
|
||
|
||
# 1. 公开 API 路径 — 放行
|
||
if path in PUBLIC_PATHS:
|
||
return await call_next(request)
|
||
|
||
# 2. 前端静态资源 — 放行
|
||
if any(path.startswith(p) for p in STATIC_PREFIXES):
|
||
return await call_next(request)
|
||
|
||
# 3. API 路径 — 需要认证
|
||
if path.startswith("/api/") or path.startswith("/output/"):
|
||
token = None
|
||
# 来源1: Authorization header
|
||
auth_header = request.headers.get("authorization", "")
|
||
if auth_header.startswith("Bearer "):
|
||
token = auth_header[7:]
|
||
# 来源2: Cookie
|
||
if not token:
|
||
token = request.cookies.get("access_token")
|
||
# 来源3: URL query param（供 <a href> / <iframe src> 等无法设 header 的场景）
|
||
if not token:
|
||
token = request.query_params.get("token")
|
||
|
||
if not token or not verify_token(token):
|
||
return JSONResponse(
|
||
status_code=401,
|
||
content={"detail": "未授权访问，请先登录"},
|
||
)
|
||
|
||
# 4. 其他路径（前端 SPA 页面）— 放行，由前端路由守卫处理
|
||
response = await call_next(request)
|
||
|
||
# 安全响应头
|
||
response.headers["X-Content-Type-Options"] = "nosniff"
|
||
response.headers["X-Frame-Options"] = "SAMEORIGIN"
|
||
response.headers["X-XSS-Protection"] = "1; mode=block"
|
||
response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
|
||
# 防止浏览器缓存敏感 API 响应
|
||
if path.startswith("/api/"):
|
||
response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate"
|
||
response.headers["Pragma"] = "no-cache"
|
||
|
||
return response
|
||
|
||
|
||
# ==================== 业务路由（全部需要认证） ====================
|
||
|
||
# 注册 API 路由
|
||
app.include_router(era2_router, prefix="/api")
|
||
|
||
# 挂载 output 静态文件（报告 HTML/JSON）— 已由中间件保护
|
||
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
|
||
app.mount("/output", StaticFiles(directory=str(OUTPUT_DIR)), name="output")
|
||
|
||
# 挂载 backend/static（如有）
|
||
if STATIC_DIR.exists():
|
||
app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")
|
||
|
||
# 挂载前端静态资源（JS/CSS 等）
|
||
if FRONTEND_DIST.exists():
|
||
assets_dir = FRONTEND_DIST / "assets"
|
||
if assets_dir.exists():
|
||
app.mount("/assets", StaticFiles(directory=str(assets_dir)), name="frontend-assets")
|
||
|
||
|
||
# SPA Fallback：所有未匹配的路由返回前端 index.html
|
||
@app.get("/{full_path:path}")
|
||
async def serve_frontend(request: Request, full_path: str):
|
||
"""
|
||
SPA 路由兜底：
|
||
- 如果请求的是 dist 目录下的真实文件（如 favicon.ico），直接返回
|
||
- 否则返回 index.html，让前端路由处理
|
||
"""
|
||
if FRONTEND_DIST.exists():
|
||
# 尝试匹配真实文件（防路径穿越: resolve 后必须在 FRONTEND_DIST 内）
|
||
file_path = (FRONTEND_DIST / full_path).resolve()
|
||
if file_path.is_file() and str(file_path).startswith(str(FRONTEND_DIST.resolve())):
|
||
return FileResponse(str(file_path))
|
||
|
||
# SPA fallback → index.html
|
||
index_path = FRONTEND_DIST / "index.html"
|
||
if index_path.exists():
|
||
return FileResponse(str(index_path))
|
||
|
||
# 前端未构建时返回 API 信息
|
||
return {
|
||
"name": "课程实施监测报告管理系统",
|
||
"version": "2.0.0",
|
||
"status": "running",
|
||
"message": "前端未构建，请访问 /api/auth/login 登录后使用 API",
|
||
}
|