static USAGE: &str = r#"
Apply a series of transformation functions to given CSV column/s. This can be used to
perform typical data-wrangling tasks and/or to harmonize some values, etc.

It has five subcommands:
 1. operations*   - 40 string, format, currency, regex & NLP operators.
 2. emptyreplace* - replace empty cells with <--replacement> string.
 3. dynfmt        - Dynamically constructs a new column from other columns using
                    the <--formatstr> template.
 4. calcconv      - parse and evaluate math expressions, with support for units
                    and conversions.
 5. summarize*    - summarize a column or group of columns using an OpenAI
                    API-compatible LLM (local or commercial), with customizable,
                    MiniJinja-templated per-record prompts.
    * subcommand is multi-column capable.

OPERATIONS (multi-column capable)
Multiple operations can be applied, with the comma-delimited operation series
applied in order:

  trim => Trim the cell
  trim,upper => Trim the cell, then transform to uppercase
  lower,simdln => Lowercase the cell, then compute the normalized
      Damerau-Levenshtein similarity to --comparand

Operations support multi-column transformations. Just make sure the
number of transformed columns with the --rename option is the same.
For example, to trim and fold to uppercase the col1,col2 and col3 columns &
rename them to newcol1,newcol2 and newcol3:

  qsv apply operations trim,upper col1,col2,col3 -r newcol1,newcol2,newcol3 file.csv

It has 40 supported operations:

  * len: Return string length
  * lower: Transform to lowercase
  * upper: Transform to uppercase
  * squeeze: Compress consecutive whitespaces
  * squeeze0: Remove whitespace
  * trim: Trim (drop whitespace left & right of the string)
  * ltrim: Left trim whitespace
  * rtrim: Right trim whitespace
  * mtrim: Trims --comparand matches left & right of the string (Rust trim_matches)
  * mltrim: Left trim --comparand matches (Rust trim_start_matches)
  * mrtrim: Right trim --comparand matches (Rust trim_end_matches)
  * strip_prefix: Removes specified prefix in --comparand
  * strip_suffix: Remove specified suffix in --comparand
  * escape - escape (Rust escape_default)
  * encode62: base62 encode
  * decode62: base62 decode
  * encode64: base64 encode
  * decode64: base64 decode
  * crc32: crc32 checksum
  * replace: Replace all matches of a pattern (using --comparand)
      with a string (using --replacement) (Rust replace)
  * regex_replace: Replace all regex matches in --comparand w/ --replacement.
      Specify <NULL> as --replacement to remove matches.
  * titlecase - capitalizes English text using Daring Fireball titlecase style
      https://daringfireball.net/2008/05/title_case
  * censor: profanity filter. Add additional comma-delimited profanities with --comparand.
  * censor_check: check if profanity is detected (boolean).
      Add additional comma-delimited profanities with -comparand.
  * censor_count: count of profanities detected.
      Add additional comma-delimited profanities with -comparand.
  * round: Round numeric values to the specified number of decimal places using
      Midpoint Nearest Even Rounding Strategy AKA "Bankers Rounding."
      Specify the number of decimal places with --formatstr (default: 3).
  * thousands: Add thousands separators to numeric values.
      Specify the separator policy with --formatstr (default: comma). The valid policies are:
      comma, dot, space, underscore, hexfour (place a space every four hex digits) and
      indiancomma (place a comma every two digits, except the last three digits).
      The decimal separator can be specified with --replacement (default: '.')
  * currencytonum: Gets the numeric value of a currency. Supports currency symbols
      (e.g. $,¥,£,€,֏,₱,₽,₪,₩,ƒ,฿,₫) and strings (e.g. USD, EUR, RMB, JPY, etc.).
      Recognizes point, comma and space separators. Is "permissive" by default, meaning it
      will allow no or non-ISO currency symbols. To enforce strict parsing, which will require
      a valid ISO currency symbol, set the --formatstr to "strict".
  * numtocurrency: Convert a numeric value to a currency. Specify the currency symbol
      with --comparand. Automatically rounds values to two decimal places. Specify
      "euro" formatting (e.g. 1.000,00 instead of 1,000.00 ) by setting --formatstr to "euro".
      Specify conversion rate by setting --replacement to a number.
  * gender_guess: Guess the gender of a name.
  * copy: Mark a column for copying
  * simdl: Damerau-Levenshtein similarity to --comparand
  * simdln: Normalized Damerau-Levenshtein similarity to --comparand (between 0.0 & 1.0)
  * simjw: Jaro-Winkler similarity to --comparand (between 0.0 & 1.0)
  * simsd: Sørensen-Dice similarity to --comparand (between 0.0 & 1.0)
  * simhm: Hamming distance to --comparand. Num of positions characters differ.
  * simod: Optimal String Alignment (OSA) Distance to --comparand.
  * eudex: Multi-lingual sounds like --comparand (boolean)
       Tested on English, Catalan, German, Spanish, Swedish and Italian dictionaries.
       It supports all C1 letters (e.g. ü, ö, æ, ß, é, etc.) and takes their sound into account.
       It should work on other European languages that use the Latin alphabet.
  * sentiment: Normalized VADER sentiment score (English only - between -1.0 to 1.0).
  * whatlang: Language Detection for 87 supported languages, with default confidence threshold
       of 0.9, which can be overridden by assigning 0.0 to 1.0 to --comparand.
       If language detection confidence is below the threshold, it will still show the best language
       guess, followed by the confidence score, ending with a question mark.
       If you want to always displays the confidence score, end the --comparand value with a
       question mark (e.g. 0.9?)
       https://github.com/greyblake/whatlang-rs/blob/master/SUPPORTED_LANGUAGES.md


EMPTYREPLACE (multi-column capable)
Replace empty cells with <--replacement> string.
Non-empty cells are not modified. See the `fill` command for more complex empty field operations.


DYNFMT
Dynamically constructs a new column from other columns using the <--formatstr> template.
The template can contain arbitrary characters. To insert a column value, enclose the
column name in curly braces, replacing all non-alphanumeric characters with underscores.

If you need to dynamically construct a column with more complex formatting requirements and
computed values, check out the py command to take advantage of Python's f-string formatting.


CALCCONV
Parse and evaluate math expressions into a new column, with support for units and conversions.
The math expression is built dynamically using the <--formatstr> template, similar to the DYNFMT
subcommand, with the addition that if the literal '<UNIT>' is found at the end of the template, the
inferred unit will be appended to the result.

For a complete list of supported units, constants, operators and functions, see https://docs.rs/cpc

Examples:

== OPERATIONS ==

  # Trim, then transform to uppercase the surname field.
  qsv apply operations trim,upper surname file.csv

  # Trim, then transform to uppercase the surname field and rename the column uppercase_clean_surname.
  qsv apply operations trim,upper surname -r uppercase_clean_surname file.csv

  # Trim, then transform to uppercase the surname field and
  # save it to a new column named uppercase_clean_surname.
  qsv apply operations trim,upper surname -c uppercase_clean_surname file.csv

  # Trim, then transform to uppercase the firstname and surname fields and
  # rename the columns ufirstname and usurname.
  qsv apply operations trim,upper firstname,surname -r ufirstname,usurname file.csv

  # Trim parentheses & brackets from the description field.
  qsv apply operations mtrim description --comparand '()<>' file.csv

  # Replace ' and ' with ' & ' in the description field.
  qsv apply operations replace description --comparand ' and ' --replacement ' & ' file.csv

  # Extract the numeric value of the Salary column in a new column named Salary_num.
  qsv apply operations currencytonum Salary -c Salary_num file.csv

  # Convert the USD_Price to PHP_Price using the currency symbol "PHP" with a conversion rate of 60.
  qsv apply operations numtocurrency USD_Price -C PHP -R 60 -c PHP_Price file.csv

  # Base64 encode the text_col column & save the encoded value into new column named encoded & decode it.
  qsv apply operations encode64 text_col -c encoded file.csv | qsv apply operations decode64 encoded

  # Compute the Normalized Damerau-Levenshtein similarity of the neighborhood column to
  # the string 'Roxbury' and save it to a new column named dln_roxbury_score.
  qsv apply operations lower,simdln neighborhood --comparand roxbury -c dln_roxbury_score boston311.csv

  # You can also use this subcommand command to make a copy of a column:
  qsv apply operations copy col_to_copy -c col_copy file.csv

