# BaseTask 公共机制与执行参数

> 本文档说明飞球 ETL 系统中所有任务共享的基类机制、运行期上下文、时间窗口计算、任务注册以及 Flow 执行流程。
> 面向开发者，便于在开发新任务时遵循统一模式。

---

## 1. BaseTask 模板方法流程

`BaseTask`（位于 `tasks/base_task.py`）采用经典的**模板方法模式**，定义了 ETL 任务的标准执行骨架。所有具体任务（ODS / DWD / DWS / INDEX）均继承此基类。

### 1.1 核心方法签名

```python
class BaseTask:
    def __init__(self, config, db_connection, api_client, logger): ...
    def get_task_code(self) -> str: ...          # 子类必须实现
    def extract(self, context: TaskContext): ...  # 子类必须实现
    def transform(self, extracted, context: TaskContext): ...  # 默认直接返回 extracted
    def load(self, transformed, context: TaskContext) -> dict: ...  # 子类必须实现
    def execute(self, cursor_data: dict | None = None) -> dict: ...  # 主入口
```

### 1.2 execute() 执行流程

`execute()` 是任务的统一入口，由调度器（TaskExecutor）调用。完整流程如下：

```
execute(cursor_data)
  │
  ├─ 1. _build_context(cursor_data)        → 构建 TaskContext（含时间窗口计算）
  │
  ├─ 2. build_window_segments(...)         → 按配置切分时间窗口为多段
  │     若无切分配置 → 退化为单段 [(window_start, window_end)]
  │
  ├─ 3. 遍历每个窗口段 (window_start, window_end)：
  │     │
  │     ├─ _build_context_for_window(...)  → 为当前段构建独立 TaskContext
  │     │
  │     ├─ extract(context)                → 从数据源提取数据
  │     │
  │     ├─ transform(extracted, context)   → 清洗/转换数据
  │     │
  │     ├─ load(transformed, context)      → 写入目标表，返回统计 counts
  │     │
  │     ├─ db.commit()                     → 提交事务
  │     │
  │     └─ _accumulate_counts(...)         → 累加各段统计
  │
  └─ 4. 构建并返回结果字典
        {status, counts, window: {start, end, minutes}, segments: [...]}
```

**关键行为：**

- 每个窗口段独立执行 E/T/L 三步，段内失败会 `db.rollback()` 并抛出异常
- 多段执行时，日志会输出进度信息（已处理天数 / 总天数）
- `transform()` 默认实现为直接返回 `extracted`，子类可选择性覆盖
- 统计累加逻辑：数值类型（int/float）求和，其他类型取首次出现的值

---

## 2. TaskContext 字段含义

`TaskContext` 是一个不可变数据类（`frozen=True`），在 E/T/L 各阶段间传递运行期信息。

```python
@dataclass(frozen=True)
class TaskContext:
    store_id: int                    # 门店 ID
    window_start: datetime           # 时间窗口起始（含时区）
    window_end: datetime             # 时间窗口结束（含时区）
    window_minutes: int              # 窗口时长（分钟）
    cursor: dict | None = None       # 游标数据（来自上次运行记录）
```

| 字段 | 类型 | 说明 |
|------|------|------|
| `store_id` | `int` | 当前门店标识，从 `config.get("app.store_id")` 读取 |
| `window_start` | `datetime` | 本次抽取的时间窗口起点，带时区信息 |
| `window_end` | `datetime` | 本次抽取的时间窗口终点，带时区信息 |
| `window_minutes` | `int` | `window_end - window_start` 的分钟数（最小为 1） |
| `cursor` | `dict \| None` | 调度器传入的游标字典，通常包含 `last_end`（上次窗口终点）等字段；首次运行时为 `None` |

---

## 3. 时间窗口计算逻辑

时间窗口由 `_get_time_window(cursor_data)` 方法计算，决定了本次任务抽取的数据范围。优先级从高到低：

### 3.1 优先级一：手动覆盖（Manual Override）

当配置中同时存在 `run.window_override.start` 和 `run.window_override.end` 时，直接使用用户指定的窗口。

- 对应 CLI 参数：`--window-start` / `--window-end`
- 支持字符串自动解析（通过 `dateutil.parser.parse`）
- 无时区信息时自动附加 `app.timezone`（默认 `Asia/Shanghai`）
- 校验：两者必须同时提供，且 `end > start`

```
window_start = run.window_override.start
window_end   = run.window_override.end
window_minutes = (end - start) 的分钟数
```

