# ODS 层任务详解

> 本文档说明飞球 ETL 系统中 ODS（操作数据存储）层的所有任务。
> ODS 层负责从上游 SaaS API 抽取原始业务数据并落地到 PostgreSQL（`ods` schema），保留源 payload 便于回溯。

---

## 概述

ODS 层采用**声明式配置**驱动的通用任务模式：由 `BaseOdsTask` + `OdsTaskSpec` 配置驱动，通过 `ODS_TASK_CLASSES` 字典动态注册，共 22 个任务。

所有 ODS 任务写入 `ods.*` 表，原始 API 响应以 JSON 格式存入 `payload` 列，元数据列（`fetched_at`、`source_file`、`content_hash` 等）自动填充。

> **历史说明**：早期版本曾有 14 个独立 ODS 任务（ORDERS、PAYMENTS、MEMBERS 等），写入不存在的 `billiards.*` schema。
> 这些任务已于 2026-02-14 废弃删除，全部由下述通用 ODS 任务替代。

### 任务总览

| 任务代码 | 动态类名 | API 端点 | 目标 ODS 表 | 说明 |
|----------|----------|----------|-------------|------|
| `ODS_ASSISTANT_ACCOUNT` | `OdsAssistantAccountsTask` | `/PersonnelManagement/SearchAssistantInfo` | `assistant_accounts_master` | 助教账号档案 |
| `ODS_SETTLEMENT_RECORDS` | `OdsOrderSettleTask` | `/Site/GetAllOrderSettleList` | `settlement_records` | 结账记录 |
| `ODS_TABLE_USE` | `OdsTableUseTask` | `/Site/GetSiteTableOrderDetails` | `table_fee_transactions` | 台费计费流水 |
| `ODS_ASSISTANT_LEDGER` | `OdsAssistantLedgerTask` | `/AssistantPerformance/GetOrderAssistantDetails` | `assistant_service_records` | 助教服务流水 |
| `ODS_STORE_GOODS_SALES` | `OdsGoodsLedgerTask` | `/TenantGoods/GetGoodsSalesList` | `store_goods_sales_records` | 门店商品销售流水 |
| `ODS_PAYMENT` | `OdsPaymentTask` | `/PayLog/GetPayLogListPage` | `payment_transactions` | 支付流水 |
| `ODS_REFUND` | `OdsRefundTask` | `/Order/GetRefundPayLogList` | `refund_transactions` | 退款流水 |
| `ODS_PLATFORM_COUPON` | `OdsCouponVerifyTask` | `/Promotion/GetOfflineCouponConsumePageList` | `platform_coupon_redemption_records` | 平台/团购券核销 |
| `ODS_MEMBER` | `OdsMemberTask` | `/MemberProfile/GetTenantMemberList` | `member_profiles` | 会员档案 |
| `ODS_MEMBER_CARD` | `OdsMemberCardTask` | `/MemberProfile/GetTenantMemberCardList` | `member_stored_value_cards` | 会员储值卡 |
| `ODS_MEMBER_BALANCE` | `OdsMemberBalanceTask` | `/MemberProfile/GetMemberCardBalanceChange` | `member_balance_changes` | 会员余额变动 |
| `ODS_RECHARGE_SETTLE` | `OdsRechargeSettleTask` | `/Site/GetRechargeSettleList` | `recharge_settlements` | 充值结算 |
| `ODS_GROUP_PACKAGE` | `OdsPackageTask` | `/PackageCoupon/QueryPackageCouponList` | `group_buy_packages` | 团购套餐定义（含详情子流程，见下方说明） |
| `ODS_GROUP_BUY_REDEMPTION` | `OdsGroupBuyRedemptionTask` | `/Site/GetSiteTableUseDetails` | `group_buy_redemption_records` | 团购套餐核销 |
| `ODS_INVENTORY_STOCK` | `OdsInventoryStockTask` | `/TenantGoods/GetGoodsStockReport` | `goods_stock_summary` | 库存汇总 |
| `ODS_INVENTORY_CHANGE` | `OdsInventoryChangeTask` | `/GoodsStockManage/QueryGoodsOutboundReceipt` | `goods_stock_movements` | 库存变化记录 |
| `ODS_TABLES` | `OdsTablesTask` | `/Table/GetSiteTables` | `site_tables_master` | 台桌维表 |
| `ODS_GOODS_CATEGORY` | `OdsGoodsCategoryTask` | `/TenantGoodsCategory/QueryPrimarySecondaryCategory` | `stock_goods_category_tree` | 库存商品分类树 |
| `ODS_STORE_GOODS` | `OdsStoreGoodsTask` | `/TenantGoods/GetGoodsInventoryList` | `store_goods_master` | 门店商品档案 |
| `ODS_TABLE_FEE_DISCOUNT` | `OdsTableDiscountTask` | `/Site/GetTaiFeeAdjustList` | `table_fee_discount_records` | 台费折扣/调账 |
| `ODS_TENANT_GOODS` | `OdsTenantGoodsTask` | `/TenantGoods/QueryTenantGoods` | `tenant_goods_master` | 租户商品档案 |
| `ODS_STAFF_INFO` | `OdsStaffInfoTask` | `/PersonnelManagement/SearchSystemStaffInfo` | `staff_info_master` | 员工档案（含在职/离职） |

