# Integration（FastAPI API + Celery Worker）｜Plan

> 对应规范：`spec_kit/Personalized Reco/modules/integration-api-worker/spec.md`
>
> 依赖（已实现）：
>
> - `server/app/features/personalized_reco/reco_engine/`：统一引擎入口 `recommend(...)`
> - `server/app/features/personalized_reco/content_repository/`：`SqlAlchemyContentRepository`
> - `server/app/db/session.py`：`get_db` / `AsyncSessionLocal`
> - `server/app/worker.py`：`celery_app`
>
> 本计划已按确认项固化：
>
> - API 路由：按场景拆分（`/v1/reco/feed`、`/v1/reco/push`、`/v1/reco/widget`）
> - locale：从 `Accept-Language` 解析并映射到 `en/tc`，缺省 `en`
> - 限流：**按客户端 IP**，**1 分钟 10 次**
> - Celery：实现 `tasks.reco.generate` + `tasks.reco.push_once`
> - Celery 内调用 async：使用 `asyncio.run(...)`
> - now 注入：支持 Header `X-Now`（方案 B），并保留请求体 `now`（spec 已定义）

---

## 1. 目标与交付物

### 1.1 目标

- 对外提供推荐能力：
  - **FastAPI**：客户端同步获取推荐结果。
  - **Celery**：后台任务式生成推荐（Push/Widget 的定时/批处理）。
- **不复制推荐逻辑**：API 与任务均只调用同一 `Reco Engine`。
- 提供基础可用的 **IP 限流**（1 分钟 10 次）。
- 支持 **now 注入** 以实现确定性回归测试。
- 多语言仅支持 **EN/TC**，不允许语言回退。

### 1.2 交付物（tasks 阶段落地）

- 新增 API 路由文件（建议）：
  - `server/app/api/v1/reco.py`
- `main.py` 注册路由：
  - `app.include_router(reco_router)`
- 新增 Celery 任务：
  - `server/app/tasks/reco.py`（包含 `tasks.reco.generate`、`tasks.reco.push_once`）
- 新增限流中间件/依赖：
  - `server/app/api/limits.py`（或 `server/app/core/ratelimit.py`）
- 单元/集成测试（至少）：
  - API schema 校验（请求/响应模型）
  - 限流行为（同 IP 超过阈值返回 429）
  - Celery 任务能跑通 `ping -> reco.generate`

---

## 2. FastAPI 设计

### 2.1 路由与接口

新增推荐路由：

- `POST /v1/reco/feed`
- `POST /v1/reco/push`
- `POST /v1/reco/widget`

说明：

- 每个路由内部将 `scene` 固定为对应场景，避免客户端传错。
- `k` 若未传：按引擎默认（建议：feed=30，push/widget=1；此默认可在 API 层写死或由调用方显式传入）。

### 2.2 请求/响应模型（建议）

请求体 `RecoRequest`（pydantic）：

- `k: Optional[int]`
- `user_profile: UserProfileV1_2`
- `already_recommended_ids: list[str|int] = []`
- `touched_or_viewed_ids: list[str|int] = []`
- `now: Optional[datetime] = None`（用于测试；生产通常不传）

响应体 `RecoResponse`（pydantic）：

- `items: list[RecommendedItem]`
- `meta: RecoMeta`

> 说明：可以直接复用引擎的 `RecoEngineResult`/`RecommendedItem`/`RecoMeta` 作为 response_model，减少重复。

### 2.3 now 注入优先级（确定性）

支持两种注入：

- Header：`X-Now`（ISO8601 字符串，如 `2026-02-02T12:00:00Z`）
- Body：`now`

建议优先级：

1. 若 `X-Now` 存在且可解析 → 使用 header 的时间
2. 否则若 body.now 存在 → 使用 body.now
3. 否则 → 使用服务端 `datetime.now(timezone.utc)`

### 2.4 locale 获取与映射（EN/TC）

来源：HTTP Header `Accept-Language`

建议解析规则（无需额外依赖）：