### 3.2 优先级二：游标续跑（Cursor Resume）

当存在游标数据且包含 `last_end` 字段时，从上次结束位置回退一段重叠时间开始：

```
window_start = cursor["last_end"] - overlap_seconds
window_end   = now
```

- `overlap_seconds` 来自 `config.get("run.overlap_seconds", 600)`，默认 600 秒（10 分钟）
- 重叠设计目的：防止因时钟偏差或事务延迟导致数据遗漏

### 3.3 优先级三：闲忙时段默认值（Idle/Busy Window）

当既无手动覆盖也无游标时，根据当前时间是否处于闲时窗口，选择不同的默认窗口长度：

```
闲时（idle）：window_minutes = run.window_minutes.default_idle  （默认 180 分钟）
忙时（busy）：window_minutes = run.window_minutes.default_busy  （默认 30 分钟）

window_start = now - window_minutes
window_end   = now
```

闲时判断逻辑：比较当前时间的 `HH:MM` 字符串是否在 `[idle_start, idle_end]` 区间内。

| 配置项 | 默认值 | 说明 |
|--------|--------|------|
| `run.idle_window.start` | `"04:00"` | 闲时开始 |
| `run.idle_window.end` | `"16:00"` | 闲时结束 |
| `run.window_minutes.default_idle` | `180` | 闲时默认窗口（分钟） |
| `run.window_minutes.default_busy` | `30` | 忙时默认窗口（分钟） |
| `run.overlap_seconds` | `600` | 游标重叠秒数 |

### 3.4 流程图

```
_get_time_window(cursor_data)
  │
  ├─ run.window_override.start/end 都存在？
  │   ├─ 是 → 解析并返回 (override_start, override_end)
  │   └─ 否 ↓
  │
  ├─ 判断闲忙时段 → 确定 window_minutes
  │
  ├─ cursor_data 存在且含 last_end？
  │   ├─ 是 → window_start = last_end - overlap_seconds
  │   └─ 否 → window_start = now - window_minutes
  │
  └─ window_end = now → 返回 (start, end, minutes)
```

---

## 4. 窗口分段（build_window_segments）

当时间窗口跨度较大时（如回溯一个月的数据），系统支持将窗口切分为多个小段逐段执行，避免单次请求数据量过大。

### 4.1 切分入口

`build_window_segments()` 位于 `utils/windowing.py`，由 `BaseTask.execute()` 调用：

```python
def build_window_segments(
    cfg,                    # 配置对象
    start: datetime,        # 窗口起点
    end: datetime,          # 窗口终点
    *,
    tz: ZoneInfo | None,    # 时区
    override_only: bool,    # 是否仅在手动覆盖时才切分
) -> List[Tuple[datetime, datetime]]:
```

### 4.2 切分策略

**`override_only=True`（默认行为）：**
- 仅当 `run.window_override.start/end` 都存在时，才按配置切分
- 否则强制 `split_unit="none"`，返回单段（即不切分）
- 设计意图：自动游标模式下窗口通常较短，无需切分；手动指定大范围时才需要

**切分配置项：**

| 配置项 | 默认值 | 说明 |
|--------|--------|------|
| `run.window_split.unit` | `"month"` | 切分单位：`none` / `day` / `week` / `month` |
| `run.window_split.days` | `1` | 按天切分时的步长（1 / 10 / 30） |
| `run.window_split.compensation_hours` | `0` | 窗口前后补偿小时数 |

### 4.3 底层切分函数 split_window()

```python
def split_window(start, end, *, tz, split_unit, compensation_hours, split_days=None):
```

**补偿机制：** 若 `compensation_hours > 0`，会在切分前将窗口向两端各扩展指定小时数，用于覆盖跨段边界的数据。

**各切分模式：**

| split_unit | 行为 |
|------------|------|
| `none` / `off` / `false` / `""` | 不切分，返回 `[(start, end)]` |
| `day` / `daily` | 按 `split_days` 天为步长切分（默认 1 天一段） |
| `week` / `weekly` | 按 7 天为步长切分 |
| `month` / `monthly` | 按自然月边界切分（每段从当前位置到下月 1 日 00:00） |

**示例：** 窗口 `2026-01-15 ~ 2026-03-10`，`split_unit=month`：

```
段 1: 2026-01-15 ~ 2026-02-01
段 2: 2026-02-01 ~ 2026-03-01
段 3: 2026-03-01 ~ 2026-03-10
```

---

## 5. TaskRegistry 注册方式与 TaskMeta 元数据