> 所有目标表均位于 `ods` schema 下。

### ODS_GROUP_PACKAGE 详情子流程

`ODS_GROUP_PACKAGE` 任务通过 `detail_endpoint` 配置启用了二级详情拉取：

| 配置项 | 值 |
|--------|-----|
| `detail_endpoint` | `/PackageCoupon/QueryPackageCouponInfo` |
| `detail_target_table` | `ods.group_buy_package_details` |
| `detail_param_builder` | `lambda rec: {"couponId": rec["id"]}` |
| `detail_data_path` | `("data",)` |
| `detail_id_column` | `id` |

执行流程：
1. 主流程从 `QueryPackageCouponList` 拉取团购列表 → 写入 `ods.group_buy_packages`
2. 子流程从 `ods.group_buy_packages` 提取所有 `id`
3. 串行调用 `QueryPackageCouponInfo`（通过 `UnifiedPipeline` + `RateLimiter`），获取每个团购的详情
4. 详情数据经字段提取后写入 `ods.group_buy_package_details`（全量快照，UPSERT on `coupon_id`）

详情表字段映射见 `docs/database/ODS/mappings/mapping_QueryPackageCouponInfo_group_buy_package_details.md`。

---

## 通用 ODS 任务架构（BaseOdsTask + OdsTaskSpec 模式）

通用 ODS 任务采用**声明式配置**驱动：开发者只需定义一个 `OdsTaskSpec` 数据类实例，由 `BaseOdsTask` 提供统一的 `execute()` 流程，再通过 `_build_task_class()` 工厂函数动态生成 Python 类，最终在 `ODS_TASK_CLASSES` 字典中注册。

核心优势：
- **零代码新增任务**：只需添加一条 `OdsTaskSpec` 配置即可接入新的 API 端点
- **Schema-aware 写入**：运行时从 `information_schema` 读取目标表结构，自动匹配列名和类型，无需手写字段映射
- **统一去重与冲突处理**：通过 `content_hash` 和 `ON CONFLICT` 策略保证幂等性

### OdsTaskSpec 配置结构

`OdsTaskSpec` 是一个不可变数据类（`@dataclass(frozen=True)`），定义了单个 ODS 任务的全部配置。

#### 核心字段

| 字段 | 类型 | 说明 |
|------|------|------|
| `code` | `str` | 任务代码，如 `ODS_PAYMENT`，用于注册和调度 |
| `class_name` | `str` | 动态生成的 Python 类名，如 `OdsPaymentTask` |
| `table_name` | `str` | 目标 ODS 表全限定名，如 `ods.payment_transactions` |
| `endpoint` | `str` | 上游 API 端点路径，如 `/PayLog/GetPayLogListPage` |
| `description` | `str` | 任务描述（中文），用于日志和文档 |

#### 分页与数据提取字段

| 字段 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `data_path` | `Tuple[str, ...]` | `("data",)` | API 响应中数据的 JSON 路径，逐层深入 |
| `list_key` | `str \| None` | `None` | 数据列表在 `data_path` 下的键名（如 `"settleList"`），为 `None` 时直接取 `data_path` 下的列表 |

#### 主键与列定义字段

| 字段 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `pk_columns` | `Tuple[ColumnSpec, ...]` | `()` | 业务主键列定义，用于冲突检测（通常为 `id`） |
| `extra_columns` | `Tuple[ColumnSpec, ...]` | `()` | 额外列定义，用于从嵌套 JSON 中提取特定字段 |

#### 时间窗口字段

| 字段 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `requires_window` | `bool` | `True` | 是否需要时间窗口参数（事实表为 `True`，维度快照表为 `False`） |
| `time_fields` | `Tuple[str, str] \| None` | `("startTime", "endTime")` | API 请求中时间窗口参数的键名对 |
| `include_site_id` | `bool` | `True` | 是否在请求中传 `siteId` 参数 |
| `extra_params` | `Dict[str, Any]` | `{}` | 额外的固定请求参数 |

