textpipe.py

#!/usr/bin/env python3
"""
textpipe -- Chain text operations from the command line. Cross-platform grep+awk+sort+uniq.

No need for grep | awk | sort | uniq | head pipelines. One tool, ordered operations.
Works on Windows, macOS, Linux. Zero dependencies.

Usage:
    py textpipe.py input.txt grep "ERROR"                          # grep for ERROR
    py textpipe.py input.txt grep "ERROR" count                    # count ERROR lines
    py textpipe.py access.log fields 7 freq head 10                # top 10 URLs
    py textpipe.py data.txt replace "foo" "bar" upper              # replace then uppercase
    py textpipe.py log.txt grep "4\\d\\d" extract "\\d{3}" freq     # HTTP error code frequency
    py textpipe.py names.txt sort unique number                    # sort, dedup, number
    py textpipe.py data.csv delim "," fields 2,3 trim              # extract CSV columns
    cat file.txt | py textpipe.py grep "TODO" upper                # stdin piping
    echo "hello world" | py textpipe.py upper                      # quick transform

Operations (applied LEFT to RIGHT):
    grep PATTERN      Keep lines matching regex
    grepv PATTERN     Keep lines NOT matching regex
    replace OLD NEW   Regex search-and-replace
    extract PATTERN   Extract regex matches (group 1 if captured, else full match)
    fields N[,M,...]  Select fields (1-indexed). e.g. fields 1,3,5
    delim CHAR        Set field delimiter (default: whitespace split)
    unique            Remove duplicate lines (preserves order)
    sort              Sort lines alphabetically
    sortnum           Sort lines numerically (by first number found)
    reverse           Reverse line order
    head N            Keep first N lines
    tail N            Keep last N lines
    count             Print line count
    freq              Frequency table (most common first)
    upper             Uppercase all lines
    lower             Lowercase all lines
    trim              Strip leading/trailing whitespace
    squeeze           Collapse consecutive blank lines to one
    number            Add line numbers
    length            Show each line's character length
    prepend TEXT      Add text before each line
    append TEXT       Add text after each line
    between S E       Keep lines between patterns S and E (inclusive)
    sum               Sum all numbers found in lines
    tally             Count lines (alias for count)
    join SEP          Join all lines into one with separator
    split CHAR        Split each line on CHAR into multiple lines
"""

import re
import sys
from collections import OrderedDict


def op_grep(lines: list[str], pattern: str) -> list[str]:
    rx = re.compile(pattern, re.IGNORECASE)
    return [l for l in lines if rx.search(l)]


def op_grepv(lines: list[str], pattern: str) -> list[str]:
    rx = re.compile(pattern, re.IGNORECASE)
    return [l for l in lines if not rx.search(l)]


def op_replace(lines: list[str], old: str, new: str) -> list[str]:
    rx = re.compile(old)
    return [rx.sub(new, l) for l in lines]


def op_extract(lines: list[str], pattern: str) -> list[str]:
    rx = re.compile(pattern)
    results = []
    for line in lines:
        for m in rx.finditer(line):
            if m.groups():
                results.append(m.group(1))
            else:
                results.append(m.group(0))
    return results


def op_fields(lines: list[str], spec: str, delimiter: str | None) -> list[str]:
    indices = []
    for part in spec.split(","):
        part = part.strip()
        if "-" in part:
            a, b = part.split("-", 1)
            a = int(a) if a else 1
            b = int(b) if b else 999
            indices.extend(range(a, b + 1))
        else:
            indices.append(int(part))

    result = []
    for line in lines:
        if delimiter:
            parts = line.split(delimiter)
        else:
            parts = line.split()
        selected = []
        for i in indices:
            if 1 <= i <= len(parts):
                selected.append(parts[i - 1])
        result.append(" ".join(selected) if not delimiter else delimiter.join(selected))
    return result


def op_unique(lines: list[str]) -> list[str]:
    seen = OrderedDict()
    for l in lines:
        seen[l] = None
    return list(seen.keys())


def op_sort(lines: list[str]) -> list[str]:
    return sorted(lines)


