refactor: remove dead code, deprecated functions, and consolidate duplicates

Author: r0BIT

taskhound/cli.py

@@ -21,7 +21,6 @@
 )
 from .opengraph import generate_opengraph_files
 from .output.bloodhound import upload_opengraph_to_bloodhound
-from .output.printer import print_results
 from .output.summary import print_decrypted_credentials, print_summary_table
 from .output.writer import write_csv, write_json, write_rich_plain
 from .parsers.highvalue import HighValueLoader
@@ -175,7 +174,6 @@ def main():
             dpapi_key=args.dpapi_key,
             concise=not args.verbose,
         )
-        print_results(lines)
     else:
         # Online mode: process targets via SMB
         # Build targets list
@@ -246,13 +244,6 @@ def main():
             # Aggregate results
             all_rows, laps_failures, laps_successes = aggregate_results(results)
 
-            # Print results (already captured in results)
-            for result in results:
-                if result.lines:
-                    print_results(result.lines)
-
-            # Summary already printed by async engine's Rich progress bar
-
         else:
             # Sequential mode (default, --threads 1)
             for tgt in targets:
@@ -267,7 +258,6 @@ def main():
                         laps_successes += 1
                     elif isinstance(laps_result, LAPSFailure):
                         laps_failures.append(laps_result)
-                print_results(lines)
 
     # Exports
     # Auto-generate JSON if OpenGraph is enabled and no explicit JSON output was specified

taskhound/output/__init__.py

@@ -1 +1,17 @@
 """Output module for TaskHound results."""
+
+# Shared color scheme for task output
+COLORS = {
+    "tier0_header": "bold red",
+    "tier0_border": "red",
+    "priv_header": "bold yellow",
+    "priv_border": "yellow",
+    "task_header": "bold green",
+    "task_border": "green",
+    "label": "dim",
+    "value": "white",
+    "password": "bold green",
+    "warning": "yellow",
+    "error": "red",
+    "success": "green",
+}

taskhound/output/printer.py

@@ -1,45 +1,14 @@
-import re
 from typing import Any, Dict, List, Optional
 
 from rich.table import Table
 
+from . import COLORS
 from ..parsers.highvalue import HighValueLoader
 from ..utils import logging as log_utils
 from ..utils.console import console
 from ..utils.date_parser import parse_iso_date
 from ..utils.sid_resolver import format_runas_with_sid_resolution
 