- 若 header 缺失/空 → `"en"`
- 若包含 `zh-TW`/`zh-HK`/`tc` → `"tc"`
- 否则默认 `"en"`

随后调用 `content_repository.types.normalize_locale(locale)` 做严格校验（保证只出 `en/tc`）。

### 2.5 依赖注入与数据库会话

API 层使用 `Depends(get_db)` 注入 `AsyncSession`：

- 在 handler 内创建 `SqlAlchemyContentRepository(session)`
- 调用 `reco_engine.recommend(repo=..., scene=..., ...)`

### 2.6 限流（按 IP：1 分钟 10 次）

实现方式（V1 推荐：无外部依赖、内存版）：

- 在 FastAPI 层添加一个依赖或中间件：
  - 从 `Request.client.host` 取 IP（若有反代需后续支持 `X-Forwarded-For`，V1 先不做）
  - 使用滑动窗口或固定窗口计数（推荐固定窗口：按分钟 bucket）
  - 超过阈值：返回 `HTTP 429`，响应体包含 `detail="rate_limited"`

注意与取舍：

- 内存限流在多进程/多实例下不共享（V1 可接受）；后续可升级为 Redis 限流。

---

## 3. Celery Worker 设计

### 3.1 任务列表

- `tasks.reco.generate`
  - 输入：与 API 等价，但建议 payload 小（user_profile + ids + scene + 可选 now/locale）
  - 输出：默认忽略结果（worker 已配置 `task_ignore_result`），但函数可返回 `items/meta` 用于调试
- `tasks.reco.push_once`
  - 输入：尽量只包含 push 需要字段（user_profile + ids + 可选 now/locale）
  - 行为：内部调用 `tasks.reco.generate(scene="push")`，并预留“写入下游”的占位函数（V1 不接真实推送系统）

### 3.2 任务内调用推荐引擎（async → sync）

由于 `Reco Engine` 为 async，Celery task 为 sync，采用：

- `asyncio.run(_run_reco_async(...))`

其中 `_run_reco_async` 负责：

- `async with AsyncSessionLocal() as session:`
- `repo = SqlAlchemyContentRepository(session)`
- `await recommend(repo=repo, ...)`

说明：

- Celery 环境通常没有运行中的事件循环，`asyncio.run` 可用。
- 若未来引入 async worker/或在已有 loop 环境中调用，再考虑改为“可复用事件循环工具”。

### 3.3 locale 与 now

- locale：
  - Celery 输入可直接传 `"en"/"tc"`，缺省 `"en"`
  - 仍通过 `normalize_locale` 严格校验
- now：
  - 任务输入支持传入 `now`（用于回归测试/离线批处理），否则用服务端当前时间

---

## 4. 一致性策略（API vs Celery）

必须保证：

- API 与任务都只调用 `reco_engine.recommend`
- 对同一份输入（含固定 now/locale），输出 items 结果一致（允许浮点微差）

建议做一个对比测试：

- 在测试中构造固定 `now`，用同样的 repo/同样的输入分别走 API handler 与 Celery 的 `_run_reco_async`，断言 `content_id` 列表一致。

---

## 5. 错误处理与返回规范

### 5.1 API 错误处理

- 请求体校验失败：FastAPI 422
- locale 不支持：返回 200 但 items 为空（由引擎兜底）或 400（可选）
  - V1 建议：保持与引擎一致，返回空 items + meta，并在 meta.config_snapshot 记录错误 stage
- 限流触发：429

### 5.2 Celery 错误处理

- 任务内部捕获异常并记录日志
- 默认不回写结果，避免 Redis 占用
- 必要时将错误信息写入任务日志或后续的可观测系统（V2）

---

## 6. 安全与性能（V1）

- 限流：按 IP 10/min，保护服务与数据库
- Payload 控制：
  - Celery 输入避免传大数组；历史集合若过大，后续演进为“传引用 ID”
- 多语言：只支持 EN/TC，不做语言回退（与 repository 口径一致）