### 5.1 TaskMeta 数据结构

```python
@dataclass
class TaskMeta:
    task_class: type                    # 任务类（BaseTask 的子类）
    requires_db_config: bool = True     # 是否需要数据库配置（游标/运行记录）
    layer: str | None = None            # 所属层："ODS" / "DWD" / "DWS" / "INDEX" / None
    task_type: str = "etl"              # 任务类型："etl" / "utility" / "verification"
    depends_on: list[str] = field(default_factory=list)  # 依赖的任务代码列表
```

| 字段 | 说明 |
|------|------|
| `task_class` | 任务的 Python 类引用，用于 `create_task()` 时实例化 |
| `requires_db_config` | `True` 表示需要游标管理和运行记录；`False` 表示工具类/校验类任务，不走游标 |
| `layer` | 标识任务所属数据层，用于 `get_tasks_by_layer()` 按层查询 |
| `task_type` | 区分 ETL 任务、工具类任务和校验类任务 |
| `depends_on` | 声明任务间的依赖关系，`_resolve_tasks()` 会据此执行拓扑排序 |

### 5.2 注册方式

`TaskRegistry` 提供 `register()` 方法，所有任务在 `orchestration/task_registry.py` 模块加载时完成注册：

```python
class TaskRegistry:
    def register(self, task_code, task_class, requires_db_config=True, layer=None, task_type="etl", depends_on=None):
        self._tasks[task_code.upper()] = TaskMeta(...)
```

**注册示例：**

```python
# ODS 层 ETL 任务（默认 requires_db_config=True, task_type="etl"）
# 通用 ODS 任务由 ODS_TASK_CLASSES 字典动态注册
for code, task_cls in ODS_TASK_CLASSES.items():
    default_registry.register(code, task_cls, layer="ODS")

# 工具类任务（不需要游标）
default_registry.register("INIT_ODS_SCHEMA", InitOdsSchemaTask, requires_db_config=False, task_type="utility")

# 校验类任务
default_registry.register("DATA_INTEGRITY_CHECK", DataIntegrityTask, requires_db_config=False, task_type="verification")

# 带依赖声明的 DWS 任务
default_registry.register("DWS_ASSISTANT_FINANCE", AssistantFinanceTask, layer="DWS",
                          depends_on=["DWS_ASSISTANT_SALARY"])
```

### 5.3 TaskRegistry 核心方法

| 方法 | 说明 |
|------|------|
| `register(task_code, task_class, ...)` | 注册任务类及元数据 |
| `create_task(task_code, config, db, api, logger)` | 根据任务代码创建实例 |
| `get_metadata(task_code) -> TaskMeta` | 查询任务元数据 |
| `get_tasks_by_layer(layer) -> list[str]` | 获取指定层的所有任务代码 |
| `is_utility_task(task_code) -> bool` | 判断是否为工具类任务（`requires_db_config=False`） |
| `get_all_task_codes() -> list[str]` | 获取所有已注册任务代码 |

### 5.4 当前注册任务统计

| 层 | 数量 | 说明 |
|----|------|------|
| ODS | 23 | 通用 ODS 任务（由 `ODS_TASK_CLASSES` 动态生成），全部默认 `skip_unchanged=True` |
| DWD | 2 | 含核心装载任务 `DWD_LOAD_FROM_ODS` 和质量检查 |
| DWS | 17 | 助教业绩（含订单流水贡献）、会员分析、财务统计、库存汇总、运维维护（原 3 个 MV 刷新/清理任务已合并为 DWS_MAINTENANCE） |
| INDEX | 5 | 回流指数、新客转化指数、关系指数、消费力指数、手动台账导入 |
| 工具类 | 7 | Schema 初始化、手动入库、归档、校验等 |
| 校验类 | 1 | 数据完整性校验 |

> 历史说明：早期版本曾有 14 个独立 ODS 任务，已于 2026-02-14 废弃删除，全部由通用 ODS 任务替代。

---

## 6. FlowRunner Flow 执行流程

`FlowRunner`（位于 `orchestration/flow_runner.py`）负责编排多层 ETL 任务的执行顺序，并可选地运行后置校验。

> 术语说明：**Connector**（数据源连接器）指对接的上游 SaaS 平台（如飞球），对应 `apps/etl/connectors/{connector_name}/`；**Flow**（执行流程）指 ETL 任务的处理链路，描述数据从哪一层流到哪一层。CLI 参数 `--flow` 传递 Flow ID，`--pipeline` 作为已弃用别名保留。

