# Transforms

> Typed data pipelines for filtering, normalizing, and partitioning scan output with step.transform.

Recon and scanning produce messy text: duplicate subdomains, mixed case,
blank lines, noise you did not ask for. BountyHub workflows treat data
transformation as a first-class, **typed** pipeline primitive instead of a pile
of `awk | sort | uniq` shell glue.

## The Shape of a Transform

A transform is a **source** plus an ordered chain of operations:

```bhw
step.transform(
    source = path.file("subs.txt").lines(),
    pipeline = transform.trim().lower().unique().sort(),
    destination = path.file("subs-clean.txt"),
)
```

- `source` — how to read the input: `path.file("subs.txt").lines()` parses the
file as newline-delimited lines. Other sources: `.text()`, `.json()`,
`.jsonl()`.
- `pipeline` — the operations applied to each element (or the collection).
- `destination` — where the result is written; must be `path.file(...)`.

The equivalent explicit form passes the source separately:

```bhw
step.transform(
    source = path.file("subs.txt"),
    pipeline = transform.lines().trim().lower().unique().sort(),
    destination = path.file("subs-clean.txt"),
)
```

Both forms compile to the same expression. Use whichever reads better to you.

## Sources

<table>
<thead>
  <tr>
    <th>
      Source
    </th>
    
    <th>
      Parses input as
    </th>
    
    <th>
      Produces
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      <code>
        path.file(f).lines()
      </code>
      
       / <code>
        transform.lines()
      </code>
    </td>
    
    <td>
      Newline-delimited lines
    </td>
    
    <td>
      <code>
        Items<string>
      </code>
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        path.file(f).text()
      </code>
      
       / <code>
        transform.text()
      </code>
    </td>
    
    <td>
      Whole file as one string
    </td>
    
    <td>
      <code>
        string
      </code>
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        path.file(f).json()
      </code>
      
       / <code>
        transform.json()
      </code>
    </td>
    
    <td>
      One JSON document
    </td>
    
    <td>
      <code>
        json
      </code>
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        path.file(f).jsonl()
      </code>
      
       / <code>
        transform.jsonl()
      </code>
    </td>
    
    <td>
      Newline-delimited JSON records
    </td>
    
    <td>
      <code>
        Items<record>
      </code>
    </td>
  </tr>
</tbody>
</table>

`record` values support field access via `.field`, which is how JSONL output
from tools like `httpx -jsonl` becomes structured data. Generic collection ops
like `.unique()` work on any item type:

```bhw
# Deduplicate identical httpx JSONL observations.
step.transform(
    source = path.file("live.jsonl").jsonl(),
    pipeline = transform.unique(),
    destination = path.file("live-dedup.jsonl"),
)
```

A bare `path.file("x")` without a parse is just a file reference — transforms
reject it ("raw file"). Always choose a source first.

## Operations

Operations are chained fluently and are statically type-checked.

### Element-wise string ops

Applied to every item:

<table>
<thead>
  <tr>
    <th>
      Op
    </th>
    
    <th>
      Effect
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      <code>
        .trim()
      </code>
    </td>
    
    <td>
      Strip surrounding whitespace
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        .lower()
      </code>
      
       / <code>
        .upper()
      </code>
    </td>
    
    <td>
      Change case
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        .replace(from, to)
      </code>
    </td>
    
    <td>
      Replace substring
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        .regex(pattern)
      </code>
    </td>
    
    <td>
      Extract regex match (optionally a capture group)
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        .regex_replace(pattern, replacement)
      </code>
    </td>
    
    <td>
      Substitute regex matches
    </td>
  </tr>
</tbody>
</table>

```bhw
# Normalize subdomains and keep only the root domain part.
step.transform(
    source = path.file("subs.txt").lines(),
    pipeline = transform.trim().lower(),
    destination = path.file("subs-clean.txt"),
)

# Extract hosts from raw httpx output lines like "https://target.com [200]".
step.transform(
    source = path.file("raw-httpx.txt").lines(),
    pipeline = transform.regex(r"https?://([^ \]]+)").unique(),
    destination = path.file("hosts.txt"),
)
```

### Collection ops

Applied to the whole stream:

<table>
<thead>
  <tr>
    <th>
      Op
    </th>
    
    <th>
      Effect
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      <code>
        .unique()
      </code>
    </td>
    
    <td>
      Remove duplicates
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        .take(n)
      </code>
      
       / <code>
        .drop(n)
      </code>
    </td>
    
    <td>
      Head/tail slicing
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        .sort()
      </code>
      
       / <code>
        .reverse()
      </code>
    </td>
    
    <td>
      Order the stream
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        .split(sep)
      </code>
    </td>
    
    <td>
      Split each item into multiple items
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        .join(sep)
      </code>
    </td>
    
    <td>
      Join all items into one string
    </td>
  </tr>
</tbody>
</table>

```bhw
# Split comma-separated targets into individual lines.
step.transform(
    source = path.file("targets.csv").lines(),
    pipeline = transform.trim().split(",").trim().lower().unique().sort(),
    destination = path.file("targets-clean.txt"),
)

# Build an nuclei-friendly single-line argument from unique hosts.
step.transform(
    source = path.file("hosts.txt").lines(),
    pipeline = transform.trim().unique().sort().join(","),
    destination = path.file("hosts-arg.txt"),
)
```