== EMPTYREPLACE ==

  # Replace empty cells in file.csv Measurement column with 'None'.
  qsv apply emptyreplace Measurement --replacement None file.csv

  # Replace empty cells in file.csv Measurement column with 'Unknown Measurement'.
  qsv apply emptyreplace Measurement --replacement 'Unknown Measurement' file.csv

  # Replace empty cells in file.csv M1,M2 and M3 columns with 'None'.
  qsv apply emptyreplace M1,M2,M3 --replacement None file.csv

  # Replace all empty cells in file.csv for columns that start with 'Measurement' with 'None'.
  qsv apply emptyreplace '/^Measurement/' --replacement None file.csv

  # Replace all empty cells in file.csv for columns that start with 'observation'
  # case insensitive with 'None'.
  qsv apply emptyreplace --replacement None '/(?i)^observation/' file.csv

== DYNFMT ==

  # Create a new column 'mailing address' from 'house number', 'street', 'city'
  # and 'zip-code' columns:
  qsv apply dynfmt --formatstr '{house_number} {street}, {city} {zip_code} USA' -c 'mailing address' file.csv

  # Create a new column 'FullName' from 'FirstName', 'MI', and 'LastName' columns:
  qsv apply dynfmt --formatstr 'Sir/Madam {FirstName} {MI}. {LastName}' -c FullName file.csv

== CALCCONV ==

  # Do simple arithmetic:
  qsv apply calcconv --formatstr '{col1} + {col2} * {col3}' --new-column result file.csv

  # Arithmetic with support for operators like % and ^:
  qsv apply calcconv --formatstr '{col1} % 3' --new-column remainder file.csv

  # Convert from one unit to another:
  qsv apply calcconv --formatstr '{col1} Fahrenheit in Celsius' -c metric_temperature file.csv

  # Mix units and conversions are automatically done for you:
  qsv apply calcconv --formatstr '{col1}km + {col2}mi in meters' -c meters file.csv

  # You can append the inferred unit at the end of the result by ending the expression with '<UNIT>':
  qsv apply calcconv --formatstr '({col1} + {col2})km to light years <UNIT>' -c light_years file.csv

  # You can even do complex temporal unit conversions:
  qsv apply calcconv --formatstr '{col1}m/s + {col2}mi/h in kilometers per h' -c kms_per_h file.csv

  # Use math functions - see https://docs.rs/cpc/latest/cpc/enum.FunctionIdentifier.html for list of functions:
  qsv apply calcconv --formatstr 'round(sqrt{col1}^4)! liters' -c liters file.csv

  # Use percentages:
  qsv apply calcconv --formatstr '10% of abs(sin(pi)) horsepower to watts' -c watts file.csv

  # Use very large numbers:
  qsv apply calcconv --formatstr '{col1} Billion Trillion * {col2} quadrillion vigintillion' -c num_atoms file.csv

== SUMMARIZE ==

The summarize subcommand sends each record to an OpenAI API-compatible LLM and stores the
returned summary in a new column. The prompt is a MiniJinja template (like `qsv template`)
rendered per record - reference any column by its "safe" name (non-alphanumeric chars become
'_'); the 1-based row number is available as {{QSV_ROWNO}}.

If no --prompt/--prompt-file is given, a default prompt that summarizes the selected column/s
is used. The LLM endpoint, model & API key are resolved with this precedence:
  CLI flag > env var (QSV_LLM_BASE_URL / QSV_LLM_MODEL / QSV_LLM_APIKEY) > built-in default
