FastAPI 后端开发口味

my_app/
├── __init__.py
├── main.py          # FastAPI app 入口 + 路由注册
├── router.py        # 路由（接口多时可拆成 router_xxx.py）
├── models.py        # SQLModel 数据库模型
├── schemas.py       # Pydantic 入参/出参
├── service.py       # 业务逻辑
├── deps.py          # 依赖注入
└── config.py        # 配置

from contextlib import asynccontextmanager
from fastapi import FastAPI
from .router import router

@asynccontextmanager
async def lifespan(app: FastAPI):
    yield

app = FastAPI(lifespan=lifespan)
app.include_router(router)

src/
├── main.py                   # 入口：创建 app，注册所有 router
├── config.py                 # 全局配置（DATABASE_URL, REDIS_URL 等）
├── database.py               # 数据库连接、Session 工厂
├── models.py                 # 全局基础模型（Base, 时间戳 mixin）
├── exceptions.py             # 全局异常处理器
├── pagination.py             # 通用分页
│
├── auth/                     # ─── 认证领域 ───
│   ├── __init__.py
│   ├── router.py             #   端点定义
│   ├── schemas.py            #   Pydantic 入参/出参
│   ├── models.py             #   数据库模型
│   ├── service.py            #   业务逻辑
│   ├── dependencies.py       #   依赖注入（如 get_current_user）
│   ├── constants.py          #   常量和错误码
│   ├── config.py             #   模块专属环境变量
│   ├── utils.py              #   工具函数（如 hash_password）
│   └── exceptions.py         #   模块专属异常（如 InvalidCredentials）
│
├── posts/                    # ─── 帖子领域（同样的 9 个文件）───
│   └── ...
└── users/                    # ─── 用户领域（同样的 9 个文件）───
    └── ...

router → dependencies → service → models

from src.auth import constants as auth_constants
from src.auth.dependencies import get_current_user

from fastapi import FastAPI
from src.auth.router import router as auth_router
from src.posts.router import router as posts_router

app = FastAPI()
app.include_router(auth_router, prefix="/auth", tags=["Auth"])
app.include_router(posts_router, prefix="/posts", tags=["Posts"])

### PostgreSQL
uv add sqlmodel asyncpg

### SQLite
uv add sqlmodel aiosqlite

from sqlalchemy.ext.asyncio import create_async_engine
from sqlmodel.ext.asyncio.session import AsyncSession

### PostgreSQL
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"
async_engine = create_async_engine(DATABASE_URL, echo=True)

### SQLite
DATABASE_URL = "sqlite+aiosqlite:///./database.db"
async_engine = create_async_engine(
    DATABASE_URL,
    echo=True,
    connect_args={"check_same_thread": False},  # SQLite 必需
)

async def get_session():
    async with AsyncSession(async_engine, expire_on_commit=False) as session:
        yield session

from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 异步建表
    async with async_engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)
    yield
    await async_engine.dispose()

app = FastAPI(lifespan=lifespan)

### 使用依赖注入
@app.get("/users/{user_id}")
async def get_user(user_id: int, session: AsyncSession = Depends(get_session)):
    user = await session.get(User, user_id)
    return user

from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import sessionmaker
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlmodel import create_engine, Session

### 异步引擎 - 主业务数据库
engine = create_async_engine("sqlite+aiosqlite:///volume/dbs/chat.db", echo=False)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

### 同步引擎 - 向量库等不支持异步的场景
engine_faq = create_engine("sqlite:///volume/dbs/faq.db")

### 分别注入
async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session(expire_on_commit=False) as session:
        yield session

def get_session_faq() -> Session:
    with Session(engine_faq) as session:
        yield session

### 异步建表 - 只建主业务表
async with engine.begin() as conn:
    await conn.run_sync(
        SQLModel.metadata.create_all,
        tables=[SessionTable.__table__, MessageTable.__table__]
    )

### 同步建表 - 只建向量库表
SQLModel.metadata.create_all(engine_faq, tables=[FAQ_CREATE.__table__])

uv add sqlmodel sqlite-vec-sqlalchemy

from sqlmodel import create_engine, Session
from sqlite_vec_sqlalchemy import enable_sqlite_vec

### FAQ 向量库 - 同步引擎
engine_faq = create_engine("sqlite:///volume/dbs/faq.db")
enable_sqlite_vec(engine_faq)  # 必须调用，加载 sqlite-vec 扩展

### 依赖注入
def get_session_faq() -> Session:
    with Session(engine_faq) as session:
        yield session

from sqlmodel import SQLModel, Field, Column, JSON
from sqlite_vec_sqlalchemy import Vector

length = 3072  # embedding 维度，根据模型调整

### 1. Base：共享业务字段
class BaseFAQ(SQLModel):
    id: int = Field(primary_key=True)
    question: str
    answer: str
    image_urls: list[str] = Field(default=[], sa_column=Column(JSON))

### 2. DB Table：带向量列，用于建表和写入
class FAQ_CREATE(BaseFAQ, table=True):
    __tablename__ = 'faq'
    embedding1: list[float] = Field(sa_column=Column(Vector(length)))  # 问题向量
    embedding2: list[float] = Field(sa_column=Column(Vector(length)))  # 答案向量

### 3. 查询结果：带距离字段，不建表
class FAQ_RES(BaseFAQ):
    distance: float | None = None
    @classmethod
    def model_validate2(cls, data: tuple):
        faq, distance = data
        a = cls.model_validate(faq)
        a.distance = distance
        return a