### 6.1 Flow 定义

系统预定义了 7 种 Flow，每种 Flow 包含一组数据层：

```python
FLOW_LAYERS = {
    "api_ods":       ["ODS"],
    "api_ods_dwd":   ["ODS", "DWD"],
    "api_full":      ["ODS", "DWD", "DWS", "INDEX"],
    "ods_dwd":       ["DWD"],
    "dwd_dws":       ["DWS"],
    "dwd_dws_index": ["DWS", "INDEX"],
    "dwd_index":     ["INDEX"],
}
```

| Flow | 包含层 | 典型场景 |
|------|--------|----------|
| `api_ods` | ODS | 仅从 API 抓取数据到 ODS |
| `api_ods_dwd` | ODS → DWD | 抓取并清洗到 DWD |
| `api_full` | ODS → DWD → DWS → INDEX | 全流程 ETL |
| `ods_dwd` | DWD | 仅执行 ODS→DWD 清洗（假设 ODS 已有数据） |
| `dwd_dws` | DWS | 仅执行 DWD→DWS 汇总 |
| `dwd_dws_index` | DWS → INDEX | 汇总 + 指数计算 |
| `dwd_index` | INDEX | 仅执行指数计算 |

### 6.2 处理模式

| 模式 | 说明 |
|------|------|
| `increment_only` | 仅执行增量 ETL，不做校验（默认） |
| `verify_only` | 跳过增量 ETL，仅执行后置校验并自动修复 |
| `increment_verify` | 先执行增量 ETL，再执行后置校验并修复 |

### 6.3 run() 执行流程

```
FlowRunner.run(flow, processing_mode, ...)
  │
  ├─ 校验 Flow 名称合法性
  │
  ├─ 设置默认时间窗口（未指定时：过去 24 小时）
  │
  ├─ 根据 processing_mode 分支：
  │
  │   ┌─ verify_only ─────────────────────────────────┐
  │   │  ├─ fetch_before_verify？                      │
  │   │  │   ├─ 是 → 先执行 ODS 任务获取 API 数据     │
  │   │  │   └─ 否 → 跳过                             │
  │   │  └─ _run_verification(...)                     │
  │   └────────────────────────────────────────────────┘
  │
  │   ┌─ increment_only / increment_verify ───────────┐
  │   │  ├─ _resolve_tasks(layers) → 解析任务列表      │
  │   │  │   └─ topological_sort() → 拓扑排序          │
  │   │  ├─ task_executor.run_tasks(tasks)             │
  │   │  └─ increment_verify？                         │
  │   │      └─ 是 → _run_verification(...)            │
  │   └────────────────────────────────────────────────┘
  │
  └─ 汇总计数 → 返回结果字典
```

### 6.4 任务解析（_resolve_tasks）

`_resolve_tasks(layers)` 根据层列表解析出具体的任务代码，并执行拓扑排序确保依赖顺序：

| 层 | 解析逻辑 |
|----|----------|
| ODS | 优先使用 `config.run.ods_tasks`，否则从 TaskRegistry 按层查询 |
| DWD | 固定返回 `["DWD_LOAD_FROM_ODS"]` |
| DWS | 优先使用 `config.run.dws_tasks`，否则从 TaskRegistry 按层查询 |
| INDEX | 优先使用 `config.run.index_tasks`，否则从 TaskRegistry 按层查询 |

> 已移除所有硬编码回退列表，统一走 `TaskRegistry.get_tasks_by_layer()` 获取任务。空 Registry + 无配置时记录警告并返回空列表。
> 解析完成后通过 `orchestration/topological_sort.py` 的 Kahn's algorithm 执行拓扑排序，确保被依赖任务先于依赖方执行。

### 6.5 数据源模式

| 模式 | 说明 |
|------|------|
| `online` | 仅从上游 API 在线抓取 |
| `offline` | 仅从本地 JSON 文件入库 |
| `hybrid` | 先抓取再入库（默认） |

---

## 7. 校验框架概述

校验框架（`tasks/verification/`）提供各层数据的批量后置校验和自动补齐功能。

### 7.1 架构

```
BaseVerifier（基类）
  ├─ OdsVerifier    — 主键 + content_hash 对比，批量 UPSERT
  ├─ DwdVerifier    — 维度 SCD2 / 事实主键对比，批量 UPSERT
  ├─ DwsVerifier    — 聚合对比，批量重算 UPSERT
  └─ IndexVerifier  — 实体覆盖对比，批量重算 UPSERT
```

