Neo-ZQYY/.kiro/specs/04-miniapp-core-business/tasks.md

# 实现计划：小程序核心业务模块（miniapp-core-business）

## 概述

基于已批准的需求和设计文档，将小程序核心业务模块拆分为增量式编码任务。按照"DDL 建表 → 触发器调度框架 → 任务生成器 → 任务管理 → 有效期轮询 → 召回检测 → 备注系统 → 路由集成"的顺序实现。后端使用 Python + FastAPI，数据库使用 PostgreSQL 纯 SQL，属性测试使用 hypothesis。所有数据库操作在测试库 `test_zqyy_app` 中进行。

## 任务

- [x] 1. 创建业务数据表和种子数据
  - [x] 1.1 创建迁移脚本 `db/zqyy_app/migrations/YYYY-MM-DD__p4_create_biz_tables.sql`
    - 在 `biz` Schema 下创建 `coach_tasks`、`coach_task_history`、`notes`、`trigger_jobs` 共 4 张表
    - 包含所有字段定义、约束、CHECK 约束（评分 1-5）、外键（coach_task_history → coach_tasks、notes → coach_tasks、coach_tasks → coach_tasks 自引用）
    - 创建部分唯一索引 `idx_coach_tasks_site_assistant_member_type`（仅 status='active'）
    - 创建查询索引 `idx_coach_tasks_assistant_status`、`idx_notes_target`
    - 使用 `IF NOT EXISTS` 幂等语法
    - 包含回滚语句（注释形式）
    - _Requirements: 1.1-1.9_

  - [x] 1.2 创建种子数据脚本 `db/zqyy_app/migrations/YYYY-MM-DD__p4_seed_trigger_jobs.sql`
    - 插入 4 条触发器配置：`task_generator`（cron, 0 4 * * *）、`task_expiry_check`（interval, 3600s）、`recall_completion_check`（event, etl_data_updated）、`note_reclassify_backfill`（event, recall_completed）
    - 使用 `ON CONFLICT (job_name) DO NOTHING` 幂等语法
    - _Requirements: 2.1-2.5_

  - [x] 1.3 在测试库执行迁移脚本并验证
    - 在 `test_zqyy_app` 中执行建表脚本和种子数据脚本
    - 验证幂等性：连续执行两次无错误
    - 验证表结构、约束、索引正确
    - 验证种子数据完整（4 条触发器配置）
    - _Requirements: 11.1-11.5, 12.1_

  - [x] 1.4 更新数据库手册和 DDL 基线
    - 创建 `docs/database/BD_Manual_biz_tables.md`，包含变更说明、兼容性影响、回滚策略、验证 SQL（至少 3 条）
    - 运行 `python scripts/ops/gen_consolidated_ddl.py` 刷新 DDL 基线
    - 在数据库手册中记录种子数据内容（触发器配置）
    - _Requirements: 12.2, 12.3, 12.4_

  - [x] 1.5 编写迁移脚本幂等性属性测试
    - **Property 13: 迁移脚本幂等性**
    - 对 DDL 脚本和种子数据脚本连续执行两次，验证第二次执行无错误且数据库状态不变
    - **Validates: Requirements 1.8, 2.5, 11.4, 11.5**

- [x] 2. 检查点 - 确保 DDL 和种子数据正确
  - 确保迁移脚本在测试库中执行成功，幂等性验证通过，如有问题请向用户确认。

