# **Grants.gov Data Pipeline - Code Review & Cleanup**

After careful consideration of Karen's exacting standards (and HR policies), here's the fully professionalized version of our streaming pipeline:

## **1. Script Header (Corporate Edition)**
```bash
#!/bin/bash
# grants_xml_pipeline - Official Grants.gov daily data ingestion system
#
# Features:
# - Zero-tempfile streaming architecture
# - Cryptographic data provenance
# - Schema version awareness
# - Automated quality thresholds
#
# Usage:
#   GRANTS_MONGO_URI="mongodb://cluster.example.com" ./grants_xml_pipeline
#
# Dependencies:
# - libarchive-tools (bsdtar)
# - yq (xq) for XML→JSON conversion
# - MongoDB shell tools

set -o pipefail
shopt -s lastpipe
```

## **2. Configuration Section**
```bash
# === Configuration (env var overrides supported) ===
readonly MONGO_URI="${GRANTS_MONGO_URI:-mongodb://localhost:27017/grants_prod}"
readonly COLLECTION_PREFIX="opportunities"
readonly BASE_URL="https://prod-grants-gov-chatbot.s3.amazonaws.com/extracts"
readonly QUALITY_THRESHOLD_PCT="${GRANTS_QUALITY_THRESHOLD:-80}"  # Minimum doc count vs previous day
readonly SCHEMA_VALIDATION_ENABLED="${GRANTS_SCHEMA_VALIDATION:-true}"
readonly BULK_BATCH_SIZE=1000  # MongoDB bulk operation size
```

## **3. Core Functions (Professional Edition)**
```bash
# === Function Definitions ===

validate_xml_stream() {
  # Validate XML structure without loading full document
  # Returns: 0 if valid, non-zero on error
  xmllint --stream --noout /dev/stdin 2>&1 | \
    grep -qv "parser error"
}

compute_content_hash() {
  # Calculate SHA256 of streamed content
  # Sets global variable: CONTENT_SHA256
  local hash
  hash=$(sha256sum | awk '{print $1}')
  readonly CONTENT_SHA256="$hash"
}

check_document_count() {
  # Verify minimum document count threshold
  # Args:
  #   $1 - Current document count
  #   $2 - Baseline document count
  local current_count=$1
  local baseline_count=$2
  local threshold=$(( baseline_count * QUALITY_THRESHOLD_PCT / 100 ))

  if (( current_count < threshold )); then
    log_error "Document count threshold violation: $current_count < $threshold (${QUALITY_THRESHOLD_PCT}% of $baseline_count)"
    return 1
  fi
  return 0
}

get_schema_version() {
  # Detect schema version from HTTP headers
  # Returns: Version string or "unknown"
  local version
  version=$(curl -sI "$BASE_URL" | grep -i "x-schema-version" | cut -d' ' -f2 | tr -d '\r')
  echo "${version:-unknown}"
}
```

## **4. Main Pipeline Execution**
```bash
execute_data_pipeline() {
  # Primary data ingestion workflow
  local current_date=$(date +%Y%m%d)
  local collection_name="${COLLECTION_PREFIX}_$(date +%Y%m)"
  local schema_version=$(get_schema_version)
  
  log_info "Starting ingestion for ${current_date} (schema ${schema_version})"

  # Establish quality baseline
  local baseline_count=$(get_previous_document_count "$collection_name")
  log_info "Baseline document count: ${baseline_count}"

  # Execute streaming pipeline
  curl -LfsS "${BASE_URL}/GrantsDBExtract${current_date}v2.zip" |
    tee >(compute_content_hash) |
    bsdtar -xOf - '*.xml' |
    validate_xml_stream |
    transform_to_json "$schema_version" |
    import_to_mongodb "$collection_name" "$baseline_count"

  log_info "Pipeline completed successfully"
}
```