### 4. 未命中记录表（可选）
class FAQ_NO_ANSWER(SQLModel, table=True):
    __tablename__ = 'faq_no_answer'
    id: int = Field(primary_key=True)
    question: str
    created_at: datetime = Field(default_factory=datetime.now)
    created_date: date = Field(default_factory=lambda: datetime.now().date(), index=True)

from sqlmodel import SQLModel
SQLModel.metadata.create_all(engine_faq, tables=[FAQ_CREATE.__table__, FAQ_NO_ANSWER.__table__])

from sqlite_vec_sqlalchemy import vec_distance_cosine
from sqlmodel import select, Session

def search_faq(question_embedding: list[float], top_k: int = 8) -> list[FAQ_RES]:
    with Session(engine_faq, expire_on_commit=False) as session:
        # 搜问题向量
        dist1 = vec_distance_cosine(FAQ_CREATE.embedding1, question_embedding).label("distance")
        stmt1 = select(FAQ_CREATE, dist1).order_by(dist1).limit(top_k)
        res1 = session.exec(stmt1).all()

        # 搜答案向量
        dist2 = vec_distance_cosine(FAQ_CREATE.embedding2, question_embedding).label("distance")
        stmt2 = select(FAQ_CREATE, dist2).order_by(dist2).limit(top_k)
        res2 = session.exec(stmt2).all()

    # 合并去重
    seen = set()
    results = []
    for item in [FAQ_RES.model_validate2(i) for i in res1 + res2]:
        if item.id not in seen:
            seen.add(item.id)
            results.append(item)
    return results

from langchain_openai import OpenAIEmbeddings

model_embedding = OpenAIEmbeddings(model='text-embedding-3-large')

async def create_faq(q: str, a: str, image_urls: list[str] = []) -> FAQ_CREATE:
    faq = FAQ_CREATE(question=q, answer=a, image_urls=image_urls)
    embeddings = await model_embedding.aembed_documents([faq.question, faq.answer])
    faq.embedding1 = embeddings[0]
    faq.embedding2 = embeddings[1]
    return faq

from langchain_core.runnables import RunnableLambda
from tqdm import tqdm

### 用 RunnableLambda 包装，便于 abatch 并行调用
@RunnableLambda
async def add_faq(foo):
    faq = FAQ_CREATE(question=foo['q'], answer=foo['a'], image_urls=foo.get('image_urls', []))
    embeddings = await model_embedding.aembed_documents([faq.question, faq.answer])
    faq.embedding1 = embeddings[0]
    faq.embedding2 = embeddings[1]
    return faq


async def sync_faq(session: Session, new_data: list[dict]):
    """
    new_data: [{'q': str, 'a': str, 'image_urls': list[str]}, ...]
    """
    import pandas as pd

    # 1. 读取现有数据
    df_old = pd.read_sql('faq', engine_faq)

    # 2. 构建新旧映射 (q, a) -> image_urls
    now_values = {(d['q'], d['a']): d.get('image_urls', []) for d in new_data}
    old_values = {(row['question'], row['answer']): row.get('image_urls', []) 
                  for _, row in df_old.iterrows()}

    # 3. 删除：旧有新无
    delete_ids = [
        row['id'] for _, row in df_old.iterrows() 
        if (row['question'], row['answer']) not in now_values
    ]
    if delete_ids:
        logger.debug(f'删除 {len(delete_ids)} 条FAQ')
        result = session.exec(select(FAQ_CREATE).where(FAQ_CREATE.id.in_(delete_ids)))
        for faq in result.all():
            session.delete(faq)
        session.commit()

    # 4. 新增：新有旧无
    add_list = [d for d in new_data if (d['q'], d['a']) not in old_values]

    # 5. 批量写入（每10条一批，abatch 并行生成 embedding）
    logger.debug(f'添加 {len(add_list)} 条FAQ')
    for i in tqdm(range(0, len(add_list), 10)):
        batch = add_list[i:i+10]
        faqs = await add_faq.abatch(batch)
        session.add_all(faqs)
        session.commit()

    # 6. 更新：键相同但附属字段变化（无需重新 embedding）
    update_list = [
        {'q': q, 'a': a, 'image_urls': urls}
        for (q, a), urls in now_values.items()
        if (q, a) in old_values and sorted(urls) != sorted(old_values[(q, a)])
    ]
    if update_list:
        logger.debug(f'更新 {len(update_list)} 条FAQ的附属字段')
        from sqlmodel import and_
        for item in tqdm(update_list):
            result = session.exec(
                select(FAQ_CREATE).where(
                    and_(FAQ_CREATE.question == item['q'], FAQ_CREATE.answer == item['a'])
                )
            )
            faq = result.first()
            if faq:
                faq.image_urls = item['image_urls']
                session.add(faq)
        session.commit()

@router.put("/sync_faq")
async def sync_faq_endpoint(session: Annotated[Session, Depends(get_session_faq)]):
    try:
        new_data = load_from_external_source()  # 从外部数据源读取
        await sync_faq(session, new_data)
        return {'status': 'ok'}
    except Exception as e:
        logger.exception(e)
        # 可选：发送异常通知（钉钉/飞书/Slack 等）
        raise

from jinja2 import Template