#### 快照与软删除字段

| 字段 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `skip_unchanged` | `bool` | `True` | 去重开关：为 True 时跳过 content_hash 与数据库最新版本相同的记录 |
| `snapshot_mode` | `SnapshotMode` | `SnapshotMode.NONE` | 快照软删除策略枚举（NONE / FULL_TABLE / WINDOW） |
| `snapshot_time_column` | `str \| None` | `None` | WINDOW 模式下用于限定软删除范围的时间列名 |

> **SnapshotMode 枚举**：
> - `NONE`：不做快照对比，不触发软删除
> - `FULL_TABLE`：全表快照——API 返回全量数据，不在返回集中的记录插入一条 `is_delete=1` 的删除版本行
> - `WINDOW`：窗口快照——仅在 `snapshot_time_column` 指定的时间列范围内做快照对比
>
> **校验约束**（`__post_init__`）：
> - `WINDOW` 模式必须指定 `snapshot_time_column`（非空字符串），否则抛出 `ValueError`
> - `FULL_TABLE` 或 `NONE` 模式不允许指定 `snapshot_time_column`，否则抛出 `ValueError`
>
> **快照模式说明**：当 `snapshot_mode` 为 `FULL_TABLE` 或 `WINDOW` 时，任务会在每个分段结束后调用 `_mark_missing_as_deleted()`，为 API 未返回但数据库中存在的记录**插入一条 `is_delete=1` 的新版本行**（而非 UPDATE 历史版本）。此行为还需配合运行时配置 `run.snapshot_missing_delete=True` 才会生效。

#### 元数据控制字段

| 字段 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `include_source_file` | `bool` | `True` | 是否写入 `source_file` 列 |
| `include_source_endpoint` | `bool` | `True` | 是否写入 `source_endpoint` 列 |
| `include_fetched_at` | `bool` | `True` | 是否写入 `fetched_at` 列。仅控制元数据写入，不影响去重行为（去重由 `skip_unchanged` 控制） |
| `include_record_index` | `bool` | `False` | 是否写入 `record_index` 列（当表 PK 包含 `record_index` 时，该序号成为冲突检测的一部分） |

### ColumnSpec 列映射定义

`ColumnSpec` 是不可变数据类，定义单个列的映射规则：

| 字段 | 类型 | 说明 |
|------|------|------|
| `column` | `str` | 目标数据库列名 |
| `sources` | `Tuple[str, ...]` | 源 JSON 字段名列表，按优先级回退（支持点号路径） |
| `required` | `bool` | 是否为必填字段（缺失时跳过整条记录） |
| `default` | `Any` | 默认值（源字段全部为空时使用） |
| `transform` | `Callable \| None` | 类型转换函数 |

代码中提供了三个快捷构造函数：

| 函数 | 用途 | transform |
|------|------|-----------|
| `_int_col(name, *sources)` | 整数列 | `TypeParser.parse_int` |
| `_decimal_col(name, *sources)` | 金额列（保留 2 位小数） | `TypeParser.parse_decimal(v, 2)` |
| `_bool_col(name, *sources)` | 布尔列 | 自定义 `_to_bool` |

---

### BaseOdsTask 通用 execute 流程

所有通用 ODS 任务共享 `BaseOdsTask.execute()` 方法，流程如下：

```
execute(cursor_data)
│
├── 1. 解析时间窗口（_resolve_window）
│   ├── 优先级：用户手动覆盖 > 游标 + MAX(fetched_at) 兜底 > 默认窗口
│   └── 若游标推进但表未实际入库，回退到 MAX(fetched_at) 作为起点
│
├── 2. 窗口分段（build_window_segments）
│   └── 按闲忙时段或配置拆分为多个子窗口
│
├── 3. 准备运行参数
│   ├── 读取 store_id、page_size
│   ├── 解析快照模式配置
│   ├── 获取表主键列（_get_table_pk_columns）
│   └── 检查表是否有 is_delete 列
│
├── 4. 逐段执行（for seg_start, seg_end in segments）
│   ├── 4a. 构建 API 请求参数（_build_params）
│   │   └── 特殊处理：GetGoodsInventoryList 端点的 siteId 需传数组格式
│   ├── 4b. 分页抓取（api.iter_paginated）→ _insert_records_schema_aware 写入
│   ├── 4c. 软删除标记（若快照模式启用）
│   └── 4d. 提交事务（db.commit）
│
├── 5. 汇总结果
│   └── 返回 {status, counts, window, segments, request_params}
│
└── 异常处理
    └── db.rollback + 记录错误日志 + 重新抛出
```

