feat: add atty package and implement TTY detection in UpdateStats display

Author: ansu86d

Cargo.lock

@@ -127,6 +127,7 @@ bytes = "1.10.1"
 rand = "0.9.2"
 indoc = "2.0.6"
 owo-colors = "4.2.2"
+atty = "0.2.14"
 json5 = "0.4.1"
 aws-config = "1.8.5"
 aws-sdk-s3 = "1.102.0"

Cargo.toml

@@ -0,0 +1,23 @@
+#!/usr/bin/env python3
+"""Run main.py and capture output to test_output.txt"""
+
+import subprocess
+import sys
+
+# Run main.py and capture output
+result = subprocess.run(
+    [sys.executable, "main.py"],
+    capture_output=True,
+    text=True,
+    timeout=10
+)
+
+# Write both stdout and stderr to test_output.txt
+with open("test_output.txt", "w") as f:
+    if result.stdout:
+        f.write(result.stdout)
+    if result.stderr:
+        f.write(result.stderr)
+
+print(f"Output saved to test_output.txt")
+print(f"Return code: {result.returncode}")

examples/custom_output_files/run_and_capture.py

@@ -403,6 +403,7 @@ pub async fn update_source_tracking_ordinal(
 /// Access for the source state table
 ////////////////////////////////////////////////////////////
 
+#[allow(dead_code)]
 pub async fn read_source_state(
     source_id: i32,
     source_key_json: &serde_json::Value,
@@ -425,6 +426,7 @@ pub async fn read_source_state(
     Ok(state)
 }
 
+#[allow(dead_code)]
 pub async fn upsert_source_state(
     source_id: i32,
     source_key_json: &serde_json::Value,

src/execution/db_tracking.rs

@@ -1,10 +1,16 @@
 use crate::prelude::*;
+use owo_colors::{AnsiColors, OwoColorize};
 
 use std::{
     ops::AddAssign,
     sync::atomic::{AtomicI64, Ordering::Relaxed},
 };
 
+/// Check if stdout is a TTY
+fn is_stdout_tty() -> bool {
+    atty::is(atty::Stream::Stdout)
+}
+
 #[derive(Default, Serialize)]
 pub struct Counter(pub AtomicI64);
 
@@ -107,6 +113,10 @@ pub struct UpdateStats {
     pub num_errors: Counter,
     /// Processing counters for tracking in-process rows.
     pub processing: ProcessingCounters,
+    /// Cumulative total count of items (for display purposes)
+    /// This represents the actual total after applying additions and deletions
+    #[serde(skip)]
+    pub cumulative_total: Counter,
 }
 
 impl UpdateStats {
@@ -119,6 +129,7 @@ impl UpdateStats {
             num_reprocesses: self.num_reprocesses.delta(&base.num_reprocesses),
             num_errors: self.num_errors.delta(&base.num_errors),
             processing: self.processing.delta(&base.processing),
+            cumulative_total: self.cumulative_total.clone(),
         }
     }
 
@@ -130,6 +141,9 @@ impl UpdateStats {
         self.num_reprocesses.merge(&delta.num_reprocesses);
         self.num_errors.merge(&delta.num_errors);
         self.processing.merge(&delta.processing);
+        // Update cumulative total: add insertions, subtract deletions
+        let net_change = delta.num_insertions.get() - delta.num_deletions.get();
+        self.cumulative_total.inc(net_change);
     }
 
     pub fn has_any_change(&self) -> bool {
@@ -205,13 +219,13 @@ impl std::fmt::Display for UpdateStats {
         // Progress bar segments
         if total > 0 {
             if num_insertions > 0 {
-                segments.push((num_insertions, "+", format!("\x1B[90m(+{} added)\x1B[0m", num_insertions)));
+                segments.push((num_insertions, "+", format!("(+{} added)", num_insertions)));
             }
             if num_deletions > 0 {
-                segments.push((num_deletions, "-", format!("\x1B[90m(-{} removed)\x1B[0m", num_deletions)));
+                segments.push((num_deletions, "-", format!("(-{} removed)", num_deletions)));
             }
             if num_updates > 0 {
-                segments.push((num_updates, "~", format!("\x1B[90m(~{} updated)\x1B[0m", num_updates)));
+                segments.push((num_updates, "~", format!("(~{} updated)", num_updates)));
             }
             if num_no_change > 0 {
                 segments.push((num_no_change, " ", "".to_string()));
@@ -220,7 +234,12 @@ impl std::fmt::Display for UpdateStats {
 
         // Error handling
         if num_errors > 0 {
-            write!(f, "{} rows failed", num_errors)?;
+            let tty = is_stdout_tty();
+            if tty {
+                write!(f, "{}", format!("{} rows failed", num_errors).color(AnsiColors::White))?;
+            } else {
+                write!(f, "{} rows failed", num_errors)?;
+            }
             if !segments.is_empty() {
                 write!(f, "; ")?;
             }
@@ -235,7 +254,7 @@ impl std::fmt::Display for UpdateStats {
             let bar_width = 40;
             let mut bar = String::new();
 
-            let percentage = ((total - num_in_process) as f64 / total as f64 * 100.0) as i64;
+            let _percentage = ((total - num_in_process) as f64 / total as f64 * 100.0) as i64;
             let mut remaining_width = bar_width;
 
             for (count, segment_type, _) in sorted_segments.iter() {
@@ -273,29 +292,49 @@ impl std::fmt::Display for UpdateStats {
             if remaining_width > 0 {
                 bar.push_str(&" ".repeat(remaining_width));
             }
-            write!(f, "[{}] {}/{} records ", bar, total - num_in_process, total)?;
+            let tty = is_stdout_tty();
+            // Use total from current operations - this represents the actual record count
+            if tty {
+                write!(f, "[{}] {}/{} records ", bar.color(AnsiColors::BrightBlack), total - num_in_process, total)?;
+            } else {
+                write!(f, "[{}] {}/{} records ", bar, total - num_in_process, total)?;
+            }
 
-            // Add segment labels
+            // Add segment labels with different grey shades for each segment type
             let mut first = true;
-            for (_, _, label) in segments.iter() {
+            for (_, segment_type, label) in segments.iter() {
                 if !label.is_empty() {
                     if !first {
                         write!(f, " ")?;
                     }
-                    write!(f, "{}", label)?;
+                    if tty {
+                        match *segment_type {
+                            "+" => write!(f, "{}", label.color(AnsiColors::BrightBlack))?,     // Lightest grey for additions
+                            "-" => write!(f, "{}", label.color(AnsiColors::White))?,           // White for removals
+                            "~" => write!(f, "{}", label.color(AnsiColors::Black))?,           // Dark grey for updates
+                            _ => write!(f, "{}", label.color(AnsiColors::Black))?,             // Black for no-change
+                        }
+                    } else {
+                        write!(f, "{}", label)?;
+                    }
                     first = false;
                 }
             }
         } else {
             write!(f, "No changes")?;
         }
 
-        // In-process info
+        // In-process info with grey coloring
         if num_in_process > 0 {
             if !segments.is_empty() {
                 write!(f, " ")?;
             }
-            write!(f, "({} in process)", num_in_process)?;
+            let tty = is_stdout_tty();
+            if tty {
+                write!(f, "{}", format!("({} in process)", num_in_process).color(AnsiColors::Black))?;
+            } else {
+                write!(f, "({} in process)", num_in_process)?;
+            }
         }
 
         Ok(())
@@ -567,17 +606,19 @@ mod tests {
         // Test with no activity
         assert_eq!(format!("{}", stats), "No changes");
 
-        // Test with in-process rows
+        // Test with in-process rows (no segments yet, so just shows in-process)
         stats.processing.start(5);
-        assert!(format!("{}", stats).contains("5 source rows IN PROCESS"));
+        let display = format!("{}", stats);
+        assert!(display.contains("5 in process"));
 
         // Test with mixed activity
         stats.num_insertions.inc(3);
         stats.num_errors.inc(1);
+        stats.cumulative_total.inc(3);
         let display = format!("{}", stats);
-        assert!(display.contains("1 source rows FAILED"));
-        assert!(display.contains("3 source rows processed"));
-        assert!(display.contains("5 source rows IN PROCESS"));
+        assert!(display.contains("1 rows failed"));
+        assert!(display.contains("(+3 added)"));
+        assert!(display.contains("5 in process"));
     }
 
     #[test]

src/execution/stats.rs

@@ -72,6 +72,7 @@ pub struct ResolvedOpArg {
 
 pub trait ResolvedOpArgExt: Sized {
     fn value<'a>(&self, args: &'a [value::Value]) -> Result<&'a value::Value>;
+    #[allow(dead_code)]
     fn take_value(&self, args: &mut [value::Value]) -> Result<value::Value>;
 }