(default base-url: http://localhost:1234/v1 ; default model: openai/gpt-oss-20b).

An API key is required when the base URL is not a localhost endpoint (set --api-key to "NONE"
to suppress sending a key). One HTTP call is made per uncached row, so summarizing large files
can be slow & costly. Results are cached on disk (keyed by base-url, model, max-tokens,
addl-props & the rendered prompt) so repeated rows & re-runs don't re-pay for inference;
use --no-cache to disable & --fresh to force fresh calls that refresh the cache.

  # Summarize the support_ticket column into a new "summary" column using a local LLM:
  qsv apply summarize support_ticket -c summary file.csv

  # Use a custom prompt referencing multiple columns against a cloud provider:
  qsv apply summarize subject,body -c summary --prompt 'Summarize: {{subject}} - {{body}}' -u https://api.openai.com/v1 -m gpt-4o-mini -k $OPENAI_API_KEY file.csv

  # Also capture per-row inference time & token usage in extra columns:
  qsv apply summarize notes -c summary --stats file.csv

For more extensive examples, see https://github.com/dathere/qsv/blob/master/tests/test_apply.rs.
See also https://github.com/dathere/qsv/wiki/Transform-and-Reshape#apply

Usage:
qsv apply operations <operations> [options] <column> [<input>]
qsv apply emptyreplace --replacement=<string> [options] <column> [<input>]
qsv apply dynfmt --formatstr=<string> [options] --new-column=<name> [<input>]
qsv apply calcconv --formatstr=<string> [options] --new-column=<name> [<input>]
qsv apply summarize [options] --new-column=<name> <column> [<input>]
qsv apply --help

apply arguments:
    <column>                        The column/s to apply the transformation to.
                                    Note that the <column> argument supports multiple columns
                                    for the operations & emptyreplace subcommands.
                                    See 'qsv select --help' for the format details.

    OPERATIONS subcommand:
        <operations>                The operation/s to apply.
        <column>                    The column/s to apply the operations to.

    EMPTYREPLACE subcommand:
        --replacement=<string>      The string to use to replace empty values.
        <column>                    The column/s to check for emptiness.

    DYNFMT subcommand:
        --formatstr=<string>        The template to use for the dynfmt operation.
                                    See DYNFMT example above for more details.
        --new-column=<name>         Put the generated values in a new column.

    CALCCONV subcommand:
        --formatstr=<string>        The calculation/conversion expression to use.
        --new-column=<name>         Put the calculated/converted values in a new column.

    SUMMARIZE subcommand:
        <column>                    The column/s whose values are summarized by the LLM.
                                    Used to build the default prompt. With a custom prompt
                                    (via the --prompt/--prompt-file options), any column
                                    can be referenced.
        --new-column=<name>         Put the generated summaries in a new column (required).

    <input>                     The input file to read from. If not specified, reads from stdin.

apply options:
    -c, --new-column <name>     Put the transformed values in a new column instead.
    -r, --rename <name>         New name for the transformed column.
    -C, --comparand=<string>    The string to compare against for replace & similarity operations.
                                Also used with numtocurrency operation to specify currency symbol.
    -R, --replacement=<string>  The string to use for the replace & emptyreplace operations.
                                Also used with numtocurrency operation to conversion rate.
    -f, --formatstr=<string>    This option is used by several subcommands:

                                OPERATIONS:
                                  currencytonum
                                    If set to "strict", will require a valid ISO currency symbol,
                                    with the currency symbol at the beginning of the string.
                                    Otherwise, only parse the numeric part of the string and ignore
                                    the currency symbol altogether.
                                    (default: permissive)

                                  numtocurrency
                                    If set to "euro", will format the currency to use "." instead of ","
                                    as separators (e.g. 1.000,00 instead of 1,000.00 )

                                  thousands
                                    The thousands separator policy to use. The available policies are:
                                    comma, dot, space, underscore, hexfour (place a space every four
                                    hex digits) and indiancomma (place a comma every two digits,
                                    except the last three digits). (default: comma)

                                  round
                                    The number of decimal places to round to (default: 3)

                                DYNFMT: the template to use to construct a new column.

    -j, --jobs <arg>            The number of jobs to run in parallel.
                                When not set, the number of jobs is set to the number of CPUs detected.
    -b, --batch <size>          The number of rows per batch to load into memory, before running in parallel.
                                Automatically determined for CSV files with more than 50000 rows.
                                Set to 0 to load all rows in one batch. Set to 1 to force batch optimization
                                even for files with less than 50000 rows.
                                [default: 50000]

SUMMARIZE options:
    -u, --base-url <url>        Base URL of the OpenAI API-compatible endpoint.
                                Precedence: this flag > QSV_LLM_BASE_URL env var > http://localhost:1234/v1
    -m, --model <model>         Model name compatible with the OpenAI API spec.
                                Precedence: this flag > QSV_LLM_MODEL env var > openai/gpt-oss-20b
    -k, --api-key <key>         API key for Bearer token authentication.
                                Precedence: this flag > QSV_LLM_APIKEY env var.
                                Set to "NONE" to suppress sending a key. Required for non-localhost URLs.
    -t, --max-tokens <n>        Maximum number of tokens in the LLM output. Set to 0 to not send a
                                max_tokens limit (automatically used for localhost endpoints).
                                [default: 10000]
    --timeout <secs>            Timeout for each LLM request in seconds (0 = no timeout).
                                [default: 300]
    --addl-props <json>         Additional model properties as a JSON object, e.g.
                                '{"reasoning_effort": "high", "temperature": 0.2}'
    --prompt <template>         MiniJinja prompt template rendered per record. Overrides the
                                default prompt. Reference columns by their safe name.
    --prompt-file <file>        Read the prompt template from a file. Ignored if --prompt is set.
    --rate-limit <secs>         Seconds to sleep between LLM requests to avoid provider rate limits.
                                Accepts fractional seconds (e.g. 0.5). [default: 0]
    --on-error <mode>           What to do when an LLM request fails: "fail" aborts; "skip" writes
                                an "<ERROR: ...>" cell and continues. [default: fail]
    --user-agent <agent>        Custom user agent for LLM requests. Supports variables like $QSV_VERSION.
    --cache-dir <dir>           Directory for the disk cache. [default: ~/.qsv-cache/apply-summarize]
                                The cache TTL defaults to 28 days; set the QSV_DISKCACHE_TTL_SECS env
                                var to change it. A value of 0 disables time-based expiration
                                (entries are cached indefinitely). Use --no-cache to disable caching.
    --no-cache                  Disable the disk cache (one LLM call per row, always).
    --fresh                     Force fresh LLM calls, refreshing any cached values.
    --stats                     Append two extra columns per row alongside --new-column:
                                "<new-column>_elapsed_ms" (inference time in milliseconds) and
                                "<new-column>_tokens" (total tokens used). For cached rows, these
                                report the values captured when the summary was first computed.

Common options:
    -h, --help                  Display this message
    -o, --output <file>         Write output to <file> instead of stdout.
    -n, --no-headers            When set, the first row will not be interpreted
                                as headers.
    -d, --delimiter <arg>       The field delimiter for reading CSV data.
                                Must be a single character. (default: ,)
    -p, --progressbar           Show progress bars. Not valid for stdin.
"#;

use std::{collections::BTreeMap, str::FromStr, sync::OnceLock, time::Duration};

use base64_simd::STANDARD as BASE64;
use cached::{ConcurrentCached, RedbCache};
use censor::{Censor, Sex, Zealous};
use cpc::eval;
use dynfmt2::Format;
use eudex::Hash;
use gender_guesser::Gender;
use indicatif::{ProgressBar, ProgressDrawTarget};
use log::debug;
use minijinja::Environment;
use qsv_currency::Currency;
use qsv_vader_sentiment_analysis::SentimentIntensityAnalyzer;
use rayon::{
    iter::{IndexedParallelIterator, ParallelIterator},
    prelude::IntoParallelRefIterator,
};
use regex::Regex;
use serde::{Deserialize, Serialize};
use serde_json::json;
use smallvec::SmallVec;
use strsim::{
    damerau_levenshtein, hamming, jaro_winkler, normalized_damerau_levenshtein, osa_distance,
    sorensen_dice,
};
use strum_macros::EnumString;
use thousands::{Separable, SeparatorPolicy, policies};
use titlecase::titlecase;
use whatlang::detect;

use crate::{
    CliResult,
    clitypes::CliError,
    config::{Config, DEFAULT_RDR_BUFFER_CAPACITY, Delimiter},
    llmutil, regex_oncelock,
    select::{SelectColumns, Selection},
    util,
    util::replace_column_value,
};

#[derive(Clone, EnumString, PartialEq)]
#[strum(use_phf)]
#[strum(ascii_case_insensitive)]
#[allow(non_camel_case_types)]
enum Operations {
    Censor,
    Censor_Check,
    Censor_Count,
    Copy,
    Crc32,
    Currencytonum,
    Decode62,
    Decode64,
    Encode62,
    Encode64,
    Escape,
    Eudex,
    Gender_Guess,
    Len,
    Lower,
    Ltrim,
    Mltrim,
    Mrtrim,
    Mtrim,
    Numtocurrency,
    Regex_Replace,
    Replace,
    Round,
    Rtrim,
    Sentiment,
    Simdl,
    Simdln,
    Simhm,
    Simjw,
    Simod,
    Simsd,
    Squeeze,
    Squeeze0,
    Strip_Prefix,
    Strip_Suffix,
    Thousands,
    Titlecase,
    Trim,
    Upper,
    Whatlang,
}

#[derive(Deserialize)]
struct Args {
    arg_column:       SelectColumns,
    cmd_operations:   bool,
    arg_operations:   String,
    cmd_dynfmt:       bool,
    cmd_emptyreplace: bool,
    cmd_calcconv:     bool,
    cmd_summarize:    bool,
    arg_input:        Option<String>,
    flag_rename:      Option<String>,
    flag_comparand:   String,
    flag_replacement: String,
    flag_formatstr:   String,
    flag_batch:       usize,
    flag_jobs:        Option<usize>,
    flag_new_column:  Option<String>,
    flag_output:      Option<String>,
    flag_no_headers:  bool,
    flag_delimiter:   Option<Delimiter>,
    flag_progressbar: bool,
    // summarize subcommand options
    flag_base_url:    Option<String>,
    flag_model:       Option<String>,
    flag_api_key:     Option<String>,
    flag_max_tokens:  u32,
    flag_timeout:     u16,
    flag_addl_props:  Option<String>,
    flag_prompt:      Option<String>,
    flag_prompt_file: Option<String>,
    flag_rate_limit:  f64,
    flag_on_error:    String,
    flag_user_agent:  Option<String>,
    flag_cache_dir:   Option<String>,
    flag_no_cache:    bool,
    flag_fresh:       bool,
    flag_stats:       bool,
}

static CENSOR: OnceLock<Censor> = OnceLock::new();
static EUDEX_COMPARAND_HASH: OnceLock<eudex::Hash> = OnceLock::new();
static REGEX_REPLACE: OnceLock<Regex> = OnceLock::new();
static SENTIMENT_ANALYZER: OnceLock<SentimentIntensityAnalyzer> = OnceLock::new();
static THOUSANDS_POLICY: OnceLock<SeparatorPolicy> = OnceLock::new();
static ROUND_PLACES: OnceLock<u32> = OnceLock::new();
static WHATLANG_CONFIDENCE_THRESHOLD: OnceLock<f64> = OnceLock::new();
static GENDER_GUESSER: OnceLock<gender_guesser::Detector> = OnceLock::new();

// default confidence threshold for whatlang language detection - 90% confidence
const DEFAULT_THRESHOLD: f64 = 0.9;

// default number of decimal places to round to
const DEFAULT_ROUND_PLACES: u32 = 3;

const NULL_VALUE: &str = "<null>";

// summarize subcommand defaults (mirror describegpt)
const DEFAULT_LLM_BASE_URL: &str = "http://localhost:1234/v1";
const DEFAULT_LLM_MODEL: &str = "openai/gpt-oss-20b";
// default disk cache TTL for apply-summarize: 28 days (matches describegpt)
const DEFAULT_SUMMARIZE_CACHE_TTL_SECS: u64 = 60 * 60 * 24 * 28;

// What we persist in the disk cache for each summarized row. We store the token count &
// elapsed time (in addition to the content) so that --stats can report them even on a
// cache hit, using the values captured when the summary was first computed.
#[derive(Serialize, Deserialize, Clone)]
struct CachedSummary {
    content:      String,
    total_tokens: u64,
    elapsed_ms:   u64,
}

// for thousands operator
static INDIANCOMMA_POLICY: SeparatorPolicy = SeparatorPolicy {
    separator: ",",
    groups:    &[3, 2],
    digits:    thousands::digits::ASCII_DECIMAL,
};

// shared regex for 3-decimal-place workaround used by currencytonum & numtocurrency
fn fract_3digits_regex() -> &'static Regex {
    static RE: OnceLock<Regex> = OnceLock::new();
    RE.get_or_init(|| Regex::new(r"\.\d\d\d$").unwrap())
}

// valid subcommands
#[derive(PartialEq)]
enum ApplySubCmd {
    Operations,
    DynFmt,
    EmptyReplace,
    CalcConv,
    Summarize,
}