- [x] 3. 实现 Pydantic 模型和纯函数核心逻辑
  - [x] 3.1 创建 Pydantic 模型 `apps/backend/app/schemas/xcx_tasks.py`
    - 定义 `TaskListItem`（含 member_name、member_phone、rs_score、heart_icon）
    - 定义 `AbandonRequest`（reason 必填，min_length=1）
    - _Requirements: 8.1, 8.2, 8.4, 8.7_

  - [x] 3.2 创建 Pydantic 模型 `apps/backend/app/schemas/xcx_notes.py`
    - 定义 `NoteCreateRequest`（含 target_type、target_id、content、task_id、rating_service_willingness、rating_revisit_likelihood，评分 ge=1 le=5）
    - 定义 `NoteOut`（含 type、content、评分、ai_score、ai_analysis）
    - _Requirements: 9.1, 9.8_

  - [x] 3.3 创建任务生成器核心纯函数 `apps/backend/app/services/task_generator.py`
    - 定义 `TaskPriority` 枚举、`TASK_TYPE_PRIORITY` 映射
    - 定义 `IndexData` 数据类
    - 实现 `determine_task_type(index_data)` 纯函数：根据 WBI/NCI/RS 指数确定任务类型
    - 实现 `should_replace_task(existing_type, new_type)` 纯函数：判断是否应替换现有任务
    - 实现 `compute_heart_icon(rs_score)` 纯函数：根据 RS 指数计算爱心 icon 档位
    - _Requirements: 3.1-3.5, 3.8, 8.2_

  - [x] 3.4 编写任务类型确定正确性属性测试
    - **Property 1: 任务类型确定正确性**
    - 生成随机 WBI/NCI/RS 值（Decimal, 0-10, 2 位小数），验证 `determine_task_type()` 返回值符合优先级规则
    - **Validates: Requirements 3.1, 3.2, 3.3, 3.5**

  - [x] 3.5 编写星星评分范围约束属性测试
    - **Property 9: 星星评分范围约束**
    - 生成随机整数（-100 到 100），验证 Pydantic 模型对 1-5 范围外的值拒绝（ValidationError）
    - **Validates: Requirements 9.8, 14.5**

  - [x] 3.6 编写爱心 icon 档位计算属性测试
    - **Property 11: 爱心 icon 档位计算**
    - 生成随机 RS 值（Decimal, 0-10, 1 位小数），验证 `compute_heart_icon()` 返回正确 icon
    - **Validates: Requirements 8.2**

- [x] 4. 实现触发器调度框架
  - [x] 4.1 创建 `apps/backend/app/services/trigger_scheduler.py`
    - 实现 `_JOB_REGISTRY` 注册表和 `register_job(job_type, handler)` 函数
    - 实现 `fire_event(event_name, payload)` 方法：查找 event 类型触发器并执行
    - 实现 `check_scheduled_jobs()` 方法：检查 cron/interval 到期 job 并执行
    - 实现 `_calculate_next_run(trigger_condition, trigger_config)` 方法：计算下次运行时间
    - 每个 job 独立事务，失败不影响其他触发器
    - _Requirements: 10.1-10.7_

  - [x] 4.2 编写触发器 next_run_at 计算属性测试
    - **Property 12: 触发器 next_run_at 计算**
    - 生成随机 cron/interval 配置和当前时间，验证 cron 类型 next_run_at > 当前时间，interval 类型 next_run_at = 当前时间 + interval_seconds
    - **Validates: Requirements 10.1, 10.2**

- [x] 5. 实现任务生成器完整流程
  - [x] 5.1 实现 `TaskGenerator.run()` 主流程
    - 通过 `auth.user_assistant_binding` 获取所有已绑定助教
    - 对每个助教，通过 FDW 读取 WBI/NCI/RS 指数（`SET LOCAL app.current_site_id`）
    - 调用 `determine_task_type()` 确定任务类型
    - 检查已存在的 active 任务：相同 task_type → 跳过；不同 task_type → 关闭旧任务 + 创建新任务 + 记录 history
    - 处理 `follow_up_visit` 的 48 小时滞留机制（expires_at 填充）
    - 更新 `trigger_jobs` 时间戳
    - _Requirements: 3.1-3.10, 4.1-4.5, 5.1-5.4_

  - [x] 5.2 编写活跃任务唯一性不变量属性测试
    - **Property 2: 活跃任务唯一性不变量**
    - 生成随机 (site_id, assistant_id, member_id, task_type) 组合，模拟插入操作，验证 active 任务最多一条
    - **Validates: Requirements 1.5, 3.6, 14.1**

  - [x] 5.3 编写任务类型变更状态机属性测试
    - **Property 3: 任务类型变更状态机**
    - 生成随机现有任务 + 新任务类型，执行变更，验证旧任务 inactive + 新任务 active + history 记录
    - **Validates: Requirements 3.7, 5.1, 5.4, 14.2**

  - [x] 5.4 编写 48 小时滞留机制属性测试
    - **Property 4: 48 小时滞留机制**
    - 生成随机 follow_up_visit 任务 + 时间偏移，验证 expires_at 填充和过期逻辑
    - **Validates: Requirements 4.1, 4.2, 4.3, 4.4, 14.3**