### 单条模式：只返回最贴近的一条
rerank_system_single = r"""
你是一个专业助手，正在筛选能解答用户问题的FAQ。

请使用json输出FAQ的id，不要输出其他信息，格式为{"faq_id": int}，
如果有多条能解答用户提问的FAQ也只输出最贴近的一条的id，
如果没有能回答提问的FAQ，请输出{"faq_id": -1}。

筛选必须严格，禁止答非所问，禁止货不对板（例如用户问a产品，但是回答的是b产品）
"""

### 多条模式：返回所有能回答的
rerank_system_multi = r"""
你是一个专业助手，正在筛选能解答用户问题的FAQ。

请使用json输出FAQ的id，不要输出其他信息，格式为{"faq_id": list[int]}，
如果有多条能解答用户提问的FAQ就要全部输出，
如果没有能回答提问的FAQ，请输出{"faq_id": []}。

筛选必须严格，禁止答非所问，禁止货不对板（例如用户问a产品，但是回答的是b产品）
"""

### 粗召结果模板
rerank_human_template = Template(r"""
粗召到的FAQ有:
{% for faq in faqs %}
faq_id: {{ faq.id }}:
问题: `{{ faq.question }}`
回答: `{{ faq.answer }}`
---
{% endfor %}
""")

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, BaseMessage

async def query_faq(question: str, history: list[BaseMessage], model, count: int = 1) -> list[FAQ_RES]:
    """
    FAQ 匹配主函数

    Args:
        question: 用户问题
        history: 对话历史（用于给 LLM 更多上下文）
        model: 用于精排的 LLM
        count: 最多返回几条（1=单条模式，>1=多条模式）

    Returns:
        匹配到的 FAQ 列表，空列表表示未命中
    """
    # ── 1. 双路粗召 ──
    emb = model_embedding.embed_query(question)
    candidates = search_faq(emb, top_k=8)  # 复用前面定义的 search_faq

    logger.debug(f'粗召到 {len(candidates)} 条FAQ')

    # ── 2. LLM 精排 ──
    rendered = rerank_human_template.render(faqs=[i.model_dump() for i in candidates])

    # 过滤对话历史：只保留 Human/AI 文本消息，排除 tool_calls
    filtered_history = [
        m for m in history 
        if isinstance(m, (HumanMessage, AIMessage))
        and not (isinstance(m, AIMessage) and m.tool_calls)
    ]

    system_prompt = rerank_system_single if count == 1 else rerank_system_multi
    messages = [
        SystemMessage(content=system_prompt),
        *filtered_history,
        HumanMessage(content=rendered),
    ]

    result = await model.with_structured_output(method='json_mode').ainvoke(messages)
    faq_id = result['faq_id']

    # 统一为列表
    faq_id_list = [faq_id] if isinstance(faq_id, int) else faq_id

    # ── 3. 过滤输出 ──
    matched = [f for f in candidates if f.id in faq_id_list][:count]

    if not matched:
        # 记录未命中，便于后续分析 FAQ 覆盖率
        with Session(engine_faq, expire_on_commit=False) as session:
            session.add(FAQ_NO_ANSWER(question=question))
            session.commit()

    return matched

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from operator import itemgetter

prompt_extract_question = ChatPromptTemplate.from_messages([
    ("system", "用一句话总结用户现在的问题。不要思考太多, 思考控制在50字以内。请使用json输出，格式为{\"question\": str}"),
    MessagesPlaceholder(variable_name='history'),
    ("human", "请提取出用户的问题，输出json"),
], template_format='jinja2')

chain_extract_question = (
    prompt_extract_question 
    | model_small.with_structured_output(method='json_mode')
    | itemgetter('question')
)

### 使用：question = await chain_extract_question.ainvoke({'history': messages})

uv add sqlmodel asyncpg pgvector sqlalchemy-pgvector

"""FastMCP + FastAPI 混合服务（JWT 认证版）

使用 Ed25519 JWT Token 进行用户身份验证。

环境变量：
    JWT_ED25519_PUBLIC_KEY: Base64 编码的 Ed25519 公钥

使用方法：
    export JWT_ED25519_PUBLIC_KEY="你的Base64公钥"
    uv run main.py
"""
import os
import time

from fastapi import FastAPI, Depends, HTTPException, Request, Response
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from fastmcp import FastMCP
from fastmcp.server.auth.auth import TokenVerifier
from fastmcp.server.middleware import Middleware, MiddlewareContext
from fastmcp.server.dependencies import get_access_token
from fastmcp.dependencies import CurrentAccessToken
from fastmcp.server.auth import AccessToken
from loguru import logger

from jwt_verifier import (
    JWTTokenVerifier,
    TokenExpiredError,
    TokenFormatError,
    TokenInvalidError,
    VerifiedUser,
)

### ============================================================
### 配置
### ============================================================

PUBLIC_KEY = os.environ.get("JWT_ED25519_PUBLIC_KEY", "")

if not PUBLIC_KEY:
    raise RuntimeError("请设置 JWT_ED25519_PUBLIC_KEY 环境变量")

verifier = JWTTokenVerifier(PUBLIC_KEY)