pub fn run(argv: &[&str]) -> CliResult<()> {
    let args: Args = util::get_args(USAGE, argv)?;
    let rconfig = Config::new(args.arg_input.as_ref())
        .delimiter(args.flag_delimiter)
        .no_headers_flag(args.flag_no_headers)
        .select(args.arg_column.clone())
        // since apply does batch processing,
        // use a bigger read buffer unless the user set their own buffer
        .set_read_buffer(if std::env::var("QSV_RDR_BUFFER_CAPACITY").is_err() {
            DEFAULT_RDR_BUFFER_CAPACITY * 10
        } else {
            DEFAULT_RDR_BUFFER_CAPACITY
        });

    let mut rdr = rconfig.reader()?;
    let mut wtr = Config::new(args.flag_output.as_ref()).writer()?;

    let headers = rdr.byte_headers()?.clone();
    let sel = rconfig.selection(&headers)?;
    if sel.is_empty() {
        return fail_incorrectusage_clierror!("No columns selected. Column selection is empty.");
    }
    // --new-column (-c) appends a single field per row, so it would produce
    // malformed CSV (one header, N data fields) when combined with multi-column
    // operations or emptyreplace. For multi-column transforms, omit --new-column
    // to update the selected columns in place; --rename (-r) is optional and only
    // changes the column names.
    if args.flag_new_column.is_some()
        && (args.cmd_operations || args.cmd_emptyreplace)
        && sel.len() > 1
    {
        return fail_incorrectusage_clierror!(
            "--new-column (-c) requires a single input column. For multi-column \
             operations/emptyreplace, omit --new-column to transform columns in place; optionally \
             use --rename (-r) to rename the transformed columns."
        );
    }
    // safety: we just checked that sel is not empty above
    let column_index = *sel.iter().next().unwrap();

    let mut headers = rdr.headers()?.clone();

    // The summarize subcommand has a fundamentally different (network-bound, sequential,
    // cached) execution model than the CPU-bound parallel subcommands, so handle it in a
    // dedicated function and return early.
    if args.cmd_summarize {
        return summarize_run(&args, &mut rdr, &mut wtr, &sel, &headers, &rconfig);
    }

    if let Some(new_name) = args.flag_rename {
        let new_col_names = util::ColumnNameParser::new(&new_name).parse()?;
        if new_col_names.len() != sel.len() {
            return fail_incorrectusage_clierror!(
                "Number of new columns does not match input column selection."
            );
        }
        for (i, col_index) in sel.iter().enumerate() {
            headers = replace_column_value(&headers, *col_index, &new_col_names[i]);
        }
    }

    // for dynfmt, safe_headers are the "safe" version of colnames - alphanumeric only,
    // all other chars replaced with underscore
    // dynfmt_fields are the columns used in the dynfmt --formatstr option
    // we prep it so we only populate the lookup vec with the index of these columns
    // so SimpleCurlyFormat is performant
    let dynfmt_template = if args.cmd_dynfmt || args.cmd_calcconv {
        if args.flag_no_headers {
            return fail_incorrectusage_clierror!("dynfmt/calcconv subcommand requires headers.");
        }

        let mut dynfmt_template_wrk = args.flag_formatstr.clone();
        let mut dynfmt_fields = Vec::new();

        // first, get the fields used in the dynfmt template
        let formatstr_re: &'static Regex = crate::regex_oncelock!(r"\{(?P<key>\w+)?\}");
        for format_fields in formatstr_re.captures_iter(&args.flag_formatstr) {
            // safety: we already checked that the regex match is valid
            dynfmt_fields.push(format_fields.name("key").unwrap().as_str());
        }
        // we sort the fields so we can do binary_search
        dynfmt_fields.sort_unstable();

        // now, get the indices of the columns for the lookup vec
        let (safe_headers, _) =
            util::safe_header_names(&headers, false, false, None, "", true, false, false);
        for (i, field) in safe_headers.iter().enumerate() {
            if dynfmt_fields.binary_search(&field.as_str()).is_ok() {
                let field_with_curly = format!("{{{field}}}");
                let field_index = format!("{{{i}}}");
                dynfmt_template_wrk = dynfmt_template_wrk.replace(&field_with_curly, &field_index);
            }
        }
        debug!("dynfmt_fields: {dynfmt_fields:?}  dynfmt_template: {dynfmt_template_wrk}");
        dynfmt_template_wrk
    } else {
        String::new()
    };

    let mut ops_vec = SmallVec::<[Operations; 4]>::new();

    let apply_cmd = if args.cmd_operations {
        ops_vec = validate_operations(
            &args.arg_operations.split(',').collect(),
            &args.flag_comparand,
            &args.flag_replacement,
            args.flag_new_column.as_ref(),
            &args.flag_formatstr,
        )?;
        ApplySubCmd::Operations
    } else if args.cmd_dynfmt {
        ApplySubCmd::DynFmt
    } else if args.cmd_emptyreplace {
        ApplySubCmd::EmptyReplace
    } else if args.cmd_calcconv {
        ApplySubCmd::CalcConv
    } else if args.cmd_summarize {
        ApplySubCmd::Summarize
    } else {
        return fail_incorrectusage_clierror!("Unknown apply subcommand.");
    };

    if !rconfig.no_headers {
        if let Some(new_column) = &args.flag_new_column {
            headers.push_field(new_column);
        }
        wtr.write_record(&headers)?;
    }

    // <NULL> (case-insensitive) on --replacement is recognized as "delete matches" by
    // regex_replace. Scope this rewrite to regex_replace only — other ops in the same chain
    // (e.g. `replace`, `numtocurrency`) still see the user's literal --replacement value, so
    // they aren't silently turned into no-ops.
    let flag_replacement = args.flag_replacement;
    let regex_replace_replacement = if apply_cmd == ApplySubCmd::Operations
        && ops_vec.contains(&Operations::Regex_Replace)
        && flag_replacement.eq_ignore_ascii_case(NULL_VALUE)
    {
        String::new()
    } else {
        flag_replacement.clone()
    };
    let flag_comparand = args.flag_comparand;
    let flag_formatstr = args.flag_formatstr;
    let flag_new_column = args.flag_new_column;

    // pre-compute a per-column selection mask so multi-column in-place transforms can
    // rebuild each record once per row instead of once per selected column. Only built
    // when --new-column is NOT set: the --new-column branch uses the single selected
    // column index directly and never consults the mask, so allocating it there would
    // be wasted, and skipping it also keeps the mask's len() == input-record width
    // (avoiding any reliance on `headers` not having been extended yet).
    let is_selected: Vec<bool> = if flag_new_column.is_some() {
        Vec::new()
    } else {
        let mut mask = vec![false; headers.len()];
        for col_index in &*sel {
            mask[*col_index] = true;
        }
        mask
    };

    // prep progress bar
    let show_progress =
        (args.flag_progressbar || util::get_envvar_flag("QSV_PROGRESSBAR")) && !rconfig.is_stdin();

    let progress = ProgressBar::with_draw_target(None, ProgressDrawTarget::stderr_with_hz(5));
    if show_progress {
        util::prep_progress(&progress, util::count_rows(&rconfig)?);
    } else {
        progress.set_draw_target(ProgressDrawTarget::hidden());
    }

    // amortize memory allocation by reusing record
    #[allow(unused_assignments)]
    let mut batch_record = csv::StringRecord::new();

    let num_jobs = util::njobs(args.flag_jobs);

    // reuse batch buffers
    let batchsize = util::optimal_batch_size(&rconfig, args.flag_batch, num_jobs);
    let mut batch = Vec::with_capacity(batchsize);
    let mut batch_results = Vec::with_capacity(batchsize);

    // main loop to read CSV and construct batches for parallel processing.
    // each batch is processed via Rayon parallel iterator.
    // loop exits when batch is empty.
    'batch_loop: loop {
        for _ in 0..batchsize {
            match rdr.read_record(&mut batch_record) {
                Ok(true) => batch.push(std::mem::take(&mut batch_record)),
                Ok(false) => break, // nothing else to add to batch
                Err(e) => {
                    return fail_clierror!("Error reading file: {e}");
                },
            }
        }

        if batch.is_empty() {
            // break out of infinite loop when at EOF
            break 'batch_loop;
        }

        // do actual apply command via Rayon parallel iterator
        batch
            .par_iter()
            .with_min_len(1024)
            .map(|record_item| {
                let mut record = record_item.clone();
                match apply_cmd {
                    ApplySubCmd::Operations => {
                        let mut cell = String::new();
                        if flag_new_column.is_some() {
                            // single-column case (validated upstream: --new-column requires
                            // sel.len() == 1 for operations/emptyreplace)
                            let col_index = *sel.iter().next().unwrap();
                            record[col_index].clone_into(&mut cell);
                            apply_operations(
                                &ops_vec,
                                &mut cell,
                                &flag_comparand,
                                &flag_replacement,
                                &regex_replace_replacement,
                                &flag_formatstr,
                            );
                            record.push_field(&cell);
                        } else {
                            // multi-column (or single-column) in-place: rebuild the record
                            // once instead of once per selected column (replace_column_value
                            // allocates a fresh StringRecord per call). Pre-size the new
                            // record to the input's exact byte/field footprint to avoid
                            // per-push growth reallocations on wide CSVs.
                            let mut new_record = csv::StringRecord::with_capacity(
                                record.as_byte_record().as_slice().len(),
                                record.len(),
                            );
                            for (i, field) in record.iter().enumerate() {
                                if is_selected[i] {
                                    cell.clear();
                                    cell.push_str(field);
                                    apply_operations(
                                        &ops_vec,
                                        &mut cell,
                                        &flag_comparand,
                                        &flag_replacement,
                                        &regex_replace_replacement,
                                        &flag_formatstr,
                                    );
                                    new_record.push_field(&cell);
                                } else {
                                    new_record.push_field(field);
                                }
                            }
                            record = new_record;
                        }
                    },
                    ApplySubCmd::EmptyReplace => {
                        if flag_new_column.is_some() {
                            // single-column case (validated upstream)
                            let col_index = *sel.iter().next().unwrap();
                            let field = &record[col_index];
                            let new_field = if field.trim().is_empty() {
                                flag_replacement.as_str()
                            } else {
                                field
                            };
                            // borrow ends before push_field
                            let new_field = new_field.to_owned();
                            record.push_field(&new_field);
                        } else {
                            // rebuild once instead of once per selected column. Pre-size
                            // to the input record's footprint to avoid push-growth reallocs.
                            let mut new_record = csv::StringRecord::with_capacity(
                                record.as_byte_record().as_slice().len(),
                                record.len(),
                            );
                            for (i, field) in record.iter().enumerate() {
                                if is_selected[i] && field.trim().is_empty() {
                                    new_record.push_field(&flag_replacement);
                                } else {
                                    new_record.push_field(field);
                                }
                            }
                            record = new_record;
                        }
                    },
                    ApplySubCmd::DynFmt => {
                        let record_vec: Vec<&str> = record.iter().collect();
                        let cell = match dynfmt2::SimpleCurlyFormat
                            .format(&dynfmt_template, &*record_vec)
                        {
                            Ok(formatted) => formatted.into_owned(),
                            Err(_) => record[column_index].to_owned(),
                        };
                        if flag_new_column.is_some() {
                            record.push_field(&cell);
                        } else {
                            record = replace_column_value(&record, column_index, &cell);
                        }
                    },
                    ApplySubCmd::CalcConv => {
                        let record_vec: Vec<&str> = record.iter().collect();
                        let formatted = match dynfmt2::SimpleCurlyFormat
                            .format(&dynfmt_template, &*record_vec)
                        {
                            Ok(formatted) => formatted.into_owned(),
                            Err(_) => record[column_index].to_owned(),
                        };

                        let (cell_for_eval, append_unit) = if formatted.ends_with("<UNIT>") {
                            // strip ALL trailing <UNIT> occurrences (matches the original
                            // trim_end_matches behavior — strip_suffix would only remove one)
                            (formatted.trim_end_matches("<UNIT>"), true)
                        } else {
                            (formatted.as_str(), false)
                        };
                        let result = if cell_for_eval.trim().is_empty() {
                            String::new()
                        } else {
                            match eval(cell_for_eval, true, false) {
                                Ok(answer) => {
                                    if append_unit {
                                        // cpc 4 changed `answer.unit` to a compound-unit vector
                                        // (`Vec<(Unit, exponent)>`); `Number::singular()` renders
                                        // it as a human-readable suffix (e.g. "watt-hour") instead
                                        // of the raw debug form (`[(WattHour, 1)]`).
                                        let unit = answer.singular();
                                        if unit.is_empty() {
                                            answer.value.to_string()
                                        } else {
                                            format!("{} {unit}", answer.value)
                                        }
                                    } else {
                                        answer.value.to_string()
                                    }
                                },
                                Err(e) => format!("ERROR: {e}"),
                            }
                        };

                        if flag_new_column.is_some() {
                            record.push_field(&result);
                        } else {
                            record = replace_column_value(&record, column_index, &result);
                        }
                    },
                    // summarize is network-bound & handled in summarize_run() before this
                    // parallel batch loop is ever reached.
                    ApplySubCmd::Summarize => unreachable!(),
                }

                record
            })
            .collect_into_vec(&mut batch_results);

        // rayon collect() guarantees original order, so we can just append results each batch
        for result_record in &batch_results {
            wtr.write_record(result_record)?;
        }

        if show_progress {
            progress.inc(batch.len() as u64);
        }

        batch.clear();
    } // end batch loop

    if show_progress {
        util::finish_progress(&progress);
    }
    Ok(wtr.flush()?)
}