### 7.2 核心接口

- `get_verifier_for_layer(layer, db_connection, logger, **kwargs)` — 工厂函数，根据层名返回对应校验器实例
- `verifier.verify_and_backfill(window_start, window_end, auto_backfill, split_unit, tables)` — 执行校验并自动补齐

### 7.3 校验结果模型

| 类 | 说明 |
|----|------|
| `VerificationResult` | 单表校验结果 |
| `VerificationSummary` | 层级汇总（total_tables / consistent_tables / total_backfilled / error_tables） |
| `VerificationStatus` | 校验状态枚举 |
| `WindowSegment` | 校验时间段 |

### 7.4 在 Flow 中的触发方式

校验由 `FlowRunner._run_verification()` 触发，支持：
- 按层逐一校验
- 按表名过滤（`verify_tables` 参数）
- 时间窗口切分（`window_split` 参数）
- ODS 层支持 API 数据对比或本地 JSON 对比两种模式

---

## 8. 相关配置速查

| 配置路径 | 默认值 | 说明 |
|----------|--------|------|
| `app.store_id` | — | 门店 ID |
| `app.timezone` | `"Asia/Shanghai"` | 系统时区 |
| `run.window_override.start` | — | 手动窗口起点 |
| `run.window_override.end` | — | 手动窗口终点 |
| `run.idle_window.start` | `"04:00"` | 闲时开始 |
| `run.idle_window.end` | `"16:00"` | 闲时结束 |
| `run.window_minutes.default_idle` | `180` | 闲时默认窗口（分钟） |
| `run.window_minutes.default_busy` | `30` | 忙时默认窗口（分钟） |
| `run.overlap_seconds` | `600` | 游标重叠秒数 |
| `run.window_split.unit` | `"month"` | 切分单位 |
| `run.window_split.days` | `1` | 按天切分步长 |
| `run.window_split.compensation_hours` | `0` | 窗口补偿小时数 |
| `run.ods_tasks` | `[]` | ODS 层任务列表覆盖 |
| `run.dws_tasks` | `[]` | DWS 层任务列表覆盖 |
| `run.index_tasks` | `[]` | INDEX 层任务列表覆盖 |

---

## 9. ODS 层去重与软删除机制

> 详细的 ODS 任务配置和各任务参数矩阵见 [`ods_tasks.md`](./ods_tasks.md) 和 [`ods_task_params_matrix.md`](./ods_task_params_matrix.md)。

ODS 层是追加写入的版本化存储，每次内容变更（包括删除）都是一个新版本行。以下是两个核心机制的概要。

### 9.1 content_hash 去重（skip_unchanged）

所有 23 个 ODS 任务默认开启去重（`skip_unchanged=True`），避免无意义的版本膨胀。

- **hash 输入**：原始 API 返回的 JSON payload + `is_delete` 字段
- **hash 算法**：`json.dumps(payload, sort_keys=True, separators=(',',':'), ensure_ascii=False)` + `|` + `is_delete`，然后 SHA-256
- **去重逻辑**：与数据库中同一业务主键的最新版本 `content_hash` 比对，相同则跳过写入
- **is_delete 参与 hash**：当记录的删除状态变化时，hash 会变化，自动产生新版本行

### 9.2 软删除（INSERT 删除版本行）

软删除采用**插入删除版本行**的方式，保持 ODS 追加写入的语义一致性，历史版本行完全不变。

**两条触发路径：**

| 路径 | 触发条件 | 机制 |
|------|---------|------|
| 路径 A（API 返回） | 上游 API 的 JSON 中自带 `is_delete` 字段 | `_normalize_is_delete_flag` 标准化后正常写入；is_delete 参与 hash → 自动产生新版本行 |
| 路径 B（快照空缺） | `snapshot_mode` 为 `FULL_TABLE` 或 `WINDOW`，且 `run.snapshot_missing_delete=True` | `_mark_missing_as_deleted` 对比本次抓取 ID 与数据库已有 ID，为缺失 ID 插入 `is_delete=1` 的新版本行 |

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

### 9.3 下游取数规约

```sql
-- DWD 层从 ODS 取最新有效版本的标准查询
SELECT DISTINCT ON (id) *
FROM ods.{table_name}
WHERE is_delete IS DISTINCT FROM 1
ORDER BY id, fetched_at DESC;
```

此查询利用 `(业务主键, fetched_at DESC)` 复合索引高效执行。