#### Schema-aware 写入（`_insert_records_schema_aware`）

核心写入方法，运行时动态适配表结构：

1. **读取表结构**：从 `information_schema.columns` 获取目标表的所有列名、数据类型
2. **读取主键**：从 `information_schema.table_constraints` 获取 PRIMARY KEY 列
3. **记录合并**：`_merge_record_layers()` 将嵌套 JSON 展平为单层字典
4. **is_delete 标准化**：`_normalize_is_delete_flag` 统一为 `0/1`
5. **content_hash 计算**：基于原始 payload（未展平的 API 返回 JSON）+ `is_delete` 计算 SHA-256 哈希
6. **content_hash 去重**（`skip_unchanged=True` 时）：与数据库中同一业务主键的最新 `content_hash` 比对，相同则跳过
7. **值映射**：逐列匹配，特殊列（`payload`、`source_file`、`fetched_at`、`content_hash`）自动填充；对 `settlement_records` 和 `recharge_settlements` 表，若记录中包含 `siteProfile` 嵌套对象，会自动回填 `tenantid`、`siteid`、`sitename` 等缺失字段
8. **冲突处理**：根据 `run.ods_conflict_mode` 配置：

   | 模式 | SQL 行为 | 说明 |
   |------|----------|------|
   | `update` | `ON CONFLICT ... DO UPDATE SET ... WHERE IS DISTINCT FROM` | 全字段对比，仅在有变化时更新 |
   | `backfill` | `ON CONFLICT ... DO UPDATE SET COALESCE(existing, new) WHERE ... IS NULL` | 仅回填 NULL 列 |
   | `nothing` | `ON CONFLICT ... DO NOTHING` | 跳过已存在记录 |

9. **批量写入**：使用 `psycopg2.extras.execute_values` 分块写入，通过 `RETURNING (xmax = 0)` 区分插入和更新

#### 软删除标记（`_mark_missing_as_deleted`）

当快照模式启用时（`snapshot_mode` 为 `FULL_TABLE` 或 `WINDOW`），任务在每个分段结束后执行软删除。软删除采用**插入删除版本行**的方式（而非 UPDATE 历史版本），保持 ODS 追加写入的语义一致性。

**触发路径：**

- **路径 A（API 返回 is_delete）**：上游 API 的 JSON 响应中自带 `is_delete`/`isDelete` 等字段，由 `_normalize_is_delete_flag` 标准化后随记录正常写入。因为 `is_delete` 参与 content_hash 计算，is_delete 变化会产生新 hash → 自动插入新版本行。此路径无需额外改造。
- **路径 B（快照空缺，`_mark_missing_as_deleted`）**：通过快照对比发现上游已不存在的记录，插入一条 `is_delete=1` 的删除版本行。

**路径 B 执行流程：**

1. 收集本次抓取到的所有业务 ID → `fetched_keys`
2. 查询快照范围内数据库中 `is_delete != 1` 的业务 ID：
   - `FULL_TABLE` 模式：全表范围
   - `WINDOW` 模式：`WHERE {snapshot_time_column} >= window_start AND < window_end`
3. 差集 = 数据库中的 ID - `fetched_keys` → 缺失 ID
4. 对每个缺失 ID：
   - 读取最新版本行（`DISTINCT ON (id) ... ORDER BY fetched_at DESC`）
   - 若最新版本已是 `is_delete=1` → 跳过（幂等性保证）
   - 否则：复制该行所有字段，设 `is_delete=1`，基于原始 payload + `is_delete=1` 重算 `content_hash`，INSERT 新版本行
5. 历史版本行完全不变

**关键特性：**
- **幂等性**：若最新版本已是 `is_delete=1`，不会重复插入删除版本
- **历史不可变**：所有历史版本行的内容保持不变，不执行任何 UPDATE
- **版本追溯**：删除操作本身也是一个版本，可通过 `fetched_at` 追溯删除时间

---

### content_hash 去重机制

`content_hash` 是通用 ODS 任务的核心去重手段，所有 22 个任务默认开启（`skip_unchanged=True`）。

#### 计算方式

`_compute_content_hash(record, payload, is_delete)` 基于**原始 payload**（未展平的 API 返回 JSON）和 **is_delete** 字段计算 SHA-256 哈希：

```python
payload_str = json.dumps(payload, sort_keys=True, separators=(',',':'), ensure_ascii=False)
raw = f"{payload_str}|{is_delete}"
hash = hashlib.sha256(raw.encode("utf-8")).hexdigest()
```

