影响程度 ↑
        │
     高 │  LLM 接口被高频刷费
        │  ┌─────────────────────────────────┐
        │  │ 防护：速率限制（核心防线）        │
        │  │ 检测：API 调用量监控 + 阈值告警  │
        │  └─────────────────────────────────┘
        │
     中 │  金融接口被轮询   管理员密码暴力破解
        │  ┌─────────────────┐  ┌──────────────────┐
        │  │ 防护：速率限制   │  │ 防护：服务端会话  │
        │  │ Redis 缓存兜底   │  │  bcrypt 密码哈希  │
        │  └─────────────────┘  └──────────────────┘
        │
     低 │  文章内容被批量搬运
        │  ┌─────────────────────────────────────┐
        │  │ 不设防——公开内容，无实际损失        │
        │  └─────────────────────────────────────┘
        │
        └───────────────────────────────────────────→ 发生概率
            低                 中                 高

双维度识别策略：

                  ┌── 携带 Cookie ──→ Cookie 维度（精确到个体）
                  │
  请求进入中间件 ──┤
                  │
                  └── 无 Cookie ──→ IP 维度（兜底）

RATE_LIMITS = [
    ("/api/chat/", 5, 60),
]
# 修改阈值 → 修改代码 → 提交 → CI → 部署

# config.py
RATE_LIMIT_CHAT = os.getenv("RATE_LIMIT_CHAT", "5,60")

# .env
RATE_LIMIT_CHAT=10,30
# 修改阈值 → 编辑 .env → 重启服务

count = redis.incr(key)       # 原子递增，返回新值
if count == 1:
    redis.expire(key, window) # 首次请求，设 TTL
return count > max_reqs       # True = 已超限

now = time.time()
timestamps = memory.get(key, [])
# 过滤出仍在窗口内的时间戳
timestamps = [t for t in timestamps if now - t < window]
if len(timestamps) >= max_reqs:
    return True  # 超限
timestamps.append(now)
memory[key] = timestamps
return False     # 未超限

主路径：Redis（推荐）
    ├── 读取：INCR（原子递增 + 返回当前值）
    ├── 过期：EXPIRE（窗口结束后自动删除）
    ├── 优势：分布式多实例共享同一计数器
    └── 劣势：Redis 服务不可用时需要降级

降级路径：进程内存
    ├── 数据结构：dict[str, list[timestamp]]
    ├── 清理：每 100 次请求扫描并删除过期 key
    ├── 优势：零外部依赖，启动即可用
    └── 劣势：多实例各自独立计数、重启丢失

# app/main.py — 注册中间件

from app.middleware.rate_limit import RateLimitMiddleware

# 中间件洋葱模型：RateLimitMiddleware 在最后注册，成为最外层
app.add_middleware(CORSMiddleware, allow_origins=["*"], ...)
app.add_middleware(SessionMiddleware, secret_key=...)
app.add_middleware(RateLimitMiddleware)  # 最外层，请求最先到达

# app/middleware/rate_limit.py — 从 config 导入并构建规则表

from app.config import (
    RATE_LIMIT_ENABLED,
    RATE_LIMIT_CHAT,
    RATE_LIMIT_INTERVIEW,
    RATE_LIMIT_MARKET,
    RATE_LIMIT_ADMIN,
    RATE_LIMIT_API,
    RATE_LIMIT_CRAWLER_UA,
)

def _parse_limit(val: str):
    """解析 'max_reqs,window_seconds' 格式，空字符串返回 None"""
    val = val.strip()
    if not val:
        return None
    parts = val.split(",")
    if len(parts) != 2:
        return None
    try:
        return int(parts[0].strip()), int(parts[1].strip())
    except ValueError:
        return None

def _build_rate_limits():
    """从配置逐条构建限流规则表"""
    rules = [
        ("/api/chat/", RATE_LIMIT_CHAT),
        ("/api/interview/", RATE_LIMIT_INTERVIEW),
        ("/api/market/", RATE_LIMIT_MARKET),
        ("/api/admin/", RATE_LIMIT_ADMIN),
    ]
    result = []
    for prefix, raw in rules:
        parsed = _parse_limit(raw)
        if parsed:
            result.append((prefix, parsed[0], parsed[1]))
    api_limit = _parse_limit(RATE_LIMIT_API)
    if api_limit:
        result.append(("/api/", api_limit[0], api_limit[1]))
    return result

# 模块加载时构建一次，后续配置变更需重启服务
RATE_LIMITS = _build_rate_limits()
CRAWLER_UA_KEYWORDS = [
    kw.strip().lower()
    for kw in RATE_LIMIT_CRAWLER_UA.split(",")
    if kw.strip()
]

# app/middleware/rate_limit.py — dispatch 方法