### ============================================================
### 自定义 JWT 验证器（用于 FastMCP 认证）
### ============================================================
class JWTMCPVerifier(TokenVerifier):
    """
    JWT 验证器，适配 FastMCP 的 TokenVerifier 协议。

    接收 Bearer Token（JWT），验证签名后返回 AccessToken。
    """

    def __init__(self, jwt_verifier: JWTTokenVerifier, required_scopes: list[str] | None = None):
        super().__init__(required_scopes=required_scopes or ["tools"])
        self.jwt_verifier = jwt_verifier

    async def verify_token(self, token: str) -> AccessToken | None:
        """
        验证 JWT Token，返回 AccessToken 对象。

        Args:
            token: JWT Token 字符串

        Returns:
            AccessToken 对象（验证成功）或 None（验证失败）
        """
        try:
            user = self.jwt_verifier.verify_token(token)
            logger.info(f"✅ JWT 验证成功！用户: {user.name} ({user.email})")
            return AccessToken(
                token=token,
                client_id=user.email or user.user_id,
                scopes=self.required_scopes,
                expires_at=None,
                claims=user.claims,
            )
        except TokenExpiredError:
            logger.warning("❌ JWT Token 已过期")
            return None
        except TokenInvalidError:
            logger.warning("❌ JWT Token 签名无效")
            return None
        except TokenFormatError as e:
            logger.warning(f"❌ JWT Token 格式错误: {e}")
            return None


### ============================================================
### MCP 日志中间件（记录谁调用了什么工具、参数、耗时）
### ============================================================
class AuditLoggingMiddleware(Middleware):
    """
    MCP 审计日志中间件。

    拦截所有 tool 调用，记录：
      - 调用者身份（client_id / email）
      - 工具名称
      - 调用参数
      - 执行耗时
      - 执行结果（成功/失败）
    """

    async def on_call_tool(self, context: MiddlewareContext, call_next):
        tool_name = context.message.name
        arguments = context.message.arguments

        # 尝试获取调用者身份
        caller = "anonymous"
        try:
            token = get_access_token()
            if token:
                caller = token.client_id or "unknown"
        except Exception:
            pass

        logger.debug(
            "[MCP Audit] 🔧 tool={tool} | caller={caller} | args={args}",
            tool=tool_name,
            caller=caller,
            args=arguments,
        )

        start = time.perf_counter()
        try:
            result = await call_next(context)
            elapsed = (time.perf_counter() - start) * 1000
            logger.debug(
                "[MCP Audit] ✅ tool={tool} | caller={caller} | elapsed={elapsed:.1f}ms",
                tool=tool_name,
                caller=caller,
                elapsed=elapsed,
            )
            return result
        except Exception as exc:
            elapsed = (time.perf_counter() - start) * 1000
            logger.debug(
                "[MCP Audit] ❌ tool={tool} | caller={caller} | elapsed={elapsed:.1f}ms | error={error}",
                tool=tool_name,
                caller=caller,
                elapsed=elapsed,
                error=str(exc),
            )
            raise


### ============================================================
### 核心业务逻辑（复用）
### ============================================================
def core_add_logic(a: int, b: int) -> int:
    return a + b


### ============================================================
### 1. 初始化 FastMCP（带 JWT 认证）
### ============================================================
mcp_auth = JWTMCPVerifier(
    jwt_verifier=verifier,
    required_scopes=["tools"],
)

mcp = FastMCP("Demo 🚀", auth=mcp_auth)

### 注册 MCP 日志中间件
mcp.add_middleware(AuditLoggingMiddleware())


### ============================================================
### 2. 注册 MCP 工具（通过 CurrentAccessToken 获取用户信息）
### ============================================================
@mcp.tool()
async def add(
    a: int,
    b: int,
    token: AccessToken = CurrentAccessToken(),
) -> str:
    """Add two numbers (Available for LLM)

    两数相加，返回结果和调用者信息。
    """
    user_email = token.client_id
    result = core_add_logic(a, b)
    return f"{a} + {b} = {result} (called by {user_email})"


@mcp.tool()
async def whoami(
    token: AccessToken = CurrentAccessToken(),
) -> dict:
    """查看当前认证用户的信息"""
    logger.info(f"token: {token}")
    return {
        "email": token.client_id,
        "scopes": token.scopes,
        "claims": token.claims,
    }


### ============================================================
### 3. 初始化 FastAPI + 挂载 MCP
### ============================================================
mcp_app = mcp.http_app(path="/")

app = FastAPI(
    title="Hybrid API & MCP Server (JWT Auth)",
    lifespan=mcp_app.lifespan,
)

### 挂载 MCP 到 /mcp 路径
app.mount("/mcp", mcp_app)


### ============================================================
### FastAPI 日志中间件（记录 REST 接口调用）
### ============================================================
@app.middleware("http")
async def fastapi_audit_logging(request: Request, call_next):
    """
    FastAPI 审计日志中间件。

    记录：
      - 请求方法 + 路径 + 查询参数
      - 调用者身份（从 Authorization header 解析 JWT）
      - 响应状态码
      - 执行耗时
    """
    caller = "anonymous"
    auth_header = request.headers.get("authorization", "")
    if auth_header.startswith("Bearer "):
        token_str = auth_header[7:]
        try:
            user = verifier.verify_token(token_str)
            caller = user.email or user.user_id
        except Exception:
            caller = "invalid-token"

    method = request.method
    path = request.url.path
    query = str(request.query_params) if request.query_params else ""

    logger.debug(
        "[REST Audit] → {method} {path} | caller={caller} | query={query}",
        method=method,
        path=path,
        caller=caller,
        query=query,
    )

    start = time.perf_counter()
    response: Response = await call_next(request)
    elapsed = (time.perf_counter() - start) * 1000

    logger.debug(
        "[REST Audit] ← {method} {path} | caller={caller} | status={status} | elapsed={elapsed:.1f}ms",
        method=method,
        path=path,
        caller=caller,
        status=response.status_code,
        elapsed=elapsed,
    )
    return response


