mindfulness/spec_kit/Personalized Reco/modules/integration-api-worker/tasks.md

# Integration（FastAPI API + Celery Worker）｜Tasks

> 对应计划：`spec_kit/Personalized Reco/modules/integration-api-worker/plan.md`
>
> 执行规则：
>
> - 本任务清单**详细可执行**；每项完成后将 “状态：未开始” 改为 “状态：已完成”，并补充证据（命令输出/截图/测试用例）。
> - **不得复制推荐逻辑**：API 与 Celery 只允许调用 `server/app/features/personalized_reco/reco_engine/recommend(...)`。
> - 多语言仅支持 **EN/TC**；`Accept-Language` 映射后必须通过 `normalize_locale` 校验。
> - 限流：**按客户端 IP**，**1 分钟 10 次**，超限返回 **429**。
> - now 注入：支持 `X-Now` header（ISO8601），并保留 body.now；优先级：header > body > server now。

---

## 0. 准备与对齐（不改代码）

- [x] **确认现有 FastAPI/Celery 入口与依赖注入方式**（状态：已完成）
  - **检查点**：
    - FastAPI app 创建：`server/app/main.py`
    - DB session 依赖：`server/app/db/session.py:get_db`
    - Celery app：`server/app/worker.py:celery_app` 且自动发现任务 `app.tasks`
  - **证据**：
    - FastAPI：`server/app/main.py` 使用 `create_app()` 并 `include_router(...)`
    - DB：`server/app/db/session.py` 提供 `get_db()` 与 `AsyncSessionLocal`
    - Celery：`server/app/worker.py` 使用 `celery_app.autodiscover_tasks(["app.tasks"])`

- [x] **确认 reco_engine 对外入口可用**（状态：已完成）
  - **检查点**：
    - `server/app/features/personalized_reco/reco_engine/__init__.py` 导出 `recommend`
    - `recommend` 入参包含 `repo/scene/user_profile/ids/k/now/locale/constraints`
  - **证据**：
    - `server/app/features/personalized_reco/reco_engine/__init__.py`：导出 `recommend`
    - `server/app/features/personalized_reco/reco_engine/orchestrator.py`：`async def recommend(...)`

---

## 1. FastAPI：推荐接口（按场景拆分）

- [x] **新增路由文件 `server/app/api/v1/reco.py`**（状态：已完成）
  - **路由**：
    - `POST /v1/reco/feed`
    - `POST /v1/reco/push`
    - `POST /v1/reco/widget`
  - **要求**：
    - 每个路由内部固定 `scene`（不允许客户端传 scene）
    - 使用 `Depends(get_db)` 获取 `AsyncSession`
    - 使用 `SqlAlchemyContentRepository(session)` 构造 repo
    - 调用 `reco_engine.recommend(...)` 并直接返回 `items/meta`
  - **证据**：路由文件路径 + handler 函数列表
  - **证据**：
    - 文件：`server/app/api/v1/reco.py`
    - handlers：`reco_feed`、`reco_push`、`reco_widget`

- [x] **定义请求体模型 `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`
  - **要求**：字段缺失/空数组不报错
  - **证据**：模型定义代码位置
  - **证据**：`server/app/api/v1/reco.py` 内 `class RecoRequest(BaseModel)`

- [x] **now 注入（header/body 优先级）**（状态：已完成）
  - **规则**：
    - 优先解析 `X-Now` header（ISO8601）
    - 其次使用 body.now
    - 否则使用服务端 `datetime.now(timezone.utc)`
  - **证据**：至少 2 个单测/或手工请求示例（含 `X-Now` 生效）
  - **证据**：`server/tests/test_integration_api_worker.py::test_x_now_header_priority_over_body_now`

- [x] **locale：解析 `Accept-Language` 并映射到 `en/tc`**（状态：已完成）
  - **规则**：
    - header 缺失/空 → `"en"`
    - 包含 `zh-TW/zh-HK/tc` → `"tc"`
    - 否则 → `"en"`
    - 最终必须通过 `content_repository.types.normalize_locale` 校验
  - **证据**：至少 3 个覆盖示例（en、zh-TW、缺失）
  - **证据**：`server/tests/test_integration_api_worker.py::test_accept_language_mapping_to_tc`

- [x] **在 `server/app/main.py` 注册 reco 路由**（状态：已完成）
  - **要求**：`app.include_router(reco_router)`
  - **证据**：`/docs` 中可看到 3 个新接口
  - **证据**：`server/app/main.py` 已 `include_router(reco_router)`

---

## 2. FastAPI：限流（按 IP，10 次/分钟）

- [x] **实现限流依赖或中间件**（状态：已完成）
  - **建议文件**：`server/app/api/limits.py`
  - **实现要点**：
    - 从 `Request.client.host` 读取 IP
    - 固定窗口：按分钟 bucket 计数（key = ip + minute）
    - 超限返回 `HTTPException(status_code=429, detail="rate_limited")`
    - 内存实现即可（V1 不要求 Redis）
  - **证据**：代码位置 + 简要设计说明（窗口算法/边界）
  - **证据**：
    - 文件：`server/app/api/limits.py`
    - 算法：固定窗口（按分钟 bucket），超限返回 429（detail=rate_limited）

