---
inclusion: manual
---

# 代码开发风格文档

本项目为 FastAPI + SQLAlchemy (asyncio) 的 Python 3.12 后端项目，应用主目录为 `app/`。

## 项目结构

- `app/config/` — 配置层：Pydantic Settings 统一配置
- `app/core/` — 核心基础设施：数据库、Redis、鉴权、中间件、异常处理、日志、统一响应
- `app/ai/` — AI 能力层：LLM 模型枚举与实例创建
- `app/api/` — 路由层：REST API 接口定义
- `app/models/` — ORM 模型层：SQLAlchemy 声明式映射
- `app/services/` — 业务逻辑层：Service 类
- `app/core/schemas/` — 公共 Schema：统一响应模型等

## 命名约定

### 文件命名
- 全部小写，下划线分隔，如 `func_permission_service.py`、`user_func_usage_log.py`
- 路由文件以业务名命名，如 `health.py`、`resume.py`
- ORM 模型文件与表名对应（去掉 `bg_` 前缀），如 `func_permission.py` 对应 `bg_func_permission`

### 类命名
- Service 以 `Service` 结尾，如 `FuncPermissionService`
- ORM 模型用 PascalCase 业务名，无后缀，如 `FuncPermission`、`UserFuncUsageLog`
- Pydantic Schema 按用途命名：请求参数以 `Param` 结尾，响应以 `Dto` 结尾，如 `ResumeParam`、`ResumeDto`
- 枚举类以大写命名，如 `LLM`

### 变量与函数命名
- 函数和变量使用 snake_case，如 `check_and_deduct`、`user_id`
- 私有函数以单下划线开头，如 `_insert_usage_log`
- 常量使用全大写下划线，如 `_FRIENDLY_MESSAGES`、`_SKIP_PATHS`

## 类型注解

- 所有函数参数和返回值必须有类型注解
- ORM 模型字段使用 `Mapped[T]` + `mapped_column()` 声明
- Pydantic 模型字段使用标准类型注解 + `Field()`
- 可选字段使用 `Optional[T]` 或 `T | None`
- 集合类型使用 `list[T]`、`dict[K, V]`（Python 3.12 内置泛型）

## 注释规范

- 模块级注释使用文件顶部的 docstring，说明模块用途和使用示例
- 类注释使用 docstring，说明对应的表名和用途
- 方法注释使用 docstring，简洁描述功能
- 复杂逻辑用行内注释 `#` 说明

### ORM 模型类注释
- 类 docstring 说明对应的表名和用途
- 特殊字段通过 `comment` 参数说明含义，如 `comment="状态 1=启用 0=禁用"`

### Service 类注释
- 模块级 docstring 说明该服务的主要功能、依赖服务、使用的表
- 格式示例：
  ```python
  """功能权限 Service

  校验用户功能权限并扣减库存，业务异常时回退。
  逻辑与 Java 端 FuncPermissionService 完全一致。
  """
  ```
- 每个方法用 docstring 简要说明逻辑流程，复杂方法可分步骤描述

## 分包规则

### API 路由（`app/api/`）
- 每个业务模块一个路由文件，如 `health.py`、`resume.py`
- 使用 `APIRouter(prefix="/xxx", tags=["xxx"])` 定义路由前缀和标签
- 在 `app/main.py` 中注册路由

### Service（`app/services/`）
- 每个业务模块一个 Service 文件
- Service 类通过构造函数接收 `AsyncSession`，如 `def __init__(self, session: AsyncSession)`
- 不使用全局 Service 实例，每次请求通过依赖注入创建

### ORM 模型（`app/models/`）
- 每个表一个模型文件
- 所有模型继承 `app.core.database.Base`
- 表名通过 `__tablename__` 指定

### Pydantic Schema（`app/core/schemas/`）
- 公共 Schema 放在 `app/core/schemas/` 下，如 `responses.py`
- 业务相关的请求/响应 Schema 放在对应的 `app/api/` 或 `app/services/` 同级目录，或集中在 `app/core/schemas/{功能模块}/` 下