- 输入是原始 API 返回的 JSON 对象（payload），天然不含 `fetched_at`、`source_file` 等元数据字段，无需排除逻辑
- `is_delete` 参与 hash 计算——当记录从正常变为删除（或反之），hash 会变化，自动产生新版本行
- 分隔符 `|` 确保 payload 和 is_delete 不会产生歧义

> 历史说明：改造前 hash 基于展平后的 `merged_rec`（排除 7 个元数据字段），由 `_sanitize_record_for_hash` 方法处理。该方法已删除，新算法更简洁且语义更清晰。

#### 去重流程

1. **计算**：对每条记录的原始 payload + is_delete 计算 SHA-256 哈希
2. **比对**：从数据库中按业务主键取最新一条记录的 `content_hash`（利用 `(业务主键, fetched_at DESC)` 索引）
3. **跳过**：若新记录的 `content_hash` 与数据库中最新记录相同，则跳过写入（计入 `skipped` 计数）

> 仅在 `skip_unchanged=True`（默认）且目标表包含 `content_hash` 列且有业务主键时生效。不再隐式依赖 `fetched_at` 列的存在。

#### 下游取数规约

DWD 层从 ODS 取最新有效版本的标准查询模式：

```sql
SELECT DISTINCT ON (id) *
FROM ods.{table_name}
WHERE is_delete IS DISTINCT FROM 1
ORDER BY id, fetched_at DESC;
```

此查询利用新增的 `(业务主键, fetched_at DESC)` 复合索引，避免全表扫描。

---

### 各任务详细配置

| 任务代码 | 需要窗口 | 快照模式（`snapshot_mode`） | `snapshot_time_column` | 特殊说明 |
|----------|----------|---------------------------|----------------------|----------|
| `ODS_ASSISTANT_ACCOUNT` | 是 | `FULL_TABLE` | — | 助教账号档案，全量抓取后标记离职/删除 |
| `ODS_SETTLEMENT_RECORDS` | 是 | `NONE` | — | 结账记录，按时间窗口增量抓取 |
| `ODS_TABLE_USE` | 否 | `WINDOW` | `create_time` | 台费计费流水 |
| `ODS_ASSISTANT_LEDGER` | 是 | `WINDOW` | `create_time` | 助教服务流水 |
| `ODS_STORE_GOODS_SALES` | 是 | `WINDOW` | `create_time` | 门店商品销售流水（2026-03-01 修复：`requires_window` 从 `False` 改为 `True`，新增 `time_fields=("startTime", "endTime")`） |
| `ODS_PAYMENT` | 否 | `NONE` | — | 支付流水 |
| `ODS_REFUND` | 否 | `WINDOW` | `pay_time` | 退款流水 |
| `ODS_PLATFORM_COUPON` | 否 | `WINDOW` | `consume_time` | 平台/团购券核销 |
| `ODS_MEMBER` | 否 | `NONE` | — | 会员档案 |
| `ODS_MEMBER_CARD` | 否 | `FULL_TABLE` | — | 会员储值卡 |
| `ODS_MEMBER_BALANCE` | 否 | `WINDOW` | `create_time` | 会员余额变动 |
| `ODS_RECHARGE_SETTLE` | 是 | `NONE` | — | 充值结算 |
| `ODS_GROUP_PACKAGE` | 否 | `FULL_TABLE` | — | 团购套餐定义 + 详情子流程（`detail_endpoint`） |
| `ODS_GROUP_BUY_REDEMPTION` | 否 | `WINDOW` | `create_time` | 团购套餐核销 |
| `ODS_INVENTORY_STOCK` | 否 | `NONE` | — | 库存汇总 |
| `ODS_INVENTORY_CHANGE` | 是 | `NONE` | — | 库存变化记录 |
| `ODS_TABLES` | 否 | `NONE` | — | 台桌维表 |
| `ODS_GOODS_CATEGORY` | 否 | `NONE` | — | 库存商品分类树 |
| `ODS_STORE_GOODS` | 否 | `FULL_TABLE` | — | 门店商品档案 |
| `ODS_TABLE_FEE_DISCOUNT` | 否 | `WINDOW` | `create_time` | 台费折扣/调账 |
| `ODS_TENANT_GOODS` | 否 | `FULL_TABLE` | — | 租户商品档案 |
| `ODS_STAFF_INFO` | 否 | `FULL_TABLE` | — | 员工档案，全量快照 |

> 所有 22 个任务默认 `skip_unchanged=True`（去重开启）。