/// Returns true if `base_url`'s host is a loopback address (localhost / 127.0.0.1 / `::1`).
///
/// Parses the URL and checks the actual host so values like `https://localhost.evil.com/v1`
/// are NOT misclassified as local (a naive substring match would). Falls back to a substring
/// check only if the URL can't be parsed.
fn is_localhost_url(base_url: &str) -> bool {
    match reqwest::Url::parse(base_url) {
        Ok(url) => matches!(
            url.host_str(),
            Some("localhost" | "127.0.0.1" | "::1" | "[::1]")
        ),
        Err(_) => base_url.contains("localhost") || base_url.contains("127.0.0.1"),
    }
}

/// Runs the `apply summarize` subcommand.
///
/// Unlike the other (CPU-bound, Rayon-parallel) apply subcommands, summarize is network-bound:
/// it renders a MiniJinja prompt per record, calls an OpenAI API-compatible LLM, and stores
/// the returned summary in a new column. It runs sequentially so HTTP errors propagate cleanly
/// and requests can be rate-limited, and it caches results on disk so repeated rows / re-runs
/// don't re-pay for inference.
#[allow(clippy::too_many_lines)]
fn summarize_run<R: std::io::Read, W: std::io::Write>(
    args: &Args,
    rdr: &mut csv::Reader<R>,
    wtr: &mut csv::Writer<W>,
    sel: &Selection,
    headers: &csv::StringRecord,
    rconfig: &Config,
) -> CliResult<()> {
    if args.flag_no_headers {
        return fail_incorrectusage_clierror!("apply summarize requires headers.");
    }
    let Some(new_column) = args.flag_new_column.as_ref() else {
        return fail_incorrectusage_clierror!("apply summarize requires --new-column (-c).");
    };

    // Sanitize header names into safe MiniJinja context keys (non-alphanumeric -> '_').
    // Use util::safe_header_names rather than a naive per-char map so that:
    //  - headers which sanitize to the same key (e.g. `a.b` and `a_b`) get unique disambiguated
    //    names (`a_b`, `a_b_2`) instead of silently colliding & overwriting each other; and
    //  - digit-leading headers (e.g. `1st_col`) get a leading `_` (check_first_char=true), since
    //    MiniJinja/Jinja identifiers can't start with a digit (`{{ 1st_col }}` would fail).
    // keep_case=true preserves the original casing. The SAME mapping drives both the default
    // prompt and the per-record context insertion.
    let (sanitized_headers, _) =
        util::safe_header_names(headers, true, false, None, "", true, false, false);

    // resolve effective base_url / model / api_key: CLI flag > env var > built-in default
    let base_url = args
        .flag_base_url
        .clone()
        .or_else(|| std::env::var("QSV_LLM_BASE_URL").ok())
        .unwrap_or_else(|| DEFAULT_LLM_BASE_URL.to_string());
    let model = args
        .flag_model
        .clone()
        .or_else(|| std::env::var("QSV_LLM_MODEL").ok())
        .unwrap_or_else(|| DEFAULT_LLM_MODEL.to_string());
    let resolved_key = args
        .flag_api_key
        .clone()
        .or_else(|| std::env::var("QSV_LLM_APIKEY").ok())
        .unwrap_or_default();

    let is_localhost = is_localhost_url(&base_url);
    // "NONE" (case-insensitive) means: explicitly don't send an Authorization header
    let suppress_key = resolved_key.eq_ignore_ascii_case("NONE");
    let api_key = if suppress_key {
        String::new()
    } else {
        resolved_key
    };
    // require an API key for non-localhost endpoints (unless explicitly suppressed with NONE)
    if !is_localhost && api_key.is_empty() && !suppress_key {
        return fail_incorrectusage_clierror!(
            "An --api-key (or QSV_LLM_APIKEY env var) is required for non-localhost endpoints. \
             Set --api-key NONE to suppress sending a key."
        );
    }

    // localhost LLMs typically don't want a max_tokens cap; mirror describegpt behavior
    let max_tokens = if is_localhost || args.flag_max_tokens == 0 {
        None
    } else {
        Some(args.flag_max_tokens)
    };

    // build the prompt template: --prompt > --prompt-file > generated default
    let prompt_str = if let Some(p) = args.flag_prompt.clone() {
        p
    } else if let Some(pf) = args.flag_prompt_file.as_ref() {
        std::fs::read_to_string(pf)
            .map_err(|e| CliError::Other(format!("Cannot read --prompt-file '{pf}': {e}")))?
    } else {
        // default: summarize the selected column/s
        let mut s = String::from("Summarize the following text concisely:\n");
        for col_index in sel.iter() {
            let safe = &sanitized_headers[*col_index];
            s.push_str(&format!("{safe}: {{{{ {safe} }}}}\n"));
        }
        s
    };

    // set up MiniJinja with qsv's shared filter set (regex, datefmt, slugify, etc.)
    let mut env = Environment::new();
    minijinja_contrib::add_to_environment(&mut env);
    env.set_unknown_method_callback(minijinja_contrib::pycompat::unknown_method_callback);
    crate::minijinja_filters::register(&mut env);
    env.add_template_owned("prompt", prompt_str)
        .map_err(|e| CliError::Other(format!("Invalid prompt template: {e}")))?;
    let template = env
        .get_template("prompt")
        .map_err(|e| CliError::Other(format!("Invalid prompt template: {e}")))?;

    // build the blocking HTTP client
    let client = util::create_reqwest_blocking_client(
        args.flag_user_agent.clone(),
        args.flag_timeout,
        Some(base_url.clone()),
    )?;

    // set up the disk cache (unless --no-cache)
    let cache: Option<RedbCache<String, CachedSummary>> = if args.flag_no_cache {
        None
    } else {
        let cache_dir = if let Some(dir) = args.flag_cache_dir.clone() {
            dir
        } else {
            util::expand_tilde("~/.qsv-cache/apply-summarize")
                .ok_or_else(|| {
                    CliError::Other("Cannot determine default cache directory".to_string())
                })?
                .to_string_lossy()
                .into_owned()
        };
        let ttl_secs = std::env::var("QSV_DISKCACHE_TTL_SECS")
            .ok()
            .and_then(|s| s.parse::<u64>().ok())
            .unwrap_or(DEFAULT_SUMMARIZE_CACHE_TTL_SECS);
        let mut dc_builder = RedbCache::builder("apply-summarize")
            .disk_dir(&cache_dir)
            .durable(true);
        // A QSV_DISKCACHE_TTL_SECS of 0 disables time-based expiration (entries
        // are cached indefinitely). v3's RedbCache builder rejects .ttl(0), so
        // only set a TTL when it is non-zero; leaving it unset means "never expire".
        if ttl_secs != 0 {
            dc_builder = dc_builder.ttl(Duration::from_secs(ttl_secs));
        }
        let dc = dc_builder
            .build()
            .map_err(|e| CliError::Other(format!("Cannot build disk cache: {e:?}")))?;
        Some(dc)
    };

    let addl_props = args.flag_addl_props.as_deref();

    // write headers: new_column (+ optional stats columns)
    let mut out_headers = headers.clone();
    out_headers.push_field(new_column);
    if args.flag_stats {
        out_headers.push_field(&format!("{new_column}_elapsed_ms"));
        out_headers.push_field(&format!("{new_column}_tokens"));
    }
    wtr.write_record(&out_headers)?;

    // progress bar
    let show_progress =
        (args.flag_progressbar || util::get_envvar_flag("QSV_PROGRESSBAR")) && !rconfig.is_stdin();
    let progress = ProgressBar::with_draw_target(None, ProgressDrawTarget::stderr_with_hz(5));
    if show_progress {
        util::prep_progress(&progress, util::count_rows(rconfig)?);
    } else {
        progress.set_draw_target(ProgressDrawTarget::hidden());
    }

    let rate_limit = if args.flag_rate_limit > 0.0 {
        Some(Duration::from_secs_f64(args.flag_rate_limit))
    } else {
        None
    };
    let skip_on_error = args.flag_on_error.eq_ignore_ascii_case("skip");

    let mut record = csv::StringRecord::new();
    let mut row_number: u64 = 0;
    let mut first_request = true;
    while rdr.read_record(&mut record)? {
        row_number += 1;

        // build the per-record template context (sanitized header -> field) + QSV_ROWNO
        let rowno_str = row_number.to_string();
        let mut ctx: BTreeMap<&str, &str> = BTreeMap::new();
        for (header, field) in sanitized_headers.iter().zip(record.iter()) {
            ctx.insert(header.as_str(), field);
        }
        ctx.insert("QSV_ROWNO", rowno_str.as_str());

        let user_prompt = template
            .render(&ctx)
            .map_err(|e| CliError::Other(format!("Prompt render error (row {row_number}): {e}")))?;

        // cache key: base_url | model | max_tokens | addl_props | rendered prompt
        let cache_key = blake3::hash(
            format!(
                "{base_url}|{model}|{max_tokens:?}|{}|{user_prompt}",
                addl_props.unwrap_or("")
            )
            .as_bytes(),
        )
        .to_hex()
        .to_string();

        // try the cache first (unless --fresh forces a new call)
        let mut cached: Option<CachedSummary> = None;
        if let Some(cache) = &cache
            && !args.flag_fresh
        {
            cached = cache
                .cache_get(&cache_key)
                .map_err(|e| CliError::Other(format!("Disk cache error: {e:?}")))?;
        }

        let summary = if let Some(cs) = cached {
            cs
        } else {
            // rate-limit between live requests
            if let Some(rl) = rate_limit
                && !first_request
            {
                std::thread::sleep(rl);
            }
            first_request = false;

            let messages = json!([{ "role": "user", "content": user_prompt }]);
            let cs = match llmutil::chat_completion(
                &client, &base_url, &api_key, &model, max_tokens, &messages, addl_props,
            ) {
                Ok(resp) => CachedSummary {
                    content:      resp.content,
                    total_tokens: resp.total_tokens,
                    elapsed_ms:   resp.elapsed_ms,
                },
                Err(e) => {
                    if skip_on_error {
                        CachedSummary {
                            content:      format!("<ERROR: {e}>"),
                            total_tokens: 0,
                            elapsed_ms:   0,
                        }
                    } else {
                        return Err(e);
                    }
                },
            };
            // only cache successful results
            if let Some(cache) = &cache
                && !cs.content.starts_with("<ERROR:")
            {
                cache
                    .cache_set(cache_key.clone(), cs.clone())
                    .map_err(|e| CliError::Other(format!("Disk cache error: {e:?}")))?;
            }
            cs
        };

        record.push_field(&summary.content);
        if args.flag_stats {
            record.push_field(&summary.elapsed_ms.to_string());
            record.push_field(&summary.total_tokens.to_string());
        }
        wtr.write_record(&record)?;

        if show_progress {
            progress.inc(1);
        }
    }

    if show_progress {
        util::finish_progress(&progress);
    }
    Ok(wtr.flush()?)
}