## 获取当前登录用户

- 通过 `RequestContext.user_id.get()` 获取当前登录用户 ID
- 或通过依赖注入 `Depends(require_login)` 获取并校验
- 需要功能权限校验时使用 `Depends(func_permission("func_code"))`

## 接口规范

- Router 只负责参数接收和调用 Service，不写业务逻辑
- 白名单路径（无需鉴权）在 `settings.auth_whitelist` 中配置
- POST 用 `@router.post()`，GET 用 `@router.get()`
- 复杂参数使用 Pydantic 模型 + `Body()`，简单参数使用 `Query()` 或 `Path()`
- 路由方法直接返回业务数据，由 `ResponseWrapMiddleware` 自动包装为 `StandardResponse`

## 异常处理

### 核心约定
- 统一异常与全局处理器都集中在 `app/core/exceptions.py`，`main.py` 只调用一次 `register_exception_handlers(app)`
- **自定义异常一律 HTTP 500**，真实业务含义靠响应体里的 `code` 区分（前端拦截器读 `code`）
- 抛异常时传一句具体原因（detail）即可，不用记错误码；不传就用类别名兜底
  - `raise BizError("不支持的商品")` → msg = `业务异常[不支持的商品]`
  - `raise BizError()` → msg = `业务异常`
- **不要** catch 后吞掉异常，交由全局异常处理器统一处理

### 自定义异常类（`app/core/exceptions.py`）
业务代码优先抛下列语义化异常，不要直接 `raise HTTPException`：

| 异常类 | code | 类别名 | 典型场景 |
|--------|------|--------|----------|
| `ParamError` | 4000 | 参数异常 | 入参缺失/非法、模型档位不支持 |
| `AssertError` | 4100 | 断言异常 | 业务前置条件不满足（由 `Assert` 工具抛出） |
| `BizError` | 4200 | 业务异常 | 资源不存在、限频、验证码、越权、会员校验 |
| `CreditError` | 4300 | 扣费异常 | 余额不足、扣费失败（前端引导充值） |
| `SysError` | 5000 | 系统异常 | 已知内部错误 / 上游模型调用失败（主动抛） |

- 所有自定义异常继承 `AppError`，处理器只在基类上注册，靠继承链自动覆盖全部子类
- 流式（SSE）场景用 `exc.to_event()` 输出 error 事件；上游异常包装用 `SysError.from_exc(exc)`
- 兜底：未建模的异常由 `global_exception_handler` 统一按系统异常（code=5000）返回，dev 环境暴露细节，生产只回通用提示

```python
from app.core.exceptions import BizError, CreditError, ParamError

if not resume:
    raise BizError("简历不存在")
if balance < cost:
    raise CreditError("积分不足")
if model not in ALLOWED:
    raise ParamError(f"不支持的模型档位: {model}")
```

### 业务断言工具（`app/core/asserts.py`）
前置条件校验优先用 `Assert` 工具类（风格参考 Spring `Assert`），条件不满足时抛 `AssertError`(code=4100)，**不要**再手写 `if not ... raise` 或裸 `assert`：

```python
from app.core.asserts import Assert

Assert.not_none(user, "用户不存在")
Assert.has_text(func_code, "功能编码不能为空")
Assert.not_empty(items, "列表不能为空")
Assert.gt(days, 0, "天数必须大于0")
Assert.eq(status, 1, "状态不可用")
```

常用方法：`is_true` / `is_false` / `not_none` / `is_none` / `has_text` / `not_empty` / `gt` / `gte` / `lt` / `lte` / `eq` / `ne`

### HTTPException
- 仅在需要返回特定 HTTP 状态码的场景（如框架层、鉴权 401/403）使用 `raise HTTPException(status_code=xxx, detail="描述")`
- 业务拒绝一律用上面的自定义异常，不要用 `HTTPException` 表达业务错误

## Redis 使用规范

