# dataflow

**Repository Path**: lonlier/dataflow

## Basic Information

- **Project Name**: dataflow
- **Description**: No description available
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-06-05
- **Last Updated**: 2026-06-05

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# DataFlow 使用说明

DataFlow 是一个通用的文件格式数据转换工具，支持 JSON、Excel、Word 等格式之间的数据互相转换。用户通过编写 YAML 配置文件来定义转换规则，无需编写代码。

---

## 目录

- [安装](#安装)
- [快速开始](#快速开始)
- [CLI 命令](#cli-命令)
- [YAML 配置完整参考](#yaml-配置完整参考)
  - [顶层结构](#顶层结构)
  - [数据源配置 (source)](#数据源配置-source)
  - [数据目标配置 (target)](#数据目标配置-target)
  - [字段映射 (mappings)](#字段映射-mappings)
  - [变换 (transform)](#变换-transform)
- [字段映射详解](#字段映射详解)
  - [简单映射](#简单映射)
  - [多字段映射](#多字段映射)
  - [数组展开 (expand)](#数组展开-expand)
  - [多行合并 (merge_rows)](#多行合并-merge_rows)
- [变换详解](#变换详解)
  - [内置变换一览](#内置变换一览)
  - [模板变换 (Jinja2)](#模板变换-jinja2)
  - [自定义 Python 变换](#自定义-python-变换)
- [路径语法参考](#路径语法参考)
  - [JSONPath](#jsonpath)
  - [Excel 单元格引用](#excel-单元格引用)
  - [Word 表格单元格引用](#word-表格单元格引用)
- [完整配置示例](#完整配置示例)
  - [JSON → Excel](#json--excel)
  - [Excel → JSON](#excel--json)
  - [Excel → Word](#excel--word)
  - [Word → Excel](#word--excel)
- [注意事项与常见问题](#注意事项与常见问题)

---

## 安装

**前置条件:** Python 3.10+，已安装 [uv](https://docs.astral.sh/uv/) 包管理器。

```bash
# 克隆项目后进入目录
cd dataflow

# 创建虚拟环境并安装（如果还没有虚拟环境）
uv venv
.venv\Scripts\activate      # Windows
# source .venv/bin/activate  # Linux/Mac

# 安装依赖
uv pip install -e .
```

安装完成后，`dataflow` 命令即可使用。

---

## 快速开始

### 第一步：准备数据文件

创建 `data/orders.json`：

```json
[
  {
    "orderId": "ORD-001",
    "customer": {"name": "张三", "phone": "138xxxx1234"},
    "product": "商品A",
    "price": 99.0
  },
  {
    "orderId": "ORD-002",
    "customer": {"name": "李四", "phone": "139yyyy5678"},
    "product": "商品B",
    "price": 50.0
  }
]
```

### 第二步：编写配置文件

创建 `config.yaml`：

```yaml
task: json_to_excel
name: "订单导入"

source:
  type: json
  path: "data/orders.json"

target:
  type: excel
  path: "output/result.xlsx"
  sheet: "订单"
  start_row: 1

mappings:
  - source: "$.orderId"
    target: "A{row}"

  - source: "$.customer.name"
    target: "B{row}"

  - source: "$.customer.phone"
    target: "C{row}"

  - source: "$.product"
    target: "D{row}"

  - source: "$.price"
    target: "E{row}"
```

### 第三步：执行转换

```bash
dataflow run config.yaml
```

输出文件 `output/result.xlsx` 将包含以下内容：

| A | B | C | D | E |
|---|---|---|---|---|
| ORD-001 | 张三 | 138xxxx1234 | 商品A | 99.0 |
| ORD-002 | 李四 | 139yyyy5678 | 商品B | 50.0 |

### 校验配置

不执行转换，仅校验配置文件是否合法：

```bash
dataflow validate config.yaml
```

---

## CLI 命令

### `dataflow run <config>`

执行转换任务。

```bash
dataflow run config.yaml           # 执行转换
dataflow run config.yaml -v        # 详细日志模式（DEBUG级别）
dataflow run config.yaml --verbose # 同上
```

| 参数 | 类型 | 说明 |
|------|------|------|
| `config` | 必填 | YAML 配置文件路径 |
| `--verbose` / `-v` | 可选 | 开启详细日志 |

### `dataflow validate <config>`

校验配置文件格式和逻辑，不执行转换。输出配置摘要表。

```bash
dataflow validate config.yaml
```

| 参数 | 类型 | 说明 |
|------|------|------|
| `config` | 必填 | YAML 配置文件路径 |

### `dataflow version`

显示版本号。

```bash
dataflow version
# 输出: DataFlow v0.1.0
```

---

## YAML 配置完整参考

### 顶层结构

```yaml
task: json_to_excel          # 必填。任务类型，见下表
name: "我的任务"              # 可选。任务名称
description: "任务描述"       # 可选。任务描述

source: { ... }              # 必填。数据源配置
target: { ... }              # 必填。数据目标配置

vars:                         # 可选。全局变量，在模板中通过 vars.xxx 访问
  level_map:
    "1": "初级"
    "2": "中级"

mappings: [ ... ]             # 字段映射列表

options: {}                   # 可选。保留未来扩展
```

**支持的任务类型：**

| task 值 | 源格式 | 目标格式 |
|---------|--------|----------|
| `json_to_excel` | JSON | Excel |
| `excel_to_json` | Excel | JSON |
| `excel_to_word` | Excel | Word |
| `word_to_excel` | Word | Excel |
| `json_to_json` | JSON | JSON |
| `excel_to_excel` | Excel | Excel |

---

### 数据源配置 (source)

```yaml
source:
  type: json                  # 必填。json | excel | word
  path: "data/**/*.json"      # 必填。文件路径或 glob 模式
  encoding: utf-8             # 可选。文件编码，默认 utf-8

  # --- JSON 专用 ---
  array_path: "$.data.items"  # 可选。JSONPath，指定数据数组在JSON中的位置

  # --- Excel 专用 ---
  sheet: "Sheet1"             # 可选。工作表名，默认第一个表
  range: "A2:F100"            # 可选。读取区域
  header_row: 1               # 可选。表头所在行号

  # --- Word 专用 ---
  table_index: 0              # 可选。读取第几个表格，默认 0（第一个）
```

**关于 `path` 字段：**
- 支持glob模式：`data/**/*.json` 会递归扫描 `data` 下所有子目录中的 `.json` 文件
- 相对路径基于当前工作目录解析

**关于 `array_path`：**
- 省略时，如果JSON是数组则直接使用，如果是对象则包装为单元素列表
- 指定后，从JSON中提取该路径对应的数组作为数据源

---

### 数据目标配置 (target)

```yaml
target:
  type: excel                 # 必填。json | excel | word
  path: "output/result.xlsx" # 必填。输出文件路径

  # --- Excel 专用 ---
  sheet: "Sheet1"             # 可选。工作表名，默认 "Sheet1"
  start_row: 1                # 可选。数据起始行，默认 1
  start_col: 1                # 可选。数据起始列，默认 1（A列）

  # --- JSON 专用 ---
  root_key: "orders"          # 可选。JSON根键名
  structure: array            # 可选。array（默认）或 object
  indent: 2                   # 可选。缩进空格数，默认 2
  ensure_ascii: false         # 可选。是否转义非ASCII字符，默认 false

  # --- Word 专用 ---
  table_index: 0              # 可选。写入第几个表格，默认 0
```

**关于输出文件：**
- 如果目标Excel/Word文件已存在，会在现有文件上修改（保留非数据区域内容）
- 如果目标文件不存在，会自动创建
- 输出目录会自动创建

---

### 字段映射 (mappings)

每个映射项定义一条"从哪取数据 → 做什么变换 → 写到哪"的规则：

```yaml
mappings:
  - source: "$.fieldName"     # 数据源路径（JSONPath）
    target: "A{row}"          # 目标位置（叶子映射必填，中间层可省略）
    transform: { ... }        # 可选。数据变换
    where: { ... }            # 可选。条件过滤
    expand: false             # 可选。是否展开数组
    merge_rows: false         # 可选。是否合并多行
    group_by: null            # merge_rows 的分组字段
    list_fields: null         # merge_rows 的列表字段
    shared_fields: []         # expand 时共享的外层字段
    mappings: []              # 子映射（expand 或 where 钻取时使用）
```

> **关于 `target`：** 只有叶子映射（最终要写出数据的映射）需要写 `target`。当映射仅用于逐层筛选/钻取数据时（使用了 `where` + `mappings`），不需要 `target`。

---

### 变换 (transform)

变换定义如何对提取到的数据进行加工。有三种类型：

```yaml
# 1. 内置变换
transform:
  type: concat
  separator: ", "

# 2. 模板变换（Jinja2）
transform:
  type: template
  template: |
    <b>{{ name }}</b>: {{ value }}

# 3. 自定义 Python 变换
transform:
  type: custom
  module: "./my_transforms.py"
  function: "my_function"
  args:
    flag: true
```

---

## 字段映射详解

### 简单映射

将 JSON/Excel/Word 中的一个字段映射到目标的一个位置：

```yaml
mappings:
  - source: "$.orderId"
    target: "A{row}"
```

`{row}` 是动态行号，会自动递增。也可以写固定行号如 `C3`。

### 多字段映射

当 `source` 是字典时，可以为每个路径指定别名，然后在模板中使用：

```yaml
mappings:
  - source:
      name: "$.customer.name"
      status: "$.status"
    target: "E{row}"
    transform:
      type: template
      template: "<b>{{ name }}</b> - 状态: {{ status }}"
```

---

### 数组展开 (expand)

当 JSON 中有嵌套数组时，使用 `expand` 将每个数组元素展开为一行，外层字段自动复用：

**输入 JSON：**
```json
{
  "orderId": "001",
  "customer": {"name": "张三"},
  "items": [
    {"product": "A", "qty": 2},
    {"product": "B", "qty": 1}
  ]
}
```

**完整配置：**
```yaml
task: json_to_excel
name: "订单展开导入"

source:
  type: json
  path: "data/orders.json"

target:
  type: excel
  path: "output/orders_expanded.xlsx"
  sheet: "订单明细"
  start_row: 1

mappings:
  - source: "$.orderId"
    target: "A{row}"

  - source: "$.customer.name"
    target: "B{row}"

  - source: "$.items"
    target: "C{row}"
    expand: true
    shared_fields:               # 这些外层字段的值会复制到每一展开行
      - "$.orderId"
      - "$.customer.name"
    mappings:                    # 数组内部字段的映射
      - source: "$.product"
        target: "C{row}"
      - source: "$.qty"
        target: "D{row}"
```

**输出 Excel：**

| A (orderId) | B (name) | C (product) | D (qty) |
|---|---|---|---|
| 001 | 张三 | A | 2 |
| 001 | 张三 | B | 1 |

> 外层字段 `orderId` 和 `name` 自动出现在每一行。

---

### 条件过滤 (where)

当源数据是列表时，可以用 `where` 过滤出满足条件的元素，只处理符合条件的项。

`where` 支持**单条件**和**多条件（AND）**两种写法。

**单条件：**

```yaml
- source: "$.productList"
  where:
    field: "attr.type"
    op: "eq"
    value: "规格"
  expand: true
  mappings: ...
```

**多条件（AND关系）：**

```yaml
- source: "$.productList"
  where:
    - field: "attr.type"
      op: "eq"
      value: "规格"
    - field: "attr.value"
      op: "contains"
      value: "200"
  expand: true
  mappings: ...
```

**支持的操作符：**

| op | 说明 | 示例 |
|----|------|------|
| `eq` | 等于 | `value: "规格"` |
| `ne` | 不等于 | `value: "A"` |
| `gt` | 大于 | `value: 100` |
| `lt` | 小于 | `value: 50` |
| `gte` | 大于等于 | `value: 100` |
| `lte` | 小于等于 | `value: 50` |
| `contains` | 包含字符串 | `value: "mg"` |
| `starts_with` | 以...开头 | `value: "SKU"` |
| `ends_with` | 以...结尾 | `value: ".0"` |
| `in` | 在列表中 | `value: ["A", "B"]` |
| `not_empty` | 非空 | 无需 value |
| `is_empty` | 为空 | 无需 value |

**完整示例——从列表中筛选特定类型的元素展开到Excel：**

**输入 JSON (`product.json`)：**
```json
{
  "brand": "优宁维",
  "productList": [
    {"attr": {"goods_sn": "SKU001", "type": "规格", "value": "100mg"}},
    {"attr": {"goods_sn": "SKU002", "type": "包装", "value": "500ml"}},
    {"attr": {"goods_sn": "SKU003", "type": "规格", "value": "200mg"}}
  ]
}
```

**配置：**
```yaml
task: json_to_excel
name: "筛选规格并导入"

source:
  type: json
  path: "product.json"

target:
  type: excel
  path: "output/result.xlsx"
  sheet: "规格"
  start_row: 1

mappings:
  - source: "$.brand"
    target: "A{row}"

  - source: "$.productList"
    target: "C{row}"
    expand: true
    where:
      field: "attr.type"
      op: "eq"
      value: "规格"
    shared_fields:
      - "$.brand"
    mappings:
      - source: "attr.goods_sn"
        target: "C{row}"
      - source: "attr.value"
        target: "D{row}"
```

**输出 Excel（只保留 type=="规格" 的2行）：**

| A (brand) | B | C (goods_sn) | D (value) |
|---|---|---|---|
| 优宁维 | | SKU001 | 100mg |
| 优宁维 | | SKU003 | 200mg |

> `where` 也可以不配合 `expand` 使用。当源数据是列表且不加 `expand` 时，`where` 过滤后的结果：
> - 如果只剩 1 个元素，自动取为单个值
> - 如果剩多个元素，仍为列表

---

### 嵌套钻取 (where + mappings)

当 JSON 有多层嵌套列表，需要通过 `where` 逐层筛选再提取最终值时，可以嵌套使用 `where` + `mappings`（不需要 `target`）作为中间层，只有叶子节点需要 `target`。

**输入 JSON：**
```json
{
  "product": {
    "brand": "优宁维",
    "templateFieldList": [
      {
        "groupName": "产品信息",
        "fieldList": [
          {"fieldSysName": "concentration", "fieldValue": "10mg/ml"},
          {"fieldSysName": "purity", "fieldValue": "99.5%"}
        ]
      },
      {
        "groupName": "包装信息",
        "fieldList": [
          {"fieldSysName": "package_size", "fieldValue": "500ml"}
        ]
      }
    ]
  }
}
```

**需求：** 找到 `groupName == "产品信息"` 的组，再在其中找到 `fieldSysName == "concentration"` 的字段，取其 `fieldValue`。

**配置：**
```yaml
task: json_to_excel
name: "嵌套钻取"

source:
  type: json
  path: "data/product.json"

target:
  type: excel
  path: "output/result.xlsx"
  start_row: 1

mappings:
  - source: "product.brand"
    target: "A{row}"

  # 中间层：筛选 groupName=="产品信息"，无 target，只有 where + mappings
  - source: "product.templateFieldList"
    where:
      - field: "groupName"
        op: "eq"
        value: "产品信息"
    mappings:
      # 第二层中间：筛选 fieldSysName=="concentration"
      - source: "fieldList"
        where:
          - field: "fieldSysName"
            op: "eq"
            value: "concentration"
        mappings:
          # 叶子节点：最终取值，有 target
          - source: "fieldValue"
            target: "F{row}"
```

**数据流解析：**
1. `product.templateFieldList` → 得到一个列表（2个元素）
2. `where: groupName == "产品信息"` → 过滤后只剩第1个元素
3. 进入该元素的 `fieldList` → 得到列表（2个元素）
4. `where: fieldSysName == "concentration"` → 过滤后只剩1个元素
5. 取 `fieldValue` → `"10mg/ml"` → 写入 `F{row}`

> **关键点：** 中间层的 mapping 没有 `target` 字段，只有 `source` + `where` + `mappings`，它们的作用是逐层深入筛选数据。只有最内层的叶子映射才需要 `target`。

---

### 多行合并 (merge_rows)

反向操作：将 Excel 中多行数据合并为 JSON 中的一个列表。

**输入 Excel：**

| B(id) | C(name) | D(phone) |
|---|---|---|
| 001 | 张三 | 138xxxx |
| 001 | 张三 | 139yyyy |

**完整配置：**
```yaml
task: excel_to_json
name: "客户数据合并导出"

source:
  type: excel
  path: "data/clients.xlsx"
  sheet: "客户信息"
  header_row: 1

target:
  type: json
  path: "output/clients.json"
  root_key: "clients"

mappings:
  - source: "col_B"
    target: "$.clients[{i}].id"
  - source: "col_C"
    target: "$.clients[{i}].name"
  - source: "col_D"
    target: "$.clients[{i}].phones"
    merge_rows: true
    group_by: "col_B"
    list_fields: "col_D"
```

**输出 JSON：**
```json
{
  "clients": [
    {"id": "001", "name": "张三", "phones": ["138xxxx", "139yyyy"]}
  ]
}
```

---

## 变换详解

### 内置变换一览

| type | 说明 | 专有参数 | 示例 |
|------|------|----------|------|
| `concat` | 拼接多个值为字符串 | `separator` | `separator: "; "`；`source` 为字典时拼接所有值 |
| `math` | 数学公式计算 | `formula` | `formula: "{0} * 1.13"` |
| `format_date` | 日期格式转换 | `from_fmt`, `to_fmt` | `from_fmt: "%Y-%m-%d"`, `to_fmt: "%Y年%m月%d日"` |
| `default` | 空值替换 | `value` | `value: "未知"` |
| `strip` | 去除首尾空白 | — | — |
| `prefix` | 添加前缀 | `prefix` | `prefix: "序号:"` |
| `suffix` | 添加后缀 | `suffix` | `suffix: "元"` |
| `lowercase` | 转小写 | — | — |
| `uppercase` | 转大写 | — | — |
| `regex_extract` | 正则提取 | `pattern` | `pattern: "\\d+"` |
| `regex_replace` | 正则替换 | `pattern`, `replacement` | `pattern: "\\s+"`, `replacement: " "` |
| `split` | 分割字符串 | `separator` | `separator: ","` |
| `join` | 合并列表为字符串 | `separator` | `separator: "；"` |
| `lookup` | 查表映射 | `table` | `table: {"1": "初级", "2": "中级"}` |
| `constant` | 输出固定值 | `value` | `value: "已导入"` |
| `int` | 转为整数 | — | — |
| `float` | 转为浮点数 | — | — |
| `str` | 转为字符串 | — | — |

**内置变换示例：**

```yaml
# 含税价计算
- source: "$.price"
  target: "E{row}"
  transform:
    type: math
    formula: "{0} * 1.13"

# 空值默认
- source: "$.remark"
  target: "F{row}"
  transform:
    type: default
    value: "无备注"

# 查表映射
- source: "$.level"
  target: "G{row}"
  transform:
    type: lookup
    table:
      "1": "初级"
      "2": "中级"
      "3": "高级"

# 日期格式化
- source: "$.date"
  target: "H{row}"
  transform:
    type: format_date
    from_fmt: "%Y-%m-%d"
    to_fmt: "%Y年%m月%d日"

# 固定值
- target: "I{row}"
  transform:
    type: constant
    value: "已导入"
```

---

### 模板变换 (Jinja2)

模板变换是最强大、最灵活的变换方式，使用 [Jinja2](https://jinja.palletsprojects.com/) 模板语法。

**基础用法：**

```yaml
- source:
    name: "$.customer.name"
    status: "$.status"
  target: "E{row}"
  transform:
    type: template
    template: |
      <b>{{ name }}</b> - 状态: {{ status }}
```

**循环：**

```yaml
- source:
    items: "$.items"
  target: "D{row}"
  transform:
    type: template
    template: |
      {% for item in items %}
      <b>{{ item.productName }}</b>: {{ item.quantity }}件；
      {% endfor %}
```

**条件判断：**

```yaml
- source:
    status: "$.status"
  target: "C{row}"
  transform:
    type: template
    template: |
      {% if status == "urgent" %}
      <color rgb="FF0000"><b>紧急</b></color>
      {% elif status == "normal" %}
      <color rgb="00AA00">正常</color>
      {% else %}
      {{ status }}
      {% endif %}
```

**使用全局变量 (vars)：**

```yaml
vars:
  level_map:
    "1": "初级"
    "2": "中级"
    "3": "高级"

mappings:
  - source:
      level: "$.level"
    target: "F{row}"
    transform:
      type: template
      template: "{{ vars.level_map[level] }}"
```

**模板中可用的变量：**

| 变量 | 说明 |
|------|------|
| `source` 中的别名 | `source` 为字典时各键名即为变量名 |
| `_` | 始终可用，指向 source 数据本身。where 筛选后单个对象也可通过 `_` 访问 |
| `vars` | YAML 配置中 `vars` 定义的全局变量 |
| `row` | 当前写入的目标行号（1起始） |
| `i` | 当前记录索引（0起始） |
| `source_file` | 当前源文件路径 |

**模板中可用的过滤器：**

DataFlow 自定义过滤器（带 `2` 后缀，避免与 Jinja2 内置冲突）：

| 过滤器 | 说明 | 示例 |
|--------|------|------|
| `round2(n)` | 四舍五入到n位小数 | `{{ price \| round2(2) }}` |
| `fmt_date(fmt)` | 日期格式化 | `{{ date \| fmt_date("%Y-%m-%d") }}` |
| `default2(val)` | 空值默认 | `{{ name \| default2("未知") }}` |
| `strip2` | 去除首尾空白 | `{{ val \| strip2 }}` |
| `upper2` | 转大写 | `{{ val \| upper2 }}` |
| `lower2` | 转小写 | `{{ val \| lower2 }}` |
| `replace2(old, new)` | 字符串替换 | `{{ val \| replace2("旧", "新") }}` |
| `regex_replace2(pattern, repl)` | 正则替换 | `{{ val \| regex_replace2("\\s+", " ") }}` |
| `length2` | 长度 | `{{ items \| length2 }}` |
| `first2` | 第一个元素 | `{{ items \| first2 }}` |
| `last2` | 最后一个元素 | `{{ items \| last2 }}` |
| `join2(sep)` | 列表合并 | `{{ items \| join2("；") }}` |
| `sort_by2(key)` | 按字段排序 | `{{ items \| sort_by2("name") }}` |
| `to_json2` | 转为JSON字符串 | `{{ data \| to_json2 }}` |
| `from_json2` | 解析JSON字符串 | `{{ str \| from_json2 }}` |
| `type2` | 返回类型名 | `{{ val \| type2 }}` |

Jinja2 内置过滤器（无需后缀，直接可用）：

| 过滤器 | 说明 | 示例 |
|--------|------|------|
| `replace` | 字符串替换 | `{{ name \| replace(" ", "-") }}` |
| `upper` | 转大写 | `{{ name \| upper }}` |
| `lower` | 转小写 | `{{ name \| lower }}` |
| `strip` | 去首尾空白 | `{{ name \| strip }}` |
| `truncate(n)` | 截断字符串 | `{{ name \| truncate(20) }}` |
| `default(val)` | 空值默认 | `{{ name \| default("N/A") }}` |
| `length` | 长度 | `{{ items \| length }}` |
| `first` | 第一个元素 | `{{ items \| first }}` |
| `last` | 最后一个元素 | `{{ items \| last }}` |
| `join(sep)` | 列表合并 | `{{ items \| join(", ") }}` |
| `sort` | 排序 | `{{ items \| sort }}` |
| `reverse` | 反转 | `{{ items \| reverse }}` |
| `abs` | 绝对值 | `{{ num \| abs }}` |
| `round(n)` | 四舍五入 | `{{ val \| round(2) }}` |
| `int` | 转整数 | `{{ val \| int }}` |
| `float` | 转浮点数 | `{{ val \| float }}` |
| `string` | 转字符串 | `{{ val \| string }}` |
| `trim` | 去首尾空白（同strip） | `{{ name \| trim }}` |
| `capitalize` | 首字母大写 | `{{ name \| capitalize }}` |
| `title` | 每个单词首字母大写 | `{{ name \| title }}` |
| `center(width)` | 居中填充 | `{{ name \| center(20) }}` |
| `ljust(width)` | 左对齐填充 | `{{ name \| ljust(20) }}` |
| `rjust(width)` | 右对齐填充 | `{{ name \| rjust(20) }}` |

**过滤器链式调用：** 多个过滤器可以用 `|` 串联，依次从左到右执行：

```yaml
# 先替换空格为横杠，再转大写
template: "{{ name|replace(' ', '-')|upper }}"

# 先去空白，再替换
template: "{{ name|strip|replace(' ', '-') }}"

# 先默认值，再替换
template: "{{ name|default('未知')|replace(' ', '-') }}"

# 列表先排序，再合并
template: "{% for item in items|sort(attribute='name') %}{{ item }}；{% endfor %}"
```

**模板输出是纯文本字符串。** 如果在模板中使用 HTML 标签（如 `<b>`、`<i>`）或 Markdown 语法（如 `**粗体**`），它们会原样写入目标文件，不渲染为富文本。这样的标签字符串可以被其他软件读取和解析。

**`_` 变量：** 无论 `source` 是何类型，模板中始终可以通过 `_` 访问原始数据。当 `source` 为字典时，各键名也可直接作为变量名使用。

---

### 模板变换实战场景

#### 从多个嵌套列表中取键值对，拼合成一个字符串

这是最常见的复杂需求：JSON 中有多个列表（如分类信息、规格参数等），需要遍历它们，提取键名和值，用 HTML 标签或自定义分隔符拼接后写入 Excel 的一个单元格。

**输入 JSON：**
```json
{
  "product": {
    "brand": "优宁维",
    "category1Name": "化工",
    "category2Name": "试剂",
    "category3Name": "生化试剂",
    "productList": [
      {"attr": {"goods_sn": "SKU001", "value": "100mg"}},
      {"attr": {"goods_sn": "SKU002", "value": "500ml"}}
    ],
    "templateFieldList": [
      {
        "groupName": "产品信息",
        "fieldList": [
          {"fieldSysName": "浓度", "fieldValue": "10mg/ml"},
          {"fieldSysName": "纯度", "fieldValue": "99.5%"},
          {"fieldSysName": "储存条件", "fieldValue": "-20°C"}
        ]
      },
      {
        "groupName": "包装信息",
        "fieldList": [
          {"fieldSysName": "包装规格", "fieldValue": "500ml"},
          {"fieldSysName": "包装类型", "fieldValue": "瓶装"}
        ]
      }
    ]
  }
}
```

**场景1：遍历所有列表的所有键值对，键名加粗、加冒号、加分号**

```yaml
- source:
    fields: "product.templateFieldList"
  target: "B{row}"
  transform:
    type: template
    template: |
      {% for group in fields %}<b>{{ group.groupName }}</b>：{% for f in group.fieldList %}<b>{{ f.fieldSysName }}</b>={{ f.fieldValue }}；{% endfor %}{% endfor %}
```

输出：`<b>产品信息</b>：<b>浓度</b>=10mg/ml；<b>纯度</b>=99.5%；<b>储存条件</b>=-20°C；<b>包装信息</b>：<b>包装规格</b>=500ml；<b>包装类型</b>=瓶装；`

**场景2：只取特定分组的键值对（用 if 条件筛选）**

```yaml
- source:
    fields: "product.templateFieldList"
  target: "C{row}"
  transform:
    type: template
    template: |
      {% for group in fields %}{% if group.groupName == "产品信息" %}{% for f in group.fieldList %}<b>{{ f.fieldSysName }}</b>: {{ f.fieldValue }}；{% endfor %}{% endif %}{% endfor %}
```

输出：`<b>浓度</b>: 10mg/ml；<b>纯度</b>: 99.5%；<b>储存条件</b>: -20°C；`

**场景3：只取值，不要键名**

```yaml
- source:
    fields: "product.templateFieldList"
  target: "D{row}"
  transform:
    type: template
    template: |
      {% for group in fields %}{% for f in group.fieldList %}{{ f.fieldValue }}；{% endfor %}{% endfor %}
```

输出：`10mg/ml；99.5%；-20°C；500ml；瓶装；`

**场景4：用 where 先筛选再模板（`_` 代表筛选后的单个对象）**

```yaml
- source: "product.templateFieldList"
  where:
    - field: "groupName"
      op: "eq"
      value: "包装信息"
  target: "E{row}"
  transform:
    type: template
    template: |
      {% for f in _.fieldList %}<i>{{ f.fieldSysName }}</i>: {{ f.fieldValue }}；{% endfor %}
```

输出：`<i>包装规格</i>: 500ml；<i>包装类型</i>: 瓶装；`

> `where` 筛选后只剩1个元素时，自动解包为单个对象，在模板中用 `_` 引用。

**场景5：多字段用 concat 拼接后作为复用值（放在 expand 前）**

```yaml
mappings:
  - source: "product.brand"
    target: "A{row}"

  # concat 拼接放在 expand 之前 → 每行都会写入
  - source:
      c1: "product.category1Name"
      c2: "product.category2Name"
      c3: "product.category3Name"
    target: "L{row}"
    transform:
      type: concat
      separator: "--"

  - source: "product.productList"
    expand: true
    shared_fields:
      - "product.brand"
    mappings:
      - source: "attr.goods_sn"
        target: "D{row}"
```

输出 L 列：`化工--试剂--生化试剂`（每行都复用）

> **关键点：** `shared_fields` 只支持简单 JSONPath 字段复制。带 `transform` 的复用值，放在 `expand` 块之前即可，框架会自动在每行重新计算并写入。

**场景6：多级 where 钻取 + 模板组合**

从一个嵌套很深的列表中，逐层筛选到目标数据，再用模板格式化：

```yaml
mappings:
  # 第一层：筛选 groupName="产品信息"
  - source: "product.templateFieldList"
    where:
      field: "groupName"
        op: "eq"
        value: "产品信息"
    mappings:
      # 第二层：筛选 fieldSysName="浓度"
      - source: "fieldList"
        where:
          - field: "fieldSysName"
            op: "eq"
            value: "浓度"
        mappings:
          # 叶子：取值并格式化
          - source: "fieldValue"
            target: "F{row}"
            transform:
              type: template
              template: "浓度：{{ _ }}"
```

---

### concat 变换与字典 source

`concat` 变换支持 `source` 为字典的形式，此时会将所有值按顺序拼接：

```yaml
- source:
    c1: "product.category1Id"
    c1name: "product.category1Name"
    c2: "product.category2Id"
    c2name: "product.category2Name"
  target: "L{row}"
  transform:
    type: concat
    separator: "-"
```

源数据 `{"c1": "10", "c1name": "化工", "c2": "101", "c2name": "试剂"}` → 输出：`10-化工-101-试剂`

---

### 自定义 Python 变换

当内置变换和模板都无法满足需求时，可以编写外部 Python 脚本：

**1. 创建 Python 文件 `my_transforms.py`：**

```python
def build_product_summary(items, separator="；"):
    """将产品列表拼接为摘要字符串"""
    parts = []
    for item in items:
        part = f"产品{item.get('name', '未知')}×{item.get('qty', 0)}"
        parts.append(part)
    return separator.join(parts)


def calculate_total(data, rate=1.0):
    """计算含税总价"""
    if isinstance(data, list):
        return sum(item.get("price", 0) * item.get("qty", 1) for item in data) * rate
    return float(data) * rate
```

**2. 在配置中引用：**

```yaml
mappings:
  - source: "$.items"
    target: "D{row}"
    transform:
      type: custom
      module: "./my_transforms.py"      # Python文件路径（相对于工作目录）
      function: "build_product_summary"  # 函数名
      args:                               # 传给函数的关键字参数
        separator: "；"

  - source: "$.items"
    target: "E{row}"
    transform:
      type: custom
      module: "./my_transforms.py"
      function: "calculate_total"
      args:
        rate: 1.13
```

**自定义函数规范：**
- 第一个参数接收 `source` 提取到的数据
- `args` 中的键值对作为关键字参数传入
- 返回值可以是 `str`、`int`、`float`、`list`、`dict` 等类型
- 如果函数不接受额外参数，可省略 `args`

---

## 路径语法参考

### JSONPath

用于 `source` 字段中，从 JSON 数据提取字段值。使用 [jsonpath-ng](https://github.com/h2non/jsonpath-ng) 扩展语法：

| 语法 | 说明 | 示例 |
|------|------|------|
| `$.fieldName` | 根级字段 | `$.orderId` |
| `$.obj.nested` | 嵌套字段 | `$.customer.name` |
| `$.array[*]` | 数组所有元素 | `$.items[*]` |
| `$.array[*].field` | 数组元素的子字段 | `$.items[*].productName` |

> 当 `source` 是字符串时使用 JSONPath；当 `source` 是字典时，值为 JSONPath，键为别名。

### Excel 单元格引用

用于 `target` 字段中，指定写入 Excel 的位置：

| 语法 | 说明 | 示例 |
|------|------|------|
| `A{row}` | 动态行号，自动递增 | `A{row}`, `B{row}` |
| `C3` | 固定单元格 | 写入第3行第C列 |
| `A1:D10` | 区域范围（用于 source.range） | 读取A1到D10区域 |

`{row}` 表示自动递增的行号（从 `start_row` 开始），是最常用的写法。

### Word 表格单元格引用

用于 `target` 字段中，指定写入 Word 表格的位置：

| 语法 | 说明 |
|------|------|
| `table[0].cell[{row}][0]` | 第0个表格的动态行第0列 |
| `table[0].cell[3][1]` | 第0个表格的第3行第1列 |

语法格式：`table[表格索引].cell[行][列]`，索引从 0 开始。

---

## 完整配置示例

### JSON → Excel

**场景：** 批量读取订单JSON文件，导入到Excel订单表。

```yaml
task: json_to_excel
name: "订单数据批量导入"

source:
  type: json
  path: "data/orders/**/*.json"    # 递归扫描子目录
  array_path: "$.data.orders"      # JSONPath提取数据数组

target:
  type: excel
  path: "output/orders.xlsx"
  sheet: "订单明细"
  start_row: 3                      # 从第3行开始写（前2行可能是标题）
  start_col: 2                      # 从B列开始写

vars:
  status_map:
    "pending": "待处理"
    "completed": "已完成"
    "cancelled": "已取消"

mappings:
  # 简单映射
  - source: "$.orderId"
    target: "B{row}"

  - source: "$.customer.name"
    target: "C{row}"

  # 内置变换：含税价
  - source: "$.totalAmount"
    target: "D{row}"
    transform:
      type: math
      formula: "{0} * 1.13"

  # 模板变换：组合字段 + HTML标签
  - source:
      name: "$.customer.name"
      phone: "$.customer.phone"
    target: "E{row}"
    transform:
      type: template
      template: "<b>{{ name }}</b>（{{ phone }}）"

  # 查表变换：状态码转中文
  - source: "$.status"
    target: "F{row}"
    transform:
      type: lookup
      table:
        "pending": "待处理"
        "completed": "已完成"

  # 数组展开
  - source: "$.items"
    target: "G{row}"
    expand: true
    shared_fields:
      - "$.orderId"
      - "$.customer.name"
    mappings:
      - source: "$.productName"
        target: "G{row}"
      - source: "$.quantity"
        target: "H{row}"
      - source: "$.unitPrice"
        target: "I{row}"

  # 固定值
  - target: "J{row}"
    transform:
      type: constant
      value: "已导入"
```

---

### Excel → JSON

**场景：** 读取Excel客户数据，导出为JSON。

```yaml
task: excel_to_json
name: "Excel客户数据导出"

source:
  type: excel
  path: "data/clients/**/*.xlsx"    # 批量读取
  sheet: "客户信息"
  range: "A2:F1000"                 # 指定读取区域
  header_row: 1                     # 第1行是表头

target:
  type: json
  path: "output/clients.json"
  root_key: "clients"
  structure: array
  indent: 2

mappings:
  - source: "col_A"
    target: "$.clients[{i}].id"

  - source: "col_B"
    target: "$.clients[{i}].name"

  # 默认值变换
  - source: "col_C"
    target: "$.clients[{i}].phone"
    transform:
      type: default
      value: "未填写"

  # 模板变换：拼接地址
  - source:
      city: "col_D"
      district: "col_E"
      street: "col_F"
    target: "$.clients[{i}].address"
    transform:
      type: template
      template: "{{ city }}{{ district }}{{ street }}"

  # 多行合并（同ID的行合并电话为列表）
  - source: "col_C"
    target: "$.clients[{i}].phones"
    merge_rows: true
    group_by: "col_A"
    list_fields: "col_C"
```

---

### Excel → Word

**场景：** 读取Excel报表数据，填入Word模板表格。

```yaml
task: excel_to_word
name: "月报生成"

source:
  type: excel
  path: "data/reports/月报.xlsx"
  sheet: "数据"
  range: "A2:E50"

target:
  type: word
  path: "template/report_template.docx"
  table_index: 0                    # 写入Word中第1个表格

mappings:
  - source: "col_A"
    target: "table[0].cell[{row}][0]"

  - source: "col_B"
    target: "table[0].cell[{row}][1]"

  # 多列组合写入Word一个单元格
  - source:
      value: "col_C"
      unit: "col_D"
    target: "table[0].cell[{row}][2]"
    transform:
      type: template
      template: "{{ value }}{{ unit }}"

  - source: "col_E"
    target: "table[0].cell[{row}][3]"
```

---

### Word → Excel

**场景：** 读取Word表格数据，导入到Excel。

```yaml
task: word_to_excel
name: "Word表格数据提取"

source:
  type: word
  path: "data/docs/**/*.docx"
  table_index: 0

target:
  type: excel
  path: "output/extracted.xlsx"
  sheet: "提取数据"
  start_row: 2

mappings:
  - source: "姓名"
    target: "A{row}"

  - source: "年龄"
    target: "B{row}"

  # 前缀
  - source: "部门"
    target: "C{row}"
    transform:
      type: prefix
      prefix: "部门-"

  # 正则提取
  - source: "联系方式"
    target: "D{row}"
    transform:
      type: regex_extract
      pattern: "\\d{11}"
```

---

## 注意事项与常见问题

### 文件路径

- `source.path` 支持 glob 模式（如 `data/**/*.json`），会递归扫描子目录
- `target.path` 如果是相对路径，基于当前工作目录解析
- 输出目录不存在时会自动创建

### JSON 数据结构

- 如果 JSON 顶层是数组，直接使用（每项为一条记录）
- 如果 JSON 顶层是对象，默认包装为单元素列表；用 `array_path` 指定数据在对象内的位置
- 每条记录会自动添加 `_source_file` 字段记录来源文件路径

### Excel 读写

- 目标 Excel 文件如果已存在，会在现有文件上修改（非数据区域内容保留）
- 如果目标文件不存在，自动创建新文件
- `start_row` 和 `start_col` 从 1 开始计数
- 列号使用字母表示（A=1, B=2, ..., Z=26, AA=27, ...）

### Word 读写

- 读取时按 `table_index` 选择表格（0 为第一个）
- Word 文档中的合并单元格可能影响数据提取的准确性
- 写入 Word 时如果目标文件已存在，会在原有文档上修改

### 模板变换输出

- 模板输出为**纯文本字符串**
- HTML 标签（`<b>`、`<i>`等）和 Markdown 语法（`**粗体**`等）会原样写入
- 这些标签可以被上传的目标软件读取和解析

### 调试技巧

1. **先用 `validate` 检查配置**：`dataflow validate config.yaml`
2. **开启详细日志**：`dataflow run config.yaml -v`
3. **简化映射逐步排查**：先测试单条映射，确认后再添加复杂映射
4. **模板调试**：在模板中输出变量类型 `{{ var | type2 }}` 确认数据结构

### 常见错误

| 错误 | 原因 | 解决方法 |
|------|------|----------|
| `Invalid JSONPath expression` | JSONPath语法错误 | 检查路径是否以 `$.` 开头 |
| `Configuration validation error` | YAML格式或字段错误 | 使用 `dataflow validate` 检查 |
| `Missing required field(s): task, source, target` | 配置缺少必填字段 | 确保 `task`、`source`、`target` 都存在 |
| `mappings.X.target Field required` | 叶子映射缺少target | 中间层映射（只有where+mappings）不需要target，但最终取值的映射必须有target |
| `Sheet 'xxx' not found` | Excel工作表名不对 | 确认sheet名称拼写 |
| `Unable to find column letter` | 列号引用格式不对 | 使用大写字母如 `A`, `B`, `AA` |
| `Custom transform module not found` | Python文件路径不对 | 使用相对于工作目录的路径 |
| `Template rendering failed` | 模板中引用了不存在的变量 | 检查source别名是否与模板中变量名一致 |

---

## 打包为 EXE

使用 PyInstaller 将程序打包为独立的 `.exe` 文件，供没有 Python 环境的用户使用。

### 安装 PyInstaller

```bash
uv pip install pyinstaller
```

### 打包命令

```bash
.venv\Scripts\pyinstaller.exe --onefile --name dataflow ^
  --hidden-import=src ^
  --hidden-import=src.transforms.builtins ^
  --hidden-import=jsonpath_ng.ext ^
  src/cli/main.py
```

打包完成后，`dist/dataflow.exe` 即为可分发的独立可执行文件（约 21MB）。

### 分发使用

将 `dataflow.exe` 复制到目标机器，无需安装 Python 或任何依赖，直接在命令行使用：

```
dataflow.exe run config.yaml
dataflow.exe validate config.yaml
dataflow.exe version
```

---

## 项目结构

```
src/
├── __init__.py                # 版本号
├── cli/                       # 命令行界面
│   ├── main.py                # CLI入口
│   └── commands/
│       ├── run.py             # 执行命令
│       └── validate.py        # 校验命令
├── engine/                    # 核心引擎
│   ├── context.py             # 执行上下文
│   ├── runner.py              # 执行引擎
│   └── validator.py           # 配置校验(Pydantic)
├── mappers/                   # 映射处理
│   ├── expander.py            # 数组展开
│   ├── merger.py              # 多行合并
│   ├── mapper.py              # 映射执行器
│   └── path_resolver.py       # 路径解析
├── readers/                   # 数据读取
│   ├── base.py                # 读取器基类
│   ├── json_reader.py         # JSON读取
│   ├── excel_reader.py        # Excel读取
│   └── word_reader.py        # Word读取
├── transforms/                # 数据变换
│   ├── registry.py            # 变换注册表
│   ├── builtins.py            # 内置变换(17种)
│   ├── template_engine.py     # Jinja2模板引擎
│   ├── template_functions.py  # 模板过滤器
│   └── custom_loader.py       # 自定义脚本加载
├── utils/                     # 工具
│   ├── errors.py              # 异常定义
│   ├── file_scanner.py        # 文件扫描
│   └── logger.py              # 日志
└── writers/                   # 数据写入
    ├── base.py                # 写入器基类
    ├── json_writer.py         # JSON写入
    ├── excel_writer.py        # Excel写入
    └── word_writer.py         # Word写入
```