import asyncio import logging import re import time from contextlib import asynccontextmanager from collections import defaultdict from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from sqlalchemy import text from app.api.v1.router import api_router from app.core.config import get_cors_allowed_origins, settings from app.core.exceptions import register_exception_handlers from app.core.login_crypto import validate_login_crypto_configuration from app.crud.user import ensure_admin_exists from app.db.base import Base from app.db.session import SessionLocal, engine from app.services.visit_scheduler import run_daily_lost_visit_job from app.services.permission_log_writer import start_log_writer, stop_log_writer from app.services.permission_metric_aggregator import run_hourly_metric_aggregation from app.services.security_access_log_writer import ( get_security_log_writer, start_security_log_writer, stop_security_log_writer, ) from app.core.security import decode_token logger = logging.getLogger("ctms.setup_config") UUID_RE = re.compile( r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}" ) setup_config_stats: dict[str, int] = defaultdict(int) @asynccontextmanager async def lifespan(_: FastAPI): stop_event = asyncio.Event() scheduler_task = asyncio.create_task(run_daily_lost_visit_job(stop_event)) aggregator_task = asyncio.create_task(run_hourly_metric_aggregation(stop_event)) await start_log_writer() await start_security_log_writer() if settings.ENV == "development": async with engine.begin() as conn: await _ensure_legacy_primary_keys(conn) await conn.run_sync(Base.metadata.create_all) async with SessionLocal() as session: await ensure_admin_exists(session) yield stop_event.set() await stop_log_writer() await stop_security_log_writer() await scheduler_task await aggregator_task async def _ensure_legacy_primary_keys(conn) -> None: # Fix legacy databases where tables with an "id" column were created without a PK. result = await conn.execute( text( """ SELECT c.relname AS table_name FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace WHERE n.nspname = 'public' AND c.relkind = 'r' AND EXISTS ( SELECT 1 FROM pg_attribute a WHERE a.attrelid = c.oid AND a.attname = 'id' AND a.attisdropped = false ) AND NOT EXISTS ( SELECT 1 FROM pg_constraint con WHERE con.conrelid = c.oid AND con.contype = 'p' ) """ ) ) tables = [row.table_name for row in result.fetchall()] for table_name in tables: await conn.execute( text(f'ALTER TABLE "{table_name}" ADD CONSTRAINT "{table_name}_pkey" PRIMARY KEY (id)') ) def create_app() -> FastAPI: validate_login_crypto_configuration() app = FastAPI( title="CTMS 后端 API", description="临床试验项目管理系统后端接口文档", version="0.1.0", debug=settings.ENV == "development", lifespan=lifespan, openapi_tags=[ {"name": "auth", "description": "认证与登录"}, {"name": "users", "description": "用户管理"}, {"name": "studies", "description": "项目管理"}, {"name": "sites", "description": "中心管理"}, {"name": "study-members", "description": "项目成员"}, {"name": "attachments", "description": "通用附件"}, {"name": "audit-logs", "description": "审计日志"}, {"name": "dashboard", "description": "项目总览与统计"}, {"name": "subjects", "description": "参与者"}, {"name": "visits", "description": "访视"}, {"name": "aes", "description": "不良事件"}, {"name": "finance", "description": "费用管理"}, {"name": "faq-categories", "description": "FAQ 分类"}, {"name": "faqs", "description": "FAQ 条目"}, {"name": "health", "description": "健康检查"}, ], ) app.add_middleware( CORSMiddleware, allow_origins=get_cors_allowed_origins(), allow_credentials=True, allow_methods=["*"], allow_headers=[ "Accept", "Authorization", "Content-Type", "X-CTMS-Client-Type", "X-CTMS-Client-Version", "X-CTMS-Client-Platform", "X-CTMS-Build-Channel", "X-CTMS-Build-Commit", ], ) @app.middleware("http") async def setup_config_monitoring_middleware(request, call_next): path = request.url.path is_setup_config_path = "/api/v1/studies/" in path and "/setup-config" in path should_security_log = path.startswith("/api/") started_at = time.perf_counter() status_code = 500 try: response = await call_next(request) except Exception: if is_setup_config_path: normalized_path = UUID_RE.sub("{study_id}", path) duration_ms = int((time.perf_counter() - started_at) * 1000) key = f"{request.method} {normalized_path} 5xx" setup_config_stats[key] += 1 logger.exception( "setup_config_request_error method=%s path=%s status=%s duration_ms=%s", request.method, normalized_path, 500, duration_ms, ) if should_security_log: _enqueue_security_access_log(request, path, status_code, started_at) raise status_code = int(response.status_code) if is_setup_config_path: normalized_path = UUID_RE.sub("{study_id}", path) duration_ms = int((time.perf_counter() - started_at) * 1000) status_bucket = f"{status_code // 100}xx" key = f"{request.method} {normalized_path} {status_bucket}" setup_config_stats[key] += 1 if status_code >= 500: logger.error( "setup_config_request method=%s path=%s status=%s duration_ms=%s", request.method, normalized_path, status_code, duration_ms, ) elif status_code >= 400: logger.warning( "setup_config_request method=%s path=%s status=%s duration_ms=%s", request.method, normalized_path, status_code, duration_ms, ) else: logger.info( "setup_config_request method=%s path=%s status=%s duration_ms=%s", request.method, normalized_path, status_code, duration_ms, ) if should_security_log: _enqueue_security_access_log(request, path, status_code, started_at) return response register_exception_handlers(app) @app.get( "/", tags=["health"], summary="服务入口说明", description="返回后端服务入口说明,避免将根路径误认为前端页面。", ) async def root() -> dict[str, str]: return { "service": "ctms-backend", "status": "ok", "health": "/health", "docs": "/docs", "api": "/api/v1", } @app.get( "/health", tags=["health"], summary="健康检查", description="返回服务存活状态。", ) async def health() -> dict[str, str]: return {"status": "ok"} @app.get( "/health/setup-config-stats", tags=["health"], summary="立项配置接口监控统计", description="返回当前进程内立项配置接口访问计数(按接口+状态段汇总)。", ) async def setup_config_health_stats() -> dict[str, dict[str, int]]: summary: dict[str, int] = dict(setup_config_stats) totals = { "2xx": sum(v for k, v in summary.items() if k.endswith(" 2xx")), "4xx": sum(v for k, v in summary.items() if k.endswith(" 4xx")), "5xx": sum(v for k, v in summary.items() if k.endswith(" 5xx")), } return {"totals": totals, "by_endpoint": summary} app.include_router(api_router, prefix="/api/v1") return app app = create_app() def _resolve_client_ip(request) -> str | None: forwarded = request.headers.get("x-forwarded-for") if forwarded: return forwarded.split(",")[0].strip() real_ip = request.headers.get("x-real-ip") if real_ip: return real_ip.strip() return request.client.host if request.client else None def _resolve_auth_context(request) -> tuple[str, str | None]: authorization = request.headers.get("authorization") or "" if not authorization.lower().startswith("bearer "): return "ANONYMOUS", None token = authorization.split(" ", 1)[1].strip() if not token: return "ANONYMOUS", None try: payload = decode_token(token) except Exception: return "INVALID_TOKEN", None subject = payload.get("sub") return ("AUTHENTICATED", str(subject)) if subject else ("INVALID_TOKEN", None) def _enqueue_security_access_log(request, path: str, status_code: int, started_at: float) -> None: writer = get_security_log_writer() if not writer: return auth_status, user_identifier = _resolve_auth_context(request) writer.enqueue( { "method": request.method, "path": path, "status_code": status_code, "elapsed_ms": round((time.perf_counter() - started_at) * 1000, 2), "client_ip": _resolve_client_ip(request), "user_agent": request.headers.get("user-agent"), "client_type": request.headers.get("x-ctms-client-type"), "client_version": request.headers.get("x-ctms-client-version"), "client_platform": request.headers.get("x-ctms-client-platform"), "build_channel": request.headers.get("x-ctms-build-channel"), "build_commit": request.headers.get("x-ctms-build-commit"), "auth_status": auth_status, "user_identifier": user_identifier, } )