### ============================================================
### 4. FastAPI REST 端点认证（使用 JWT 验证）
### ============================================================
bearer_scheme = HTTPBearer(
    scheme_name="JWT Bearer Token",
    description="输入 JWT Token（不需要加 Bearer 前缀）",
)


async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme),
) -> VerifiedUser:
    """从 Authorization: Bearer <token> 提取并验证用户身份"""
    token = credentials.credentials

    try:
        return verifier.verify_token(token)
    except TokenExpiredError:
        raise HTTPException(status_code=401, detail="Token expired")
    except TokenInvalidError:
        raise HTTPException(status_code=401, detail="Invalid token")
    except TokenFormatError as e:
        raise HTTPException(status_code=400, detail=f"Malformed token: {e}")


### ============================================================
### 5. 注册 FastAPI REST 接口（通过 Depends 获取用户信息）
### ============================================================
@app.get("/api/add")
async def add_api(a: int, b: int, user: VerifiedUser = Depends(get_current_user)):
    """Standard REST API (Available for Web/Mobile)"""
    result = core_add_logic(a, b)
    return {
        "result": result,
        "user": {
            "user_id": user.user_id,
            "email": user.email,
            "name": user.name,
        },
        "source": "fastapi",
    }


@app.get("/api/whoami")
async def whoami_api(user: VerifiedUser = Depends(get_current_user)):
    """查看当前认证用户的信息（REST 版）"""
    return {
        "user": {
            "user_id": user.user_id,
            "email": user.email,
            "name": user.name,
            "claims": user.claims,
        },
        "source": "fastapi",
    }


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8000)

### jwt_verifier.py

"""JWT Ed25519 Token 验证器

使用 Ed25519 公钥验证 JWT Token，提取用户信息。

依赖：
    uv add PyJWT cryptography

环境变量：
    JWT_ED25519_PUBLIC_KEY: Base64 编码的 Ed25519 公钥
"""

import base64
from dataclasses import dataclass, field
from typing import Any

import jwt
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
from cryptography.hazmat.primitives.serialization import load_pem_public_key
import os 

### ============================================================
### 异常定义
### ============================================================

class TokenExpiredError(Exception):
    """Token 已过期"""
    pass


class TokenInvalidError(Exception):
    """Token 签名无效或不可信"""
    pass


class TokenFormatError(Exception):
    """Token 格式错误，无法解析"""
    pass


### ============================================================
### 验证后的用户信息
### ============================================================

@dataclass
class VerifiedUser:
    """JWT 验证成功后返回的用户信息"""
    user_id: str
    email: str
    name: str
    claims: dict[str, Any] = field(default_factory=dict)


### ============================================================
### 辅助函数
### ============================================================

def _load_from_pem(pem_bytes: bytes) -> Ed25519PublicKey:
    """从 PEM 字节加载 Ed25519 公钥"""
    key = load_pem_public_key(pem_bytes)
    if not isinstance(key, Ed25519PublicKey):
        raise ValueError(f"PEM 中包含的不是 Ed25519 公钥，而是 {type(key).__name__}")
    return key


### ============================================================
### JWT Token 验证器
### ============================================================

class JWTTokenVerifier:
    """使用 Ed25519 公钥验证 JWT Token

    支持三种公钥格式：
    1. Base64(PEM)  — PEM 字符串的 Base64 编码（常用于环境变量）
    2. 原始 PEM     — -----BEGIN PUBLIC KEY----- 开头的字符串
    3. Base64(raw)  — 32 字节原始公钥的 Base64 编码

    Usage:
        verifier = JWTTokenVerifier("Base64EncodedPublicKey...")
        user = verifier.verify_token("eyJhbGciOi...")
    """

    def __init__(self, public_key_b64: str):
        self._public_key = self._load_public_key(public_key_b64.strip())

    @staticmethod
    def _load_public_key(key_input: str) -> Ed25519PublicKey:
        """智能加载 Ed25519 公钥，自动检测格式"""

        # 情况 1：直接就是 PEM 明文
        if key_input.startswith("-----BEGIN"):
            return _load_from_pem(key_input.encode())

        # Base64 解码
        try:
            key_bytes = base64.b64decode(key_input)
        except Exception as e:
            raise ValueError(f"无法 Base64 解码公钥: {e}")

        # 情况 2：解码后是 PEM 字符串（Base64 包裹的 PEM）
        try:
            pem_text = key_bytes.decode("utf-8")
            if pem_text.strip().startswith("-----BEGIN"):
                return _load_from_pem(pem_text.encode())
        except UnicodeDecodeError:
            pass  # 不是文本，继续尝试原始字节

        # 情况 3：解码后正好是 32 字节原始公钥
        if len(key_bytes) == 32:
            try:
                return Ed25519PublicKey.from_public_bytes(key_bytes)
            except Exception as e:
                raise ValueError(f"无法从原始字节加载公钥: {e}")

        raise ValueError(
            f"无法识别公钥格式（解码后 {len(key_bytes)} 字节）。"
            "支持: PEM 明文 / Base64(PEM) / Base64(32字节原始公钥)"
        )

    def verify_token(self, token: str) -> VerifiedUser:
        """验证 JWT Token 并返回用户信息

        Args:
            token: JWT Token 字符串

        Returns:
            VerifiedUser 对象

        Raises:
            TokenExpiredError: Token 已过期
            TokenInvalidError: Token 签名无效
            TokenFormatError: Token 格式错误
        """
        try:
            payload = jwt.decode(
                token,
                self._public_key,
                algorithms=["EdDSA"],
            )
        except jwt.ExpiredSignatureError:
            if not os.getenv("DEBUG"):
                raise TokenExpiredError("Token 已过期")
            # DEBUG 模式下允许过期 Token，重新解码但跳过过期验证
            payload = jwt.decode(
                token,
                self._public_key,
                algorithms=["EdDSA"],
                options={"verify_exp": False},
            )
        except jwt.InvalidTokenError as e:
            raise TokenInvalidError(f"Token 无效: {e}")
        except Exception as e:
            raise TokenFormatError(f"Token 格式错误: {e}")

        return VerifiedUser(
            user_id=payload.get("sub", ""),
            email=payload.get("email", ""),
            name=payload.get("name", ""),
            claims=payload,
        )