- 通过 `app.core.redis.redis_client` 或依赖注入 `Depends(get_redis)` 获取客户端
- key 命名与 Java 端保持一致，如 `login:token:{userId}`
- 值统一 JSON 序列化（`json.dumps` / `json.loads`）
- 设置 TTL 时使用 `ex` 参数（秒）

## 数据库设计风格

- 与 Java 端共享同一数据库，表结构由 Java 端管理
- 表名以 `bg_` 前缀，下划线命名，如 `bg_func_permission`
- 主键 `id`，类型 `BigInteger`
- 时间字段使用 `DateTime` 类型，包含 `create_time` 和 `update_time`
- 逻辑删除字段 `is_delete`，类型 `BigInteger`，0=正常，非0=删除
- 状态字段用 `Integer`，0/1 表示，通过 `comment` 说明含义
- 查询使用 SQLAlchemy `select()` + `where()` 构建条件
- 更新使用 `update()` + `where()` + `values()`
- 会话通过 `get_db()` 依赖注入获取，自动 commit/rollback/close

## 异步规范

- 所有数据库操作、Redis 操作、HTTP 请求使用 `async/await`
- Service 方法统一使用 `async def`
- 路由处理函数统一使用 `async def`
- 避免在异步上下文中使用同步阻塞操作

## AI 调用规范

- 业务代码**不直接使用** `LLM` 枚举，而是从 `app.ai.model_config` 中引用对应模块的场景配置类
- `model_config.py` 中每个模块一个 class，每个场景一个类属性，属性值为预创建的 `ChatOpenAI` 实例
- 修改模型或调整参数只需改 `model_config.py` 一个文件，业务代码不动
- AI 调用应做好异常捕获和容错，单次失败不应影响整体流程
- 长耗时 AI 调用考虑异步执行

### 模型引用示例
```python
from app.ai.model_config import SkillGapModel, NovaChatModel

# chain 中直接使用配置类属性（已经是 ChatOpenAI 实例）
_plan_chain = (
    ChatPromptTemplate.from_messages([...])
    | SkillGapModel.AGENT_PLAN
    | StrOutputParser()
)

# 非 chain 场景直接 await 调用
result = await NovaChatModel.CHAT.ainvoke(messages)
```

### 新增 AI 场景步骤
1. 在 `app/ai/model_config.py` 对应模块的 class 中新增一个类属性，指定模型和参数
2. 在业务代码中 `from app.ai.model_config import XxxModel`，引用该属性
3. 如需新增模块，在 `model_config.py` 中新建一个 class

### AI 输出 JSON 解析
- LLM 返回的 JSON 经常被 markdown 代码块（` ```json ... ``` `）包裹，**禁止**直接使用 LangChain 的 `JsonOutputParser`
- 统一使用 `app.tool.json_helper.parse_llm_json` 解析 AI 输出的 JSON 文本
- `parse_llm_json` 会自动剥离 markdown 代码块标记，并通过 `json_repair` 做容错修复
- **不要**在各模块中自行编写 JSON 清洗/解析逻辑，统一复用 `parse_llm_json`

## 代码格式规范

### 紧凑风格
- 避免过度换行，保持代码紧凑易读
- 链式调用尽量写在一行，除非超过 120 字符
- 方法参数列表较多时，可适当换行但保持紧凑
- f-string 拼接优先写在一行

### 示例

**推荐（紧凑风格）：**
```python
# 查询语句一行
result = await session.execute(select(FuncPermission).where(FuncPermission.func_code == func_code, FuncPermission.status == 1))

# 链式操作一行
perm = result.scalar_one_or_none()

# f-string 拼接一行
log.info(f"功能权限校验 userId:{user_id} funcCode:{func_code}")

# 方法参数紧凑排列
async def check_and_deduct(self, user_id: int, func_code: str) -> int:

# 多条件 where 紧凑排列
result = await self.session.execute(select(UserFuncPermissionStock).where(
    UserFuncPermissionStock.user_id == user_id, UserFuncPermissionStock.func_code == func_code))
```