def op_sortnum(lines: list[str]) -> list[str]:
    def num_key(line):
        m = re.search(r"-?\d+\.?\d*", line)
        return float(m.group()) if m else 0
    return sorted(lines, key=num_key)


def op_reverse(lines: list[str]) -> list[str]:
    return list(reversed(lines))


def op_head(lines: list[str], n: int) -> list[str]:
    return lines[:n]


def op_tail(lines: list[str], n: int) -> list[str]:
    return lines[-n:] if n > 0 else []


def op_count(lines: list[str]) -> list[str]:
    return [str(len(lines))]


def op_freq(lines: list[str]) -> list[str]:
    counts: dict[str, int] = {}
    for l in lines:
        counts[l] = counts.get(l, 0) + 1
    ranked = sorted(counts.items(), key=lambda x: -x[1])
    return [f"{c:>6}  {val}" for val, c in ranked]


def op_upper(lines: list[str]) -> list[str]:
    return [l.upper() for l in lines]


def op_lower(lines: list[str]) -> list[str]:
    return [l.lower() for l in lines]


def op_trim(lines: list[str]) -> list[str]:
    return [l.strip() for l in lines]


def op_squeeze(lines: list[str]) -> list[str]:
    result = []
    prev_blank = False
    for l in lines:
        is_blank = l.strip() == ""
        if is_blank and prev_blank:
            continue
        result.append(l)
        prev_blank = is_blank
    return result


def op_number(lines: list[str]) -> list[str]:
    w = len(str(len(lines)))
    return [f"{i+1:>{w}}  {l}" for i, l in enumerate(lines)]


def op_length(lines: list[str]) -> list[str]:
    return [f"{len(l):>5}  {l}" for l in lines]


def op_prepend(lines: list[str], text: str) -> list[str]:
    return [text + l for l in lines]


def op_append(lines: list[str], text: str) -> list[str]:
    return [l + text for l in lines]


def op_between(lines: list[str], start: str, end: str) -> list[str]:
    rx_start = re.compile(start)
    rx_end = re.compile(end)
    result = []
    inside = False
    for l in lines:
        if not inside and rx_start.search(l):
            inside = True
        if inside:
            result.append(l)
        if inside and rx_end.search(l):
            inside = False
    return result


def op_sum(lines: list[str]) -> list[str]:
    total = 0.0
    for l in lines:
        for m in re.finditer(r"-?\d+\.?\d*", l):
            total += float(m.group())
    if total == int(total):
        return [str(int(total))]
    return [f"{total:.4f}"]


def op_join(lines: list[str], sep: str) -> list[str]:
    return [sep.join(lines)]


def op_split(lines: list[str], char: str) -> list[str]:
    result = []
    for l in lines:
        result.extend(l.split(char))
    return result