from loguru import logger
import signal
import os
import time

class GracefulRunner:
    def __init__(self):
        self._keep_running = True
        self.original_sigint_handler = None
        self.original_sigterm_handler = None

    def __bool__(self):
        """支持 `if not runner:` 这种优雅写法"""
        return self._keep_running

    def setup_signal_handlers(self):
        self.original_sigint_handler = signal.signal(signal.SIGINT, self._handle_signal)
        self.original_sigterm_handler = signal.signal(signal.SIGTERM, self._handle_signal)

    def _handle_signal(self, signum, frame):
        if not self._keep_running:
            return
        logger.info(f"收到信号 {signum}，标记停止运行...")
        self._keep_running = False

        import random
        time.sleep(random.random())  # 避免多worker同时抢占文件描述符

        # 转发信号给 uvicorn，让它正常走关闭流程
        if signum == signal.SIGINT and self.original_sigint_handler:
            if callable(self.original_sigint_handler):
                self.original_sigint_handler(signum, frame)
            else:
                signal.signal(signal.SIGINT, signal.SIG_DFL)
                os.kill(os.getpid(), signal.SIGINT)
        elif signum == signal.SIGTERM and self.original_sigterm_handler:
            if callable(self.original_sigterm_handler):
                self.original_sigterm_handler(signum, frame)
            else:
                signal.signal(signal.SIGTERM, signal.SIG_DFL)
                os.kill(os.getpid(), signal.SIGTERM)

runner = GracefulRunner()

@asynccontextmanager
async def lifespan(app: FastAPI):
    runner.setup_signal_handlers()
    yield
    logger.success('服务已优雅关闭')

from .utils.runner import runner

### 简单用法：在循环中检查
async for chunk in stream:
    if not runner:
        logger.warning("收到终止信号，正在优雅退出...")
        break
    yield chunk

from dataclasses import dataclass, field
from enum import Enum, auto

class StopReason(Enum):
    BACKEND_SHUTDOWN = auto()  # 后端终止信号
    NEW_MESSAGE = auto()       # 用户发送新消息

@dataclass
class StopSignal:
    reason: StopReason
    data: str  # 要返回给客户端的数据

@dataclass
class StreamController:
    session: AsyncSession
    session_id: str
    lock: Any
    check_interval: float = 5.0

    _last_check_time: float = field(default=0.0, init=False)

    async def check_should_stop(self) -> StopSignal | None:
        # 1. 后端终止 — 立即检查
        if not runner:
            return StopSignal(reason=StopReason.BACKEND_SHUTDOWN, data=...)

        # 2. 其他条件 — 按间隔检查（避免频繁查库）
        now = time.time()
        if now - self._last_check_time < self.check_interval:
            return None
        self._last_check_time = now

        # 例：检查是否有新消息覆盖了当前回复
        await self.session.refresh(self._ticket)
        if self._ticket.lock != self.lock:
            return StopSignal(reason=StopReason.NEW_MESSAGE, data=...)

        return None

controller = StreamController(session, session_id, lock)
async for chunk in stream:
    if stop := await controller.check_should_stop():
        yield stop.data
        await stream.aclose()
        return
    yield process(chunk)

class SessionTable(SQLModel, table=True):
    session_id: str = Field(primary_key=True)
    lock: UUID | None = Field(default=None)   # 乐观锁：每条新消息刷新
    # ...

@router.post("/send")
async def send_message(request: ReceivedMessage, session: ...):
    lock = uuid4()  # 生成新锁

    # 获取 session，更新锁
    ticket = await session.get(SessionTable, session_id)
    ticket.lock = lock
    session.add(ticket)

    # 同时写入新消息
    db_message = MessageTable.model_validate(request)
    session.add(db_message)

    # 一次 commit，保证锁和消息同时生效
    await session.commit()

    # 后续用这个 lock 创建 StreamController
    controller = StreamController(session, session_id, lock)

### 在 StreamController.check_should_stop() 中：
await self.session.refresh(self._ticket)
if self._ticket.lock != self.lock:
    # lock 变了 → 有新消息进来了 → 中断当前生成
    return StopSignal(reason=StopReason.NEW_MESSAGE, data=...)