-# Color scheme for task output
-COLORS = {
-    "tier0_header": "bold red",
-    "tier0_border": "red",
-    "priv_header": "bold yellow",
-    "priv_border": "yellow",
-    "task_header": "bold green",
-    "task_border": "green",
-    "label": "dim",
-    "value": "white",
-    "password": "bold green",
-    "warning": "yellow",
-    "error": "red",
-    "success": "green",
-}
-
-
-def print_results(lines: List[str]):
-    """
-    Legacy print function - no longer prints to console.
-
-    Tables are now printed directly by format_block() via print_task_table().
-    This function exists for backward compatibility but does nothing since
-    the Rich tables are already printed when format_block is called.
-
-    The text lines are still used for file output (--plain flag).
-    """
-    # Tables are already printed by print_task_table() in format_block()
-    # This function is kept for API compatibility but no longer prints
-    pass
-
 
 def print_task_table(
     kind: str,

taskhound/output/writer.py

@@ -8,21 +8,9 @@
 from rich.console import Console
 from rich.table import Table
 
+from . import COLORS
 from ..utils.logging import good
 
-# Colors for task types (matching printer.py)
-COLORS = {
-    "tier0_header": "bold red",
-    "tier0_border": "red",
-    "priv_header": "bold yellow",
-    "priv_border": "yellow",
-    "task_header": "bold green",
-    "task_border": "green",
-    "label": "cyan",
-    "value": "white",
-    "password": "bold green",
-}
-
 
 def _rows_to_dicts(rows: List[Any]) -> List[Dict]:
     """Convert TaskRow objects to dicts for serialization."""
@@ -123,21 +111,6 @@ def _format_task_table(row_dict: Dict[str, Any]) -> Table:
     return table
 
 
-def write_plain(outdir: str, host: str, lines: List[str]):
-    """
-    Legacy plain text writer - writes raw text lines.
-
-    .. deprecated::
-        Use write_rich_plain() instead for Rich-formatted output with tables.
-    """
-    os.makedirs(outdir, exist_ok=True)
-    safe = host.replace(":", "_")
-    path = os.path.join(outdir, f"{safe}.txt")
-    with open(path, "w", encoding="utf-8") as f:
-        f.write("\n".join(lines) + ("\n" if lines else ""))
-    good(f"Wrote results to {path}")
-
-
 def write_rich_plain(outdir: str, all_rows: List[Any], force_color: bool = True):
     """
     Write Rich-formatted output files grouped by host.

taskhound/utils/console.py

@@ -198,13 +198,6 @@ def update(item: str, success: bool = True, error_msg: Optional[str] = None):
             )
 
 
-def update_progress_status(status_text: str):
-    """Update the progress bar status text (for external use)."""
-    global _progress, _progress_task_id
-    if _progress and _progress_task_id is not None:
-        _progress.update(_progress_task_id, status=status_text)
-
-
 @contextmanager
 def spinner(description: str = "Processing"):
     """
@@ -238,36 +231,6 @@ def spinner(description: str = "Processing"):
         pass  # Spinner removed automatically (transient=True)
 
 
-# =============================================================================
-# Collecting Status (for per-target output during scanning)
-# =============================================================================
-
-def collecting_start(target: str):
-    """Show that we're starting to collect from a target."""
-    console.print(f"[dim][Collecting][/] {target} [dim]...[/]")
-
-
-def collecting_done(target: str, task_count: int, priv_count: int):
-    """Show successful collection from a target."""
-    console.print(
-        f"[dim][Collecting][/] {target} [green][+][/] "
-        f"[dim]({task_count} tasks, {priv_count} privileged)[/]"
-    )
-
-
-def collecting_skip(target: str, reason: str):
-    """Show that a target was skipped."""
-    console.print(f"[dim][Collecting][/] {target} [yellow][SKIP][/] [dim]({reason})[/]")
-
-
-def collecting_fail(target: str, error_msg: str):
-    """Show that collection from a target failed."""
-    # Truncate long error messages
-    if len(error_msg) > 60:
-        error_msg = error_msg[:57] + "..."
-    console.print(f"[dim][Collecting][/] {target} [red][-][/] [dim]({error_msg})[/]")
-
-
 # =============================================================================
 # Summary Table
 # =============================================================================

tests/test_console_utils.py

@@ -295,57 +295,3 @@ def test_tier0_takes_precedence_over_priv(self):
         assert "[PRIV]" not in result
 
 
-# ============================================================================
-# Test: collecting_* functions
-# ============================================================================
-
-class TestCollectingFunctions:
-    """Tests for collecting status functions."""
-    
-    @patch.object(console, 'console')
-    def test_collecting_done(self, mock_console):
-        """Should print success message with counts."""
-        from taskhound.utils.console import collecting_done
-        
-        collecting_done("DC01", task_count=10, priv_count=3)
-        mock_console.print.assert_called_once()
-        call_arg = mock_console.print.call_args[0][0]
-        assert "DC01" in call_arg
-        assert "10 tasks" in call_arg
-        assert "3 privileged" in call_arg
-    
-    @patch.object(console, 'console')
-    def test_collecting_skip(self, mock_console):
-        """Should print skip message with reason."""
-        from taskhound.utils.console import collecting_skip
-        
-        collecting_skip("DC02", "Access denied")
-        mock_console.print.assert_called_once()
-        call_arg = mock_console.print.call_args[0][0]
-        assert "DC02" in call_arg
-        assert "SKIP" in call_arg
-        assert "Access denied" in call_arg
-    
-    @patch.object(console, 'console')
-    def test_collecting_fail(self, mock_console):
-        """Should print failure message."""
-        from taskhound.utils.console import collecting_fail
-        
-        collecting_fail("DC03", "Connection timeout")
-        mock_console.print.assert_called_once()
-        call_arg = mock_console.print.call_args[0][0]
-        assert "DC03" in call_arg
-        assert "Connection timeout" in call_arg
-    
-    @patch.object(console, 'console')
-    def test_collecting_fail_truncates_long_error(self, mock_console):
-        """Should truncate error messages over 60 characters."""
-        from taskhound.utils.console import collecting_fail
-        
-        long_error = "This is a very long error message that exceeds the sixty character limit significantly"
-        collecting_fail("DC04", long_error)
-        mock_console.print.assert_called_once()
-        call_arg = mock_console.print.call_args[0][0]
-        assert "..." in call_arg
-        assert long_error not in call_arg
-

tests/test_output_writer.py

@@ -3,9 +3,9 @@
 
 Tests cover:
 - _rows_to_dicts helper function
-- write_plain function
 - write_json function
 - write_csv function
+- write_rich_plain function
 """
 
 import csv
@@ -14,7 +14,7 @@
 import pytest
 from unittest.mock import MagicMock, patch
 
-from taskhound.output.writer import _rows_to_dicts, write_plain, write_json, write_csv
+from taskhound.output.writer import _rows_to_dicts, write_json, write_csv
 
 
 # ============================================================================
@@ -91,65 +91,6 @@ def test_mixed_objects_and_dicts(self, sample_task_dict, sample_task_object):
         assert result[1] == {"host": "WS01.example.com", "path": "\\MaintTask", "type": "TASK"}
 
 
-# ============================================================================
-# Unit Tests: write_plain
-# ============================================================================
-
-
-class TestWritePlain:
-    """Tests for write_plain function"""
-
-    @patch('taskhound.output.writer.good')
-    def test_creates_output_directory(self, mock_good, temp_output_dir):
-        """Should create output directory if it doesn't exist"""
-        subdir = temp_output_dir / "subdir"
-        lines = ["Line 1", "Line 2"]
-        
-        write_plain(str(subdir), "host1", lines)
-        
-        assert subdir.exists()
-
-    @patch('taskhound.output.writer.good')
-    def test_writes_lines_to_file(self, mock_good, temp_output_dir):
-        """Should write lines to file with newlines"""
-        lines = ["Line 1", "Line 2", "Line 3"]
-        
-        write_plain(str(temp_output_dir), "host1.example.com", lines)
-        
-        output_file = temp_output_dir / "host1.example.com.txt"
-        assert output_file.exists()
-        
-        content = output_file.read_text()
-        assert content == "Line 1\nLine 2\nLine 3\n"
-
-    @patch('taskhound.output.writer.good')
-    def test_empty_lines_creates_file(self, mock_good, temp_output_dir):
-        """Should create file even with empty lines"""
-        write_plain(str(temp_output_dir), "host1", [])
-        
-        output_file = temp_output_dir / "host1.txt"
-        assert output_file.exists()
-        content = output_file.read_text()
-        assert content == ""
-
-    @patch('taskhound.output.writer.good')
-    def test_colon_replaced_in_filename(self, mock_good, temp_output_dir):
-        """Should replace colons with underscores in filename"""
-        write_plain(str(temp_output_dir), "host:port", ["test"])
-        
-        output_file = temp_output_dir / "host_port.txt"
-        assert output_file.exists()
-
-    @patch('taskhound.output.writer.good')
-    def test_logs_success_message(self, mock_good, temp_output_dir):
-        """Should log success message"""
-        write_plain(str(temp_output_dir), "host1", ["test"])
-        
-        mock_good.assert_called_once()
-        call_arg = mock_good.call_args[0][0]
-        assert "Wrote results to" in call_arg
-
-
 # ============================================================================
 # Unit Tests: write_json
 # ============================================================================