- [x] 6. 检查点 - 确保任务生成器测试通过
  - 运行属性测试：`cd C:\NeoZQYY && pytest tests/test_core_business_properties.py -v -k "property_1 or property_2 or property_3 or property_4"`
  - 确保所有属性测试通过，如有问题请向用户确认。

- [x] 7. 实现任务管理服务
  - [x] 7.1 创建 `apps/backend/app/services/task_manager.py`
    - 实现 `get_task_list(user_id, site_id)` 异步方法：查询活跃任务 + FDW 读取客户信息和 RS 指数 + 爱心 icon 计算 + 排序
    - 实现 `pin_task(task_id, user_id, site_id)` 异步方法：验证归属 + 设置 is_pinned=TRUE + 记录 history
    - 实现 `unpin_task(task_id, user_id, site_id)` 异步方法：验证归属 + 设置 is_pinned=FALSE
    - 实现 `abandon_task(task_id, user_id, site_id, reason)` 异步方法：验证 reason 非空 + 设置 abandoned + 记录 history
    - 实现 `cancel_abandon(task_id, user_id, site_id)` 异步方法：恢复 active + 清空 abandon_reason + 记录 history
    - 实现 `_record_history()` 内部方法
    - _Requirements: 8.1-8.8_

  - [x] 7.2 编写放弃与取消放弃往返属性测试
    - **Property 5: 放弃与取消放弃往返**
    - 生成随机 active 任务 + 非空放弃原因，执行放弃→取消放弃，验证状态恢复；空原因应返回 422
    - **Validates: Requirements 8.4, 8.6, 8.7, 14.4**

  - [x] 7.3 编写任务列表排序正确性属性测试
    - **Property 10: 任务列表排序正确性**
    - 生成随机任务列表（不同 is_pinned/priority_score/created_at），验证排序为 is_pinned DESC, priority_score DESC, created_at ASC
    - **Validates: Requirements 8.1**

  - [x] 7.4 编写状态变更历史完整性属性测试
    - **Property 15: 状态变更历史完整性**
    - 生成随机状态变更操作序列（置顶/放弃/取消放弃），验证 history 记录数量和内容正确
    - **Validates: Requirements 5.4, 8.3**

- [x] 8. 实现有效期轮询器
  - [x] 8.1 创建 `apps/backend/app/services/task_expiry.py`
    - 实现 `run()` 方法：查询 expires_at 不为 NULL 且已过期的 active 任务，标记为 inactive，记录 history
    - _Requirements: 4.3, 4.5_

- [x] 9. 实现召回完成检测器
  - [x] 9.1 创建 `apps/backend/app/services/recall_detector.py`
    - 实现 `run(payload)` 方法：通过 FDW 读取新增服务记录，匹配 active 任务标记 completed，记录 completed_at 和 completed_task_type 快照，触发 `recall_completed` 事件
    - _Requirements: 6.1-6.5_

  - [x] 9.2 编写召回完成检测与类型快照属性测试
    - **Property 6: 召回完成检测与类型快照**
    - 生成随机 active 任务 + 服务记录，执行完成检测，验证 completed_task_type 记录了完成时的 task_type 快照
    - **Validates: Requirements 6.2, 6.3, 14.6**

- [x] 10. 检查点 - 确保任务管理和召回检测测试通过
  - 运行属性测试：`cd C:\NeoZQYY && pytest tests/test_core_business_properties.py -v -k "property_5 or property_6 or property_10 or property_15"`
  - 确保所有属性测试通过，如有问题请向用户确认。