### 查询该 session 下所有有效消息
stmt = (
    select(MessageTable)
    .where(
        MessageTable.session_id == session_id,
        MessageTable.send_status.in_([
            MessageStatus.SUCCESS,
            MessageStatus.PROCESSING,
            MessageStatus.INITIALIZING
        ]),
    )
    .order_by(MessageTable.created_at)
)
db_messages = (await session.exec(stmt)).all()

### 从最后一个 checkpoint 开始，收集所有新消息作为输入
new_messages = get_new_messages(db_messages)

if db_messages[-1].message_id != db_message.message_id:
    logger.warning('新消息比老消息更先写入数据库')
    # 不影响正确性，后排部队转作先头部队

用户发消息A → 写入消息A + lock=uuid_a → commit → 开始生成回复A
用户发消息B → 写入消息B + lock=uuid_b → commit → 开始生成回复B
                                                    ↓
回复A生成中 → check_should_stop() → 发现 lock≠uuid_a → 中断回复A
回复B生成中 → 读取消息A+B作为完整上下文 → 生成基于A+B的回复

### 基础 + FastAPI面板 + cron支持
pip install fastscheduler[all]

from fastapi import FastAPI
from fastscheduler import FastScheduler
from fastscheduler.fastapi_integration import create_scheduler_routes

app = FastAPI()
scheduler = FastScheduler(quiet=True)

### 挂载控制面板到 /scheduler/
app.include_router(create_scheduler_routes(scheduler))

@scheduler.every(30).seconds
async def background_task():
    print("Background work")

@scheduler.daily.at("02:00")
async def nightly_cleanup():
    await clean_old_sessions(days=10)

scheduler.start()

### 间隔调度
@scheduler.every(10).seconds
@scheduler.every(5).minutes
@scheduler.every(2).hours

### 定时调度
@scheduler.daily.at("09:00")
@scheduler.hourly.at(":30")
@scheduler.weekly.monday.at("10:00")
@scheduler.weekly.weekdays.at("09:00")

### Cron 表达式
@scheduler.cron("0 9 * * MON-FRI")  # 工作日9点
@scheduler.cron("*/15 * * * *")     # 每15分钟

### 一次性任务
@scheduler.once(60)                  # 60秒后执行一次
@scheduler.at("2026-12-25 00:00:00") # 指定时间执行

### 时区支持
@scheduler.daily.at("09:00", tz="Asia/Shanghai")

### 超时 & 重试（指数退避）
@scheduler.every(5).minutes.timeout(30).retries(3)
def flaky_api_call():
    ...

### 暂停 / 恢复 / 取消
scheduler.pause_job("job_0")
scheduler.resume_job("job_0")
scheduler.cancel_job("job_0")

scheduler = FastScheduler(
    storage="sqlmodel",                    # 持久化到数据库（默认 json 文件）
    database_url="sqlite:///scheduler.db", # 或 postgresql://...
    max_workers=20,                        # 并发任务数
    max_history=5000,                      # 最大历史记录数
    history_retention_days=8,              # 历史保留天数
    max_dead_letters=500,                  # 最大失败记录数
)

外部数据源(Google Sheets等)
    ↓ update_table()
SQLite (持久化)  ←→  内存 DataFrame (高速读取)
    ↑                    ↑
  重启恢复            文件锁触发热更新

import pandas as pd
import json
import time
from pathlib import Path
from functools import wraps
from sqlmodel import create_engine

engine_config = create_engine('sqlite:///volume/dbs/config.db')

class ConfigTable:

    def __init__(self):
        self.path = Path('volume/dbs/update.lock')
        if not self.path.exists():
            self.path.touch()
        self.update_time = self.path.stat().st_mtime
        self.master_table = self._load_master_table()
        # 启动时一次性加载所有配置表到内存
        self.tables = {
            tn: pd.read_sql(tn, engine_config, index_col=index_col)
            if index_col else pd.read_sql(tn, engine_config)
            for tn, index_col in self.master_table['index_col'].items()
        }

    @property
    def config(self):
        """通过 tables.config['表名', '列名'] 访问"""
        return self.master_table.loc

    @staticmethod
    def _load_master_table():
        mt = pd.read_sql_table('table_master', engine_config, index_col='table_name')
        mt['index_col'] = mt['index_col'].apply(json.loads)
        mt['bool_col'] = mt['bool_col'].apply(json.loads)
        return mt

    @staticmethod
    def ensure_new(func):
        """装饰器：访问前检查文件锁时间戳，有变化才重新加载"""
        @wraps(func)
        def wrapper(self, *args, **kwargs):
            if self.path.stat().st_mtime != self.update_time:
                new_master = self._load_master_table()
                # 只重新加载发生变化的表（对比 update_time 列）
                old_time = self.master_table['update_time']
                new_time = new_master['update_time']
                changed = new_time[new_time != old_time.reindex(new_time.index)].index
                for tn in changed:
                    self.tables[tn] = pd.read_sql(
                        tn, engine_config,
                        index_col=new_master.loc[tn, 'index_col']
                    )
                self.master_table = new_master
                self.update_time = self.path.stat().st_mtime
            return func(self, *args, **kwargs)
        return wrapper

    @ensure_new
    def __getitem__(self, table_name: str) -> pd.DataFrame:
        if table_name not in self.tables:
            raise KeyError(f'Table {table_name} not found')
        return self.tables[table_name]

tables = ConfigTable()