class RateLimitMiddleware(BaseHTTPMiddleware):

    async def dispatch(self, request: Request, call_next):
        path = request.url.path

        # 先判断是否需要种 cookie（独立于限流逻辑）
        needs_cookie = (
            not request.cookies.get(CLIENT_ID_COOKIE)
            and not path.startswith("/api/")
            and not path.startswith("/static/")
        )

        # 限流逻辑：仅对 API 路径生效
        if RATE_LIMIT_ENABLED and RATE_LIMITS and path.startswith("/api/"):
            if not self._is_crawler(request):
                limit = self._match_limit(path)
                if limit:
                    max_reqs, window = limit
                    client_key = self._get_client_key(request)
                    if self._is_limited(client_key, max_reqs, window):
                        return JSONResponse(
                            status_code=429,
                            content={"detail": "请求过于频繁，请稍后再试"},
                            headers={"Retry-After": str(window)},
                        )

        response = await call_next(request)

        # 首次访问页面时种下 Cookie
        if needs_cookie:
            self._seed_client_cookie(response)

        return response

# app/middleware/rate_limit.py — 客户端标识生成

def _get_client_key(self, request: Request) -> str:
    """优先 Cookie 维度，兜底 IP 维度"""
    cid = request.cookies.get(CLIENT_ID_COOKIE)
    if cid:
        return f"ck:{cid}"

    # IP 取 x-forwarded-for 中的真实客户端地址
    ip = request.client.host if request.client else "0.0.0.0"
    forwarded = request.headers.get("x-forwarded-for")
    if forwarded:
        ip = forwarded.split(",")[0].strip()
    return f"ip:{ip}"

def _seed_client_cookie(self, response: Response):
    """种下 30 天有效期的客户端标识 Cookie"""
    import uuid
    response.set_cookie(
        key=CLIENT_ID_COOKIE,
        value=str(uuid.uuid4()),
        max_age=86400 * 30,
        path="/",
        httponly=True,
        samesite="lax",
    )

# app/middleware/rate_limit.py — 限流状态检查

def _is_limited(self, key: str, max_reqs: int, window: int) -> bool:
    now = time.time()

    # 主路径：Redis 原子操作
    redis = get_redis()
    if redis:
        try:
            count = redis.incr(key)
            if count == 1:
                redis.expire(key, window)
            return count > max_reqs
        except Exception:
            pass  # 降级到内存

    # 降级路径：内存滑动窗口
    timestamps = self._memory.get(key, [])
    timestamps = [t for t in timestamps if now - t < window]
    if len(timestamps) >= max_reqs:
        self._memory[key] = timestamps
        return True
    timestamps.append(now)
    self._memory[key] = timestamps
    return False

# app/middleware/rate_limit.py — 爬虫检测

def _is_crawler(self, request: Request) -> bool:
    ua = (request.headers.get("user-agent") or "").lower()
    return any(kw in ua for kw in CRAWLER_UA_KEYWORDS)

# 关键词列表（从环境变量 RATE_LIMIT_CRAWLER_UA 解析）：
# googlebot, bingbot, duckduckbot, baiduspider,
# yandexbot, sogou, applebot, slurp,
# gptbot, claude-web, anthropic-ai, ccbot,
# oai-searchbot, perplexitybot, facebookexternalhit

<!-- app/templates/base.html — 默认不启用，由 DEVTOOLS_POLICY 控制 -->
{% if devtools_policy == 'deny' or (devtools_policy == 'admin-only' and not is_admin) %}
<script>
(function(){
    function blockKey(e) {
        if (e.key === 'F12' ||
            (e.ctrlKey && e.shiftKey && ['I','J','C'].includes(e.key.toUpperCase())) ||
            (e.ctrlKey && e.key === 'U')) {
            e.preventDefault();
            e.stopPropagation();
            return false;
        }
    }
    document.addEventListener('keydown', blockKey, true);
    document.addEventListener('contextmenu', function(e){ e.preventDefault(); return false; }, true);
})();
</script>
{% endif %}

# app/routes/pages.py — 注册全局模板变量
from app.config import DEVTOOLS_POLICY

templates = Jinja2Templates(directory=str(TEMPLATES_DIR))
templates.env.globals["devtools_policy"] = DEVTOOLS_POLICY

RATE_LIMIT_ENABLED=false
DEVTOOLS_POLICY=allow

RATE_LIMIT_ENABLED=true
RATE_LIMIT_CHAT={从5到15次每分钟，视用户量调整}
RATE_LIMIT_INTERVIEW={同上}
# 其他路径使用默认不限流

RATE_LIMIT_ENABLED=true
RATE_LIMIT_CHAT={适当严格}
RATE_LIMIT_INTERVIEW={适当严格}
RATE_LIMIT_MARKET={中等频率}
RATE_LIMIT_ADMIN={适当宽松}
RATE_LIMIT_API={通用兜底}
DEVTOOLS_POLICY=admin-only