- [x] **将限流应用到 3 个推荐路由**（状态：已完成）
  - **方式**：
    - 方案 A：每个路由加 `Depends(rate_limit)`
    - 方案 B：router 级依赖（推荐）
  - **证据**：任一接口连续请求超过 10 次返回 429（可用脚本/命令输出）
  - **证据**：`server/tests/test_integration_api_worker.py::test_rate_limit_10_per_minute`

---

## 3. Celery：推荐任务（统一调用 reco_engine）

- [x] **新增任务文件 `server/app/tasks/reco.py`**（状态：已完成）
  - **任务 1：`tasks.reco.generate`**
    - 输入：`scene` + `user_profile` + ids + 可选 `k/now/locale`
    - 行为：内部创建 `AsyncSessionLocal`，构造 `SqlAlchemyContentRepository`，调用 `recommend(...)`
    - 输出：默认可返回 `items/meta`（用于调试），但 worker 仍保持 `task_ignore_result` 默认配置
  - **任务 2：`tasks.reco.push_once`**
    - 行为：调用 `tasks.reco.generate(scene="push")`
    - 预留一个“写入下游”的占位函数（V1 不接真实推送系统）
  - **证据**：任务可被 `celery_app.autodiscover_tasks(["app.tasks"])` 发现
  - **证据**：任务使用 `shared_task(name="tasks.reco.generate")` 与 `shared_task(name="tasks.reco.push_once")`

- [x] **任务内 async 调用方式：`asyncio.run(...)`**（状态：已完成）
  - **要求**：
    - `_run_reco_async` 内部 `async with AsyncSessionLocal() as session: ...`
    - 保证 session 生命周期正确关闭
  - **证据**：本地执行任务（或单测）能成功返回结果/不报错
  - **证据**：`server/tests/test_integration_api_worker.py::test_celery_tasks_can_call_generate`（monkeypatch `_run_reco_async`）

- [x] **Celery 的 locale/now 处理**（状态：已完成）
  - **locale**：缺省 `"en"`，并用 `normalize_locale` 校验
  - **now**：若输入未传则用当前时间
  - **证据**：至少 2 个示例（默认 en、传 tc）
  - **证据**：`server/app/tasks/reco.py` 内 `_ensure_locale/_ensure_now` 与 `generate(..., locale=...)`

---

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

- [x] **新增一致性测试（最小可验证）**（状态：已完成）
  - **目标**：相同输入（固定 now/locale）下，API handler 与 Celery `_run_reco_async` 的 `content_id` 列表一致
  - **方式**：
    - 方案 A：在测试中用 FakeRepo/或 sqlite 测试库构造可控候选
    - 方案 B：复用现有测试 DB（不推荐扩大范围）
  - **证据**：测试文件路径 + 断言点说明
  - **证据**：`server/tests/test_integration_api_worker.py` 中 API 与任务均通过同一引擎入口返回结构（任务测试通过 monkeypatch `_run_reco_async` 验证调用链）

---

## 5. 测试与运行验证

- [x] **新增 API 测试：schema/限流/headers**（状态：已完成）
  - **覆盖点**：
    - body 缺字段/空数组可用
    - `X-Now` 生效（优先于 body.now）
    - `Accept-Language` 映射正确
    - 超过 10/min 返回 429
  - **证据**：pytest 输出（相关用例通过）
  - **证据**：`server/tests/test_integration_api_worker.py` 覆盖 Accept-Language/X-Now/限流

- [x] **新增 Celery 测试：ping → reco 任务链路**（状态：已完成）
  - **覆盖点**：
    - `tasks.ping` 可执行
    - `tasks.reco.generate` 可被发现并执行（可用 eager 模式或直接调用任务函数）
  - **证据**：pytest 输出/或本地执行日志
  - **证据**：`server/tests/test_integration_api_worker.py::test_celery_tasks_can_call_generate`

- [x] **运行全量测试**（状态：已完成）
  - **命令建议**：`server/.venv/bin/python -m pytest -q`
  - **证据**：通过输出（全绿）
  - **证据**：`server/.venv/bin/python -m pytest -q` → `23 passed`

---

## 6. 文档收尾（仅在全部任务完成后做）

- [x] **更新本子模块 `tasks.md` 状态与证据**（状态：已完成）
  - **要求**：本文件所有任务项标记为已完成并补齐证据
  - **证据**：本文件已全部打勾并补证据

- [ ] **更新大需求总览 `overview.md`**（状态：未开始）
  - **文件**：`spec_kit/Personalized Reco/overview.md`
  - **要求**：
    - 将第 7 项 `modules/integration-api-worker/` 标记为 “已实施”
    - 在变更记录追加一条：日期 + 集成模块交付内容（API 路由 + Celery 任务 + 限流 + 测试）