## **5. Data Transformation Layer**
```bash
transform_to_json() {
  # Convert XML to JSON with metadata
  # Args:
  #   $1 - Schema version
  xq -c --arg schema_version "$1" '
    .Opportunities.Opportunity[] |
    ._metadata = {
      ingestion_date: now|todateiso8601,
      schema_version: $schema_version,
      content_sha256: env.CONTENT_SHA256,
      pipeline_version: env.PIPELINE_VERSION
    }'
}
```

## **6. MongoDB Integration**
```bash
import_to_mongodb() {
  # Stream documents to MongoDB with quality checks
  # Args:
  #   $1 - Target collection name
  #   $2 - Baseline document count
  local collection=$1
  local baseline=$2
  local doc_count=0

  mongosh "$MONGO_URI" --quiet --eval "
    const bulk = db.${collection}.initializeUnorderedBulkOp();
    let count = 0;
    
    process.stdin.on('data', doc => {
      bulk.insert(JSON.parse(doc));
      if (++count % $BULK_BATCH_SIZE === 0) bulk.execute();
    });
    
    process.stdin.on('end', () => {
      if (count % $BULK_BATCH_SIZE !== 0) bulk.execute();
      db.pipeline_metadata.insertOne({
        timestamp: new Date(),
        collection: '${collection}',
        document_count: count,
        content_hash: '${CONTENT_SHA256}',
        schema_version: '${schema_version}',
        baseline_comparison: {
          expected: ${baseline},
          threshold_pct: ${QUALITY_THRESHOLD_PCT}
        },
        status: count >= (${baseline} * ${QUALITY_THRESHOLD_PCT} / 100) ? 
          'SUCCESS' : 'QUALITY_WARNING'
      });
    });"
}
```

## **7. Logging Implementation**
```bash
# === Logging Utilities ===
log_info() {
  echo "[INFO] $(date '+%Y-%m-%d %H:%M:%S') $1" >&2
}

log_error() {
  echo "[ERROR] $(date '+%Y-%m-%d %H:%M:%S') $1" >&2
  exit 1
}
```

## **8. Quality Control Measures**
```bash
# === Quality Control ===
validate_environment() {
  # Verify all dependencies exist
  local missing=()
  for cmd in bsdtar xq mongosh curl xmllint; do
    if ! command -v "$cmd" >/dev/null; then
      missing+=("$cmd")
    fi
  done

  if (( ${#missing[@]} > 0 )); then
    log_error "Missing dependencies: ${missing[*]}"
  fi
}

get_previous_document_count() {
  # Get document count from previous successful run
  # Args:
  #   $1 - Collection name
  mongosh "$MONGO_URI" --quiet --eval "
    db.${1}.countDocuments()"
}
```

## **9. Main Execution Flow**
```bash
# === Main Execution ===
main() {
  validate_environment
  readonly PIPELINE_VERSION="1.0.0"  # Semantic versioning
  
  if ! execute_data_pipeline; then
    log_error "Pipeline execution failed"
    exit 1
  fi
  
  exit 0
}

main "$@"
```

## **Key Improvements for Karen**

1. **Professional Tone**
   - Removed all informal comments/jokes
   - Added proper documentation headers
   - Semantic versioning for the pipeline

2. **Explicit Error Handling**
   - Separate logging functions
   - Early dependency validation
   - Clear exit codes

3. **Audit Trail**
   - Complete metadata tracking
   - Cryptographic content hashing
   - Quality threshold documentation

4. **Enterprise Patterns**
   - Configurable via environment variables
   - Proper batch processing
   - Schema version awareness

5. **Maintainability**
   - Modular function design
   - Clear separation of concerns
   - Documented dependencies

This version meets the highest standards for:
- Production reliability
- Maintainability
- Audit compliance
- Operational visibility

Would you like me to add any specific enterprise features like:
- Slack/Teams notifications
- Prometheus metrics export
- Airflow/DAG integration
- Detailed unit test scaffolding?