def parse_and_run(args: list[str]):
    """Parse args, read input, apply operations in order."""
    if not args or args[0] in ("-h", "--help"):
        print(__doc__.strip())
        return

    # Determine input source
    idx = 0
    input_file = None

    # Check if first arg is a file or an operation
    if args[0] not in get_op_names() and args[0] != "-":
        input_file = args[0]
        idx = 1

    # Read input
    if input_file and input_file != "-":
        try:
            with open(input_file, "r", encoding="utf-8", errors="replace") as f:
                lines = [l.rstrip("\n\r") for l in f]
        except FileNotFoundError:
            print(f"Error: file not found: {input_file}", file=sys.stderr)
            sys.exit(1)
        except Exception as e:
            print(f"Error reading {input_file}: {e}", file=sys.stderr)
            sys.exit(1)
    else:
        if sys.stdin.isatty() and not input_file:
            # No file, no pipe -- check if first arg might be a file
            print("Error: no input. Provide a file or pipe stdin.", file=sys.stderr)
            print("Usage: py textpipe.py <file> [operations...]", file=sys.stderr)
            sys.exit(1)
        lines = [l.rstrip("\n\r") for l in sys.stdin]

    # Parse and apply operations in order
    delimiter = None  # for fields op

    while idx < len(args):
        op = args[idx].lower()
        idx += 1

        if op == "grep":
            if idx >= len(args):
                print("Error: grep requires a pattern", file=sys.stderr); sys.exit(1)
            lines = op_grep(lines, args[idx]); idx += 1
        elif op == "grepv":
            if idx >= len(args):
                print("Error: grepv requires a pattern", file=sys.stderr); sys.exit(1)
            lines = op_grepv(lines, args[idx]); idx += 1
        elif op == "replace":
            if idx + 1 >= len(args):
                print("Error: replace requires OLD NEW", file=sys.stderr); sys.exit(1)
            lines = op_replace(lines, args[idx], args[idx + 1]); idx += 2
        elif op == "extract":
            if idx >= len(args):
                print("Error: extract requires a pattern", file=sys.stderr); sys.exit(1)
            lines = op_extract(lines, args[idx]); idx += 1
        elif op == "fields":
            if idx >= len(args):
                print("Error: fields requires column spec (e.g. 1,3,5)", file=sys.stderr); sys.exit(1)
            lines = op_fields(lines, args[idx], delimiter); idx += 1
        elif op == "delim":
            if idx >= len(args):
                print("Error: delim requires a character", file=sys.stderr); sys.exit(1)
            delimiter = args[idx]; idx += 1
        elif op in ("unique", "uniq"):
            lines = op_unique(lines)
        elif op == "sort":
            lines = op_sort(lines)
        elif op == "sortnum":
            lines = op_sortnum(lines)
        elif op in ("reverse", "rev"):
            lines = op_reverse(lines)
        elif op == "head":
            if idx >= len(args):
                print("Error: head requires a number", file=sys.stderr); sys.exit(1)
            lines = op_head(lines, int(args[idx])); idx += 1
        elif op == "tail":
            if idx >= len(args):
                print("Error: tail requires a number", file=sys.stderr); sys.exit(1)
            lines = op_tail(lines, int(args[idx])); idx += 1
        elif op in ("count", "tally"):
            lines = op_count(lines)
        elif op == "freq":
            lines = op_freq(lines)
        elif op == "upper":
            lines = op_upper(lines)
        elif op == "lower":
            lines = op_lower(lines)
        elif op == "trim":
            lines = op_trim(lines)
        elif op == "squeeze":
            lines = op_squeeze(lines)
        elif op in ("number", "nl"):
            lines = op_number(lines)
        elif op in ("length", "len"):
            lines = op_length(lines)
        elif op == "prepend":
            if idx >= len(args):
                print("Error: prepend requires text", file=sys.stderr); sys.exit(1)
            lines = op_prepend(lines, args[idx]); idx += 1
        elif op == "append":
            if idx >= len(args):
                print("Error: append requires text", file=sys.stderr); sys.exit(1)
            lines = op_append(lines, args[idx]); idx += 1
        elif op == "between":
            if idx + 1 >= len(args):
                print("Error: between requires START END patterns", file=sys.stderr); sys.exit(1)
            lines = op_between(lines, args[idx], args[idx + 1]); idx += 2
        elif op == "sum":
            lines = op_sum(lines)
        elif op == "join":
            if idx >= len(args):
                print("Error: join requires a separator", file=sys.stderr); sys.exit(1)
            lines = op_join(lines, args[idx]); idx += 1
        elif op == "split":
            if idx >= len(args):
                print("Error: split requires a character", file=sys.stderr); sys.exit(1)
            lines = op_split(lines, args[idx]); idx += 1
        else:
            print(f"Error: unknown operation '{op}'", file=sys.stderr)
            print(f"Operations: {', '.join(get_op_names())}", file=sys.stderr)
            sys.exit(1)

    # Output
    for line in lines:
        print(line)


def get_op_names() -> set[str]:
    return {
        "grep", "grepv", "replace", "extract", "fields", "delim",
        "unique", "uniq", "sort", "sortnum", "reverse", "rev",
        "head", "tail", "count", "tally", "freq",
        "upper", "lower", "trim", "squeeze",
        "number", "nl", "length", "len",
        "prepend", "append", "between", "sum", "join", "split",
    }


def main():
    parse_and_run(sys.argv[1:])


if __name__ == "__main__":
    main()