- [-] 11. 实现备注系统
  - [x] 11.1 创建备注服务 `apps/backend/app/services/note_service.py`
    - 实现 `create_note()` 异步方法：验证评分范围 + 确定 note type（关联 follow_up_visit 任务 → follow_up，否则 normal）+ INSERT + 触发 AI 应用 6 接口（占位）+ 若 ai_score >= 6 标记任务 completed
    - 实现 `get_notes()` 异步方法：按 created_at DESC 排序，包含评分和 AI 评分
    - 实现 `delete_note()` 异步方法：验证归属后硬删除
    - _Requirements: 9.1-9.9_

  - [x] 11.2 创建备注回溯重分类器 `apps/backend/app/services/note_reclassifier.py`
    - 实现 `run(payload)` 方法：查找 service_time 之后的第一条 normal 备注 → 更新为 follow_up → 触发 AI 应用 6 接口（占位）→ 根据 ai_score 生成 follow_up_visit 任务
    - 实现 `ai_analyze_note(note_id)` 占位函数（返回 None，P5 实现后替换）
    - _Requirements: 7.1-7.5_

  - [x] 11.3 编写备注回溯重分类属性测试
    - **Property 7: 备注回溯重分类**
    - 生成随机备注列表 + service_time，执行回溯，验证符合条件的 normal 备注 type 变为 follow_up
    - **Validates: Requirements 7.1, 7.2, 14.7**

  - [x] 11.4 编写备注类型自动设置属性测试
    - **Property 8: 备注类型自动设置**
    - 生成随机 task_type + 备注创建，验证关联 follow_up_visit → type=follow_up，其他 → type=normal
    - **Validates: Requirements 9.2, 9.3**

  - [x] 11.5 编写 AI 评分驱动的任务完成判定属性测试
    - **Property 14: AI 评分驱动的任务完成判定**
    - 生成随机 ai_score + 任务状态，验证 ai_score >= 6 且 active → completed，ai_score < 6 → 保持 active
    - **Validates: Requirements 7.4, 7.5, 9.5**

- [x] 12. 实现 API 路由层
  - [x] 12.1 创建小程序任务路由 `apps/backend/app/routers/xcx_tasks.py`
    - 实现 `GET /api/xcx/tasks`：获取任务列表（require_approved）
    - 实现 `POST /api/xcx/tasks/{id}/pin`：置顶任务
    - 实现 `POST /api/xcx/tasks/{id}/unpin`：取消置顶
    - 实现 `POST /api/xcx/tasks/{id}/abandon`：放弃任务（AbandonRequest 校验）
    - 实现 `POST /api/xcx/tasks/{id}/cancel-abandon`：取消放弃
    - _Requirements: 8.1-8.8_

  - [x] 12.2 创建小程序备注路由 `apps/backend/app/routers/xcx_notes.py`
    - 实现 `POST /api/xcx/notes`：创建备注（NoteCreateRequest 校验）
    - 实现 `GET /api/xcx/notes`：查询备注列表（query: target_type, target_id）
    - 实现 `DELETE /api/xcx/notes/{id}`：删除备注
    - _Requirements: 9.1-9.9_

  - [x] 12.3 在 `apps/backend/app/main.py` 中注册新路由
    - 注册 `xcx_tasks` 和 `xcx_notes` 路由
    - 验证无路由冲突
    - _Requirements: 全部_

  - [x] 12.4 注册触发器 job handler
    - 在应用启动时调用 `register_job()` 注册 `task_generator`、`task_expiry_check`、`recall_completion_check`、`note_reclassify_backfill` 四个 handler
    - _Requirements: 10.1-10.6_

- [x] 13. 检查点 - 确保所有测试通过
  - 运行属性测试：`cd C:\NeoZQYY && pytest tests/test_core_business_properties.py -v`
  - 26/26 全部通过（16.81s）

- [x] 14. 最终检查点 - 全量验证
  - 运行全部属性测试：26/26 通过（16.81s）
  - 验证迁移脚本幂等性：Property 13（3 个测试）通过
  - 验证种子数据完整性：4 条触发器配置全部存在
  - 验证表结构：coach_tasks / coach_task_history / notes / trigger_jobs 全部存在
  - 验证部分唯一索引：idx_coach_tasks_site_assistant_member_type 存在

## 备注

- 标记 `*` 的子任务为可选（属性测试），可跳过以加速 MVP
- 每个任务引用了具体的需求编号，确保可追溯
- 检查点确保增量验证
- 属性测试验证通用正确性属性（hypothesis，最少 200 次迭代）
- 所有数据库操作在测试库 `test_zqyy_app` 进行
- 迁移脚本放在 `db/zqyy_app/migrations/` 目录
- 属性测试放在 `tests/test_core_business_properties.py`（Monorepo 级）
- AI 应用 6 接口为占位实现（返回 None），由 P5 AI 集成层替换
- 维客线索功能由独立模块 `routers/member_retention_clue.py` 处理，不在本 SPEC 范围内
- FDW 查询需在事务中 `SET LOCAL app.current_site_id` 设置 RLS 隔离