run_pipeline

#!/usr/bin/env bash
set -e

# Required temporary storage
TMP_DIR=$(mktemp -d /tmp/dedup.XXXXXX)

cleanup() {
    rm -rf "${TMP_DIR}"
}

trap cleanup EXIT SIGINT SIGTERM

# Help documentation
usage() {
    echo -e "Usage: $0 [--debug] <input_glob> <output_directory>"
    echo -e "This script runs the record linkage pipeline on your local host.\n"
    echo -e "Params:"
    echo -e "  - <input_globs>\tcomma separated list of globs to deduplicate; first pipeline stage is parallelized across globs"
    echo -e "  - <output_dir>\tdirectory to save the deuplicated files to\n"
    echo -e "Example: $0 'input_dir/[a-m]*.txt,input_dir/[n-z]*.txt' output_dir/\n"
    exit 1
}

# Initialize debug mode variable
RUST_LOG="error"

# Parse command line arguments
while [[ "$#" > 0 ]]; do
    case "$1" in
        --debug)
            RUST_LOG="info"
            shift 1
            ;;
        --help)
            usage
            exit 0
            ;;
        *)
            break
            ;;
    esac
done

export RUST_LOG

# Check for correct number of arguments
if [ "$#" -ne 2 ]; then
    echo "Error: You must enter exactly two parameters."
    usage
fi

# Validate inputs
validate_glob() {
    local pattern="$1"
    local files=( $pattern )

    # Check if the expansion returns anything
    if [ ${#files[@]} -eq 0 ]; then
        echo "No files match the glob '$pattern'."
        echo "Please provide a valid glob pattern."
        return 1 # Return 1 for error state
    else
        return 0 # Return 0 for success state
    fi
}

validate_directory() {
    if [ ! -d "$1" ]; then
        echo "Error: '$1' is not a valid directory."
        exit 2
    fi
}

log() {
    echo -e "\033[32m${1}\033[0m"
}

IFS=',' read -ra INPUT_GLOBS <<< "$1"
for i in "${INPUT_GLOBS[@]}"; do
    validate_glob "$i"
done

OUTPUT_DIR="$2"
validate_directory "$OUTPUT_DIR"

# Start the pipeline
log "Hashing documents..."
for GLOB in "${INPUT_GLOBS[@]}"; do
    GLOB="${GLOB/#\~/$HOME}"
    GLOB="${GLOB/#\./$PWD}"
    ./target/release/hash_documents "$GLOB" "$TMP_DIR" &
done

wait

log "\nIntermediate hash files (first char of filename is the shard):"
ls "$TMP_DIR/"

SHARDS=($(ls "$TMP_DIR/" | sed 's/^\(.\).*/\1/' | sort -u))

log "\nDeduping files based on their hashes..."

for SHARD in "${SHARDS[@]}"; do
    ./target/release/dedup_hashes "$TMP_DIR/${SHARD}_*" "$OUTPUT_DIR/$SHARD.csv" &
done

wait

log "\nUniques (first char of the filename is the shard):"
ls $OUTPUT_DIR/*.csv

echo ""
log "Run the following command to get uniques: ls $OUTPUT_DIR/*.csv | xargs cat"