fn validate_operations(
    operations: &Vec<&str>,
    flag_comparand: &str,
    flag_replacement: &str,
    flag_new_column: Option<&String>,
    flag_formatstr: &str,
) -> Result<SmallVec<[Operations; 4]>, CliError> {
    let mut censor_invokes = 0_u8;
    let mut copy_invokes = 0_u8;
    let mut eudex_invokes = 0_u8;
    let mut regex_replace_invokes = 0_u8;
    let mut replace_invokes = 0_u8;
    let mut sentiment_invokes = 0_u8;
    let mut sim_invokes = 0_u8;
    let mut strip_invokes = 0_u8;
    let mut whatlang_invokes = 0_u8;

    // SmallVec::new() preserves inline storage for the common case of <=4 ops;
    // with_capacity(n) forces a heap allocation when n > 0, even when n <= 4.
    let mut ops_vec = SmallVec::new();

    for op in operations {
        let Ok(operation) = Operations::from_str(op) else {
            return fail_incorrectusage_clierror!("Unknown '{op}' operation");
        };
        match operation {
            Operations::Censor | Operations::Censor_Check | Operations::Censor_Count => {
                if flag_new_column.is_none() {
                    return fail_incorrectusage_clierror!(
                        "--new-column (-c) is required for censor operations."
                    );
                }
                if censor_invokes == 0 {
                    // OnceLock::set can only fail if already set; the invokes guard makes
                    // that path unreachable here.
                    let mut censored_words = Censor::Standard + Zealous + Sex;
                    for word in flag_comparand.split(',') {
                        censored_words += word.trim();
                    }
                    let _ = CENSOR.set(censored_words);
                }
                censor_invokes = censor_invokes.saturating_add(1);
            },
            Operations::Copy => {
                if flag_new_column.is_none() {
                    return fail_incorrectusage_clierror!(
                        "--new-column (-c) is required for copy operation."
                    );
                }
                copy_invokes = copy_invokes.saturating_add(1);
            },
            Operations::Eudex => {
                if flag_comparand.is_empty() || flag_new_column.is_none() {
                    return fail_incorrectusage_clierror!(
                        "--comparand (-C) and --new-column (-c) are required for eudex."
                    );
                }
                if eudex_invokes == 0 {
                    let _ = EUDEX_COMPARAND_HASH.set(eudex::Hash::new(flag_comparand));
                }
                eudex_invokes = eudex_invokes.saturating_add(1);
            },
            Operations::Mtrim | Operations::Mltrim | Operations::Mrtrim
                if flag_comparand.is_empty() =>
            {
                return fail_incorrectusage_clierror!(
                    "--comparand (-C) is required for match trim operations."
                );
            },
            Operations::Regex_Replace => {
                if flag_comparand.is_empty() || flag_replacement.is_empty() {
                    return fail_incorrectusage_clierror!(
                        "--comparand (-C) and --replacement (-R) are required for regex_replace \
                         operation."
                    );
                }
                if regex_replace_invokes == 0 {
                    let re = match regex::Regex::new(flag_comparand) {
                        Ok(re) => re,
                        Err(err) => {
                            return fail_incorrectusage_clierror!(
                                "regex_replace expression error: {err:?}"
                            );
                        },
                    };
                    let _ = REGEX_REPLACE.set(re);
                }
                regex_replace_invokes = regex_replace_invokes.saturating_add(1);
            },
            Operations::Replace => {
                if flag_comparand.is_empty() || flag_replacement.is_empty() {
                    return fail_incorrectusage_clierror!(
                        "--comparand (-C) and --replacement (-R) are required for replace \
                         operation."
                    );
                }
                replace_invokes = replace_invokes.saturating_add(1);
            },
            Operations::Sentiment => {
                if flag_new_column.is_none() {
                    return fail_incorrectusage_clierror!(
                        "--new-column (-c) is required for sentiment operation."
                    );
                }
                if sentiment_invokes == 0 {
                    let _ = SENTIMENT_ANALYZER.set(SentimentIntensityAnalyzer::new());
                }
                sentiment_invokes = sentiment_invokes.saturating_add(1);
            },
            Operations::Simdl
            | Operations::Simdln
            | Operations::Simjw
            | Operations::Simsd
            | Operations::Simhm
            | Operations::Simod => {
                if flag_comparand.is_empty() || flag_new_column.is_none() {
                    return fail_incorrectusage_clierror!(
                        "--comparand (-C) and --new-column (-c) are required for similarity \
                         operations."
                    );
                }
                sim_invokes = sim_invokes.saturating_add(1);
            },
            Operations::Strip_Prefix | Operations::Strip_Suffix => {
                if flag_comparand.is_empty() {
                    return fail_incorrectusage_clierror!(
                        "--comparand (-C) is required for strip operations."
                    );
                }
                strip_invokes = strip_invokes.saturating_add(1);
            },
            Operations::Thousands => {
                THOUSANDS_POLICY.get_or_init(|| match flag_formatstr {
                    "dot" => policies::DOT_SEPARATOR,
                    "space" => policies::SPACE_SEPARATOR,
                    "underscore" => policies::UNDERSCORE_SEPARATOR,
                    "hexfour" => policies::HEX_FOUR,
                    "indiancomma" => INDIANCOMMA_POLICY,
                    _ => policies::COMMA_SEPARATOR,
                });
            },
            Operations::Round => {
                ROUND_PLACES.get_or_init(|| {
                    flag_formatstr
                        .parse::<u32>()
                        .unwrap_or(DEFAULT_ROUND_PLACES)
                });
            },
            Operations::Whatlang => {
                if flag_new_column.is_none() {
                    return fail_incorrectusage_clierror!(
                        "--new-column (-c) is required for whatlang language detection."
                    );
                }

                if whatlang_invokes == 0 {
                    let threshold = if flag_comparand.is_empty() {
                        DEFAULT_THRESHOLD
                    } else {
                        let preparsed_threshold;
                        let show_confidence = if flag_comparand.ends_with('?') {
                            preparsed_threshold = flag_comparand.trim_end_matches('?');
                            true
                        } else {
                            preparsed_threshold = flag_comparand;
                            false
                        };
                        let desired_threshold = preparsed_threshold
                            .parse::<f64>()
                            .unwrap_or(DEFAULT_THRESHOLD);
                        // desired threshold can be 0.0 to 1.0
                        let final_threshold = if (0.0..=1.0).contains(&desired_threshold) {
                            desired_threshold
                        } else {
                            // its outside the valid range
                            // just set it to the default threshold
                            DEFAULT_THRESHOLD
                        };
                        if show_confidence {
                            -final_threshold
                        } else {
                            final_threshold
                        }
                    };
                    let _ = WHATLANG_CONFIDENCE_THRESHOLD.set(threshold);
                }
                whatlang_invokes = whatlang_invokes.saturating_add(1);
            },
            Operations::Gender_Guess => {
                if flag_new_column.is_none() {
                    return fail_incorrectusage_clierror!(
                        "--new-column (-c) is required for gender_guess."
                    );
                }
                GENDER_GUESSER.get_or_init(gender_guesser::Detector::new);
            },
            _ => {},
        }
        ops_vec.push(operation);
    }
    if censor_invokes > 1
        || copy_invokes > 1
        || eudex_invokes > 1
        || regex_replace_invokes > 1
        || replace_invokes > 1
        || sentiment_invokes > 1
        || sim_invokes > 1
        || strip_invokes > 1
        || whatlang_invokes > 1
    {
        return fail_incorrectusage_clierror!(
            "you can only use censor({censor_invokes}), copy({copy_invokes}), \
             eudex({eudex_invokes}), regex_replace({regex_replace_invokes}), \
             replace({replace_invokes}), sentiment({sentiment_invokes}), \
             similarity({sim_invokes}), strip({strip_invokes}), and whatlang({whatlang_invokes}) \
             ONCE per operation series."
        );
    }

    Ok(ops_vec) // no validation errors
}

#[inline]
fn apply_operations(
    ops_vec: &SmallVec<[Operations; 4]>,
    cell: &mut String,
    comparand: &str,
    replacement: &str,
    regex_replacement: &str,
    formatstr: &str,
) {
    for op in ops_vec {
        match op {
            Operations::Len => {
                itoa::Buffer::new().format(cell.len()).clone_into(cell);
            },
            Operations::Lower => {
                *cell = cell.to_lowercase();
            },
            Operations::Upper => {
                *cell = cell.to_uppercase();
            },
            Operations::Squeeze => {
                let squeezer: &'static Regex = regex_oncelock!(r"\s+");
                *cell = squeezer.replace_all(cell, " ").into_owned();
            },
            Operations::Squeeze0 => {
                let squeezer: &'static Regex = regex_oncelock!(r"\s+");
                *cell = squeezer.replace_all(cell, "").into_owned();
            },
            Operations::Trim => {
                *cell = String::from(cell.trim());
            },
            Operations::Ltrim => {
                *cell = String::from(cell.trim_start());
            },
            Operations::Rtrim => {
                *cell = String::from(cell.trim_end());
            },
            Operations::Mtrim => {
                let chars_to_trim: &[char] = &comparand.chars().collect::<Vec<_>>();
                *cell = String::from(cell.trim_matches(chars_to_trim));
            },
            Operations::Mltrim => {
                *cell = String::from(cell.trim_start_matches(comparand));
            },
            Operations::Mrtrim => {
                *cell = String::from(cell.trim_end_matches(comparand));
            },
            Operations::Encode64 => {
                *cell = BASE64.encode_to_string(cell.as_bytes());
            },
            Operations::Decode64 => {
                *cell = match BASE64.decode_to_vec(cell.as_bytes()) {
                    Ok(decoded) => simdutf8::basic::from_utf8(&decoded)
                        .unwrap_or_default()
                        .to_owned(),
                    Err(e) => format!("decoding64 error: {e:?}"),
                };
            },
            Operations::Encode62 => {
                *cell = match cell.parse::<u128>() {
                    Ok(num) => base62::encode(num),
                    Err(e) => format!("encode62 error: {e:?}"),
                };
            },
            Operations::Decode62 => {
                *cell = match base62::decode(cell.as_str()) {
                    Ok(decoded) => decoded.to_string(),
                    Err(e) => format!("decode62 error: {e:?}"),
                };
            },
            Operations::Crc32 => {
                let mut crc32_hasher = crc32fast::Hasher::new();
                crc32_hasher.update(cell.as_bytes());
                itoa::Buffer::new()
                    .format(crc32_hasher.finalize())
                    .clone_into(cell);
            },
            Operations::Gender_Guess => {
                // safety: we set GENDER_GUESSER in validate_operations()
                let gender_detector = GENDER_GUESSER.get().unwrap();
                *cell = match gender_detector.get_gender(cell) {
                    Gender::Male => "Male".to_owned(),
                    Gender::Female => "Female".to_owned(),
                    Gender::MayBeMale => "MayBeMale".to_owned(),
                    Gender::MayBeFemale => "MayBeFemale".to_owned(),
                    Gender::BothMaleFemale => "BothMaleFemale".to_owned(),
                    Gender::NotSure => "NotSure".to_owned(),
                    Gender::NotFound => "NotFound".to_owned(),
                };
            },
            Operations::Escape => {
                *cell = cell.escape_default().to_string();
            },
            Operations::Strip_Prefix => {
                if let Some(stripped) = cell.strip_prefix(comparand) {
                    *cell = String::from(stripped);
                }
            },
            Operations::Strip_Suffix => {
                if let Some(stripped) = cell.strip_suffix(comparand) {
                    *cell = String::from(stripped);
                }
            },
            Operations::Titlecase => {
                *cell = titlecase(cell);
            },
            Operations::Replace => {
                *cell = cell.replace(comparand, replacement);
            },
            Operations::Regex_Replace => {
                // safety: we set REGEX_REPLACE in validate_operations()
                let regexreplace = REGEX_REPLACE.get().unwrap();
                *cell = regexreplace
                    .replace_all(cell, regex_replacement)
                    .into_owned();
            },
            Operations::Censor => {
                // safety: we set CENSOR in validate_operations()
                let censor = CENSOR.get().unwrap();
                *cell = censor.censor(cell);
            },
            Operations::Censor_Check => {
                // safety: we set CENSOR in validate_operations()
                let censor = CENSOR.get().unwrap();
                if censor.check(cell) { "true" } else { "false" }.clone_into(cell);
            },
            Operations::Censor_Count => {
                // safety: we set CENSOR in validate_operations()
                let censor = CENSOR.get().unwrap();
                itoa::Buffer::new()
                    .format(censor.count(cell))
                    .clone_into(cell);
            },
            Operations::Thousands => {
                if let Ok(num) = cell.parse::<f64>() {
                    //safety: we set THOUSANDS_POLICY in validate_operations()
                    let mut temp_string = num.separate_by_policy(*THOUSANDS_POLICY.get().unwrap());

                    // If the number has no fractional part, use the formatted string as-is.
                    // Otherwise, swap in the requested decimal separator from --replacement.
                    // We compare against == 0.0 (rather than > 0.0 in the else branch) because
                    // f64::fract() preserves sign: (-1.5).fract() == -0.5, so a `> 0.0` test
                    // would skip negative fractional numbers and leave their decimal point as
                    // `.` regardless of --replacement.
                    if num.fract() == 0.0 {
                        *cell = temp_string;
                    } else {
                        // if replacement is empty, use the default decimal separator (.)
                        *cell = if replacement.is_empty() {
                            temp_string
                        } else {
                            // else replace the decimal separator (last '.') w/ the requested one
                            match temp_string.rfind('.') {
                                Some(last_dot) => {
                                    temp_string.replace_range(last_dot..=last_dot, replacement);
                                    temp_string
                                },
                                None => temp_string,
                            }
                        };
                    }
                }
            },
            Operations::Round => {
                if let Ok(num) = cell.parse::<f64>() {
                    // safety: we set ROUND_PLACES in validate_operations()
                    *cell = util::round_num(num, *ROUND_PLACES.get().unwrap());
                }
            },
            Operations::Currencytonum => {
                // Handle currency strings with 3 decimal places by appending a 0
                // to make it 4 decimal places without affecting the value
                // This is a workaround around current limitation of qsv-currency
                // and also of upstream currency-rs, that it cannot
                // handle currency amounts properly with three decimal places
                let fract_3digits = fract_3digits_regex();
                let cell_val = if fract_3digits.is_match(cell) {
                    format!("{cell}0")
                } else {
                    cell.clone()
                };

                if let Ok(currency_val) = Currency::from_str(&cell_val) {
                    if formatstr == "strict" {
                        // Process ISO currency values
                        let currency_coins = currency_val.value();
                        let coins = format!("{currency_coins:03}");

                        if currency_val.is_iso_currency() {
                            if coins == "000" {
                                *cell = "0.00".to_string();
                            } else {
                                let coinlen = coins.len();
                                if coinlen > 2 {
                                    let decpoint = coinlen - 2;
                                    let coin_num = &coins[..decpoint];
                                    let coin_frac = &coins[decpoint..];
                                    *cell = format!("{coin_num}.{coin_frac}");
                                }
                            }
                        }
                    } else {
                        // For non-strict mode, extract numeric parts from currency strings
                        // using regex that handles various formats including thousand separators
                        // we ignore the currency symbol altogether
                        let numparts_re: &'static Regex =
                            regex_oncelock!(r"-?(?:\d{1,3}(?:[,. ]\d{3})+|(?:\d+))(?:[,.]\d+)?");

                        if let Some(numparts) = numparts_re.find(cell) {
                            // use the same workaround as above to handle 3 decimal places
                            let numparts_str = numparts.as_str();
                            let numparts_val = if fract_3digits.is_match(numparts_str) {
                                format!("{numparts_str}0")
                            } else {
                                numparts_str.to_string()
                            };

                            if let Ok(extracted_currency) = Currency::from_str(&numparts_val) {
                                let currency_coins = extracted_currency.value();
                                let coins = format!("{currency_coins:03}");

                                if coins == "000" {
                                    *cell = "0.00".to_string();
                                } else {
                                    let coinlen = coins.len();
                                    if coinlen > 2 {
                                        let decpoint = coinlen - 2;
                                        let coin_num = &coins[..decpoint];
                                        let coin_frac = &coins[decpoint..];
                                        *cell = format!("{coin_num}.{coin_frac}");
                                    }
                                }
                            }
                        }
                    }
                }
            },
            Operations::Numtocurrency => {
                // same 3 decimal place workaround as currencytonum
                let fract_3digits = fract_3digits_regex();
                let cell_val = if fract_3digits.is_match(cell) {
                    format!("{cell}0")
                } else {
                    cell.clone()
                };

                if let Ok(currency_value) = Currency::from_str(&cell_val) {
                    let currency_wrk = currency_value
                        .convert(replacement.parse::<f64>().unwrap_or(1.0_f64), comparand);
                    *cell = if formatstr == "euro" {
                        format!("{currency_wrk:e}")
                    } else {
                        format!("{currency_wrk}")
                    };
                }
            },
            Operations::Simdl => {
                itoa::Buffer::new()
                    .format(damerau_levenshtein(cell, comparand))
                    .clone_into(cell);
            },
            Operations::Simdln => {
                zmij::Buffer::new()
                    .format_finite(normalized_damerau_levenshtein(cell, comparand))
                    .clone_into(cell);
            },
            Operations::Simjw => {
                zmij::Buffer::new()
                    .format_finite(jaro_winkler(cell, comparand))
                    .clone_into(cell);
            },
            Operations::Simsd => {
                zmij::Buffer::new()
                    .format_finite(sorensen_dice(cell, comparand))
                    .clone_into(cell);
            },
            Operations::Simhm => {
                let ham_val = hamming(cell, comparand);
                match ham_val {
                    Ok(val) => itoa::Buffer::new().format(val).clone_into(cell),
                    Err(_) => *cell = String::from("ERROR: Different lengths"),
                }
            },
            Operations::Simod => {
                itoa::Buffer::new()
                    .format(osa_distance(cell, comparand))
                    .clone_into(cell);
            },
            Operations::Eudex => {
                // safety: we set EUDEX_COMPARAND_HASH in validate_operations()
                let eudex_comparand_hash = EUDEX_COMPARAND_HASH.get().unwrap();
                let cell_hash = Hash::new(cell);
                *cell = format!("{}", (cell_hash - *eudex_comparand_hash).similar());
            },
            Operations::Sentiment => {
                // safety: we set SENTIMENT_ANALYZER in validate_operations()
                let sentiment_analyzer = SENTIMENT_ANALYZER.get().unwrap();
                let sentiment_scores = sentiment_analyzer.polarity_scores(cell);
                zmij::Buffer::new()
                    .format_finite(*sentiment_scores.get("compound").unwrap_or(&0.0))
                    .clone_into(cell);
            },
            Operations::Whatlang => {
                let lang_info = detect(cell);
                if let Some(lang_info) = lang_info {
                    // safety: we set WHATLANG_CONFIDENCE_THRESHOLD in validate_operations()
                    let whatlang_confidence_threshold =
                        *WHATLANG_CONFIDENCE_THRESHOLD.get().unwrap();
                    let lang_confidence = lang_info.confidence();
                    let lang = lang_info.lang();
                    if lang_confidence >= whatlang_confidence_threshold.abs() {
                        if whatlang_confidence_threshold >= 0.0 {
                            *cell = format!("{lang:?}");
                        } else {
                            *cell = format!("{lang:?}({lang_confidence:.3})");
                        }
                    } else {
                        // if confidence < confidence_threshold; show best-guessed language,
                        // confidence to 3 decimal places enclosed in parens and
                        // end with a question mark
                        *cell = format!("{lang:?}({lang_confidence:.3})?");
                    }
                }
            },
            Operations::Copy => {}, // copy is a noop
        }
    }
}