async def update_table(self, table_name: str):
    # 1. 从外部源拉取数据
    df = pd.read_csv(f'{source_url}/csv/{table_name}', ...)

    # 2. 写入 SQLite（持久化）
    df.to_sql(table_name, engine_config, index=True, if_exists='replace')

    # 3. 更新 master_table 的时间戳
    self.master_table.loc[table_name, 'update_time'] = int(time.time() * 1000)
    self.master_table.to_sql('master_table', engine_config, ...)

    # 4. 更新内存中的表
    self.tables[table_name] = df

    # 5. 触摸文件锁，通知其他进程
    self.path.touch()
    self.update_time = self.path.stat().st_mtime

@register_dataframe_accessor("safe_at")
class SafeLocAccessor:
    # 回退规则：索引找不到时，尝试这些默认值
    fallback_rules = {
        'game_cd': 0,      # 找不到特定游戏配置 → 用默认配置
        'lang': 'en',      # 找不到特定语言 → 用英语
    }

    def __getitem__(self, key: tuple):
        k0, k1 = key  # (索引值, 列名)
        if k0 in self._df.index:
            return self._df.at[key]
        # 多级索引：用笛卡尔积遍历所有回退组合
        ...

### 直接取值（可能 KeyError）
value = tables['prompt_config'].at['game_a', 'system_prompt']

### 安全取值（自动回退）
value = tables['prompt_config'].safe_at['game_a', 'system_prompt']
### 如果 game_a 不存在，自动尝试 game_cd=0 的默认配置

### 全局缓存
game_expire_dict = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时从API加载
    response = await client.get(f'{api_url}/config')
    game_expire_dict.update(response.json())
    yield

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse, HTMLResponse
from fastapi.staticfiles import StaticFiles

app = FastAPI()

### 挂载静态文件（HTML/CSS/JS）
app.mount("/static", StaticFiles(directory="static"), name="static")

@app.get("/", response_class=HTMLResponse)
async def index():
    with open("static/index.html") as f:
        return f.read()

@app.post("/chat")
async def chat(request: Request):
    body = await request.json()
    message = body["message"]

    async def generate():
        # 替换成你的 LLM 调用（LangGraph / OpenAI / Anthropic 等）
        async for chunk in your_llm_stream(message):
            yield f"data: {chunk}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

// 流式读取 SSE 响应
const response = await fetch("/chat", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ message: userInput })
});

const reader = response.body.getReader();
const decoder = new TextDecoder();

while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    const text = decoder.decode(value);
    // 解析 SSE data: 行，追加到聊天气泡里
    appendToChat(text);
}

from fastapi import FastAPI
from chainlit.utils import mount_chainlit

app = FastAPI()

@app.get("/api/health")
def health():
    return {"status": "ok"}

mount_chainlit(app=app, target="my_cl_app.py", path="/chainlit")

import chainlit as cl

@cl.on_chat_start
async def on_chat_start():
    graph = ...  # 编译好的 LangGraph
    cl.user_session.set("graph", graph)

@cl.on_message
async def on_message(message: cl.Message):
    graph = cl.user_session.get("graph")
    msg = cl.Message(content="")

    async for chunk in graph.astream(
        {"messages": [{"role": "user", "content": message.content}]},
        stream_mode="messages"
    ):
        await msg.stream_token(chunk.content)

    await msg.send()

from dowhen import when

### 在 SomeClass.some_method 中，找到 'target_line' 这行代码
### 当 condition 为真时，用 .do() 中的代码替换/插入
(
    when(
        SomeClass.some_method,
        'target_line_of_code',        # 定位到方法中的哪一行
        condition='some_condition'     # 可选：仅在条件成立时执行
    )
    .do("replacement_code")
)

from langfuse.langchain.CallbackHandler import LangchainCallbackHandler
from dowhen import when

(
    when(
        LangchainCallbackHandler.on_chain_error,
        'observation = self._detach_observation(run_id)',
        condition=(
            'isinstance(error, GeneratorExit)'
            'or "PreconsicousError" in str(type(error))'
            'or "CancelledError" in str(type(error))'
        )
    )
    .do("level = None")  # 将 level 设为 None，跳过错误记录
)

do2 = '''
if observation is not None:
    observation.update(
        status_message=str(error),
        level="WARNING",
        input=kwargs.get("inputs"),
        cost_details={"total": 0},
    ).end()
observation = None
'''

(
    when(
        LangchainCallbackHandler.on_tool_error,
        'if observation is not None:',
        condition='(error.__class__.__name__ in ["ChangeToolError"])'
    )
    .do(do2)
)

from langchain_deepseek import ChatDeepSeek

do4 = '''
messages = self._convert_input(input_).to_messages()
reasoning_content_map = {}
for i, msg in enumerate(messages):
    if hasattr(msg, "additional_kwargs"):
        rc = msg.additional_kwargs.get("reasoning_content")
        if rc is not None:
            reasoning_content_map[i] = rc

for i, message in enumerate(payload["messages"]):
    if message["role"] == "assistant":
        if i in reasoning_content_map:
            message["reasoning_content"] = reasoning_content_map[i]
        elif "deepseek-reasoner" in self.model_name and "reasoning_content" not in message:
            message["reasoning_content"] = " "
'''

(
    when(
        ChatDeepSeek._get_request_payload,
        'for message in payload["messages"]:',
    )
    .do(do4)
)

from loguru import logger
import sys

logger.remove()
logger.add(sys.stderr, level="INFO")