### Lambda ops: `map` / `filter` / `flat_map`

The only way to compute per-item values or drop records by predicate. Each
takes an ordinary one-parameter function (lambda or named `def`), captured
deferred and applied at runtime:

```bhw
# Append a port to every host.
step.transform(
    source = path.file("hosts.txt").lines(),
    pipeline = transform.map(lambda host: "{}:8080".format(host)),
    destination = path.file("hosts-8080.txt"),
)

# Keep only high-severity JSONL records, then expand each into its URLs.
step.transform(
    source = path.file("findings.jsonl").jsonl(),
    pipeline = transform.filter(lambda row: row.severity == "high").flat_map(lambda row: row.urls),
    destination = path.file("high-urls.txt"),
)
```

Named functions work too (`transform.map(add_port)` where
`def add_port(host): ...`). Signatures are checked: `map` takes exactly one
parameter.

### Execution model

Ops fall into four execution kinds; knowing them predicts memory behavior:

<table>
<thead>
  <tr>
    <th>
      Kind
    </th>
    
    <th>
      Ops
    </th>
    
    <th>
      Behavior
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      Element-wise (streaming)
    </td>
    
    <td>
      <code>
        trim
      </code>
      
      , <code>
        lower
      </code>
      
      , <code>
        upper
      </code>
      
      , <code>
        replace
      </code>
      
      , <code>
        regex
      </code>
      
      , <code>
        regex_replace
      </code>
      
      , <code>
        map
      </code>
    </td>
    
    <td>
      One item in, zero/one out; constant memory
    </td>
  </tr>
  
  <tr>
    <td>
      Streaming selection/expansion
    </td>
    
    <td>
      <code>
        filter
      </code>
      
      , <code>
        flat_map
      </code>
      
      , <code>
        split
      </code>
    </td>
    
    <td>
      Zero or more out per item; constant memory
    </td>
  </tr>
  
  <tr>
    <td>
      Stateful (streaming)
    </td>
    
    <td>
      <code>
        take(n)
      </code>
      
      , <code>
        drop(n)
      </code>
      
      , <code>
        unique()
      </code>
    </td>
    
    <td>
      Small buffered state; <code>
        unique
      </code>
      
       memory grows with the unique set
    </td>
  </tr>
  
  <tr>
    <td>
      Barrier (buffering)
    </td>
    
    <td>
      <code>
        sort
      </code>
      
      , <code>
        reverse
      </code>
      
      , <code>
        join(sep)
      </code>
    </td>
    
    <td>
      Buffers the whole sequence before emitting
    </td>
  </tr>
</tbody>
</table>

Prefer streaming ops for 50k-host lists; reserve `sort`/`reverse`/`join` for
bounded outputs.

## Types Are Checked Statically

Every value in a transform has a semantic type:

`String`, `Number`, `Bool`, `File`, `Directory`, `Json`, `Record`,
`Items<T>`, `Batch<T>`

Each operation declares which types it accepts, so mistakes fail at save time,
not at 3 AM mid-scan:

```bhw
# ERROR: .lower() needs strings, but json() produces a Json document.
transform.json().lower()

# OK: .unique() works on any collection, including JSONL records.
transform.jsonl().unique()
```

Collections serialize newline-delimited on write, so `Items<string>` writes one
value per line.

## Partitioning

Large datasets can be split into partitions for bounded parallel execution.
Partition strategies come from the `partition` namespace:

- `partition.count(n)` — approximately `n` partitions; sizes differ by at most
one.
- `partition.items(n)` — at most `n` items per partition.

```bhw
# Split 50k live hosts into batches of 500 for parallel nuclei runs.
step.transform(
    source = path.file("live.txt").lines(),
    pipeline = transform.trim().unique().partition(partition.items(500)),
    destination = path.file("batch-{partition_seq}.txt"),
)
```

Rules:

- A partitioned pipeline must use the `{partition_seq}` placeholder in its
destination, so each partition gets its own file (`batch-0.txt`,
`batch-1.txt`, ...).
- Non-partitioned pipelines must **not** contain the placeholder.
- `partition` is terminal — nothing can be chained after it.
- `partition()` takes only a strategy: `partition.count(n)` or
`partition.items(n)`. A bare integer is rejected.
- Execution is deterministic and order-preserving; an empty dataset produces
zero partitions.

Pair partitioning with [`step.for_each`](/workflows/for-each) to process each
batch in parallel. The `for_each` items can be a typed transform expression
directly:

```bhw
activity(
    name = "distributed-nuclei",
    triggers = [trigger.on(probe.succeeded())],
    steps = [
        step.for_each(
            # Unique live hosts, split into ~20 balanced batches.
            items = path.file("live.txt").lines().unique().partition(partition.count(20)),
            do = lambda batch: [
                # Materialize this batch as a file and scan it.
                step.exec(
                    command = ["nuclei", "-l", batch.as_file(), "-severity", "critical", "-o", "findings.txt"],
                ),
            ],
        ),
    ],
)
```

Alternatively, write partitions to files first and fan out over them with
`path.glob(base = "batches", pattern = "scan-*.txt")`.

Records support field access (`row.url`, `row.severity`), method-style string
ops on fields (`row.host.lower()`), and index access (`row.urls[0]`) inside
`map`/`filter` lambdas and dataset mappings.