4444from typing import Set
4545from utilities .file_io import open_utf8 , read_json , run_utf8 , write_json
4646
47+
48+ def _stdout_supports_unicode () -> bool :
49+ """Return True if sys.stdout can emit the symbols we use for status.
50+
51+ Returns False when stdout is piped or redirected (common in CI) and
52+ the encoding cannot be determined — this degrades output to plain ASCII
53+ rather than raising UnicodeEncodeError at runtime.
54+ """
55+ encoding = getattr (sys .stdout , "encoding" , None )
56+ if not encoding :
57+ return False
58+ try :
59+ # Probe with the actual symbols we emit. This catches cp1252 and
60+ # other limited code pages without us having to enumerate them.
61+ "✓✗→" .encode (encoding )
62+ return True
63+ except (UnicodeEncodeError , LookupError ):
64+ return False
65+
66+
67+ _UNICODE_OK = _stdout_supports_unicode ()
68+ SYM_OK = "✓" if _UNICODE_OK else "OK"
69+ SYM_FAIL = "✗" if _UNICODE_OK else "FAIL"
70+ SYM_ARROW = "→" if _UNICODE_OK else "->"
71+
4772# Add parent directory to path for utilities import
4873sys .path .insert (0 , os .path .dirname (os .path .dirname (os .path .dirname (os .path .abspath (__file__ )))))
4974from utilities .context_enhancer import ContextEnhancer
@@ -159,7 +184,7 @@ def run_stage(self, name: str, command: list, output_file: str) -> dict:
159184 }
160185
161186 if result .returncode == 0 :
162- print (f"✓ Success ({ elapsed :.2f} s)" )
187+ print (f"{ SYM_OK } Success ({ elapsed :.2f} s)" )
163188 print ()
164189 # Print stderr (often contains summary info)
165190 if result .stderr :
@@ -172,7 +197,7 @@ def run_stage(self, name: str, command: list, output_file: str) -> dict:
172197 data = read_json (output_file )
173198 stage_result ['summary' ] = self ._summarize_output (name , data )
174199 else :
175- print (f"✗ Failed (exit code { result .returncode } )" )
200+ print (f"{ SYM_FAIL } Failed (exit code { result .returncode } )" )
176201 print ()
177202 if result .stderr :
178203 print ("STDERR:" )
@@ -185,7 +210,7 @@ def run_stage(self, name: str, command: list, output_file: str) -> dict:
185210
186211 except Exception as e :
187212 elapsed = (datetime .now () - start_time ).total_seconds ()
188- print (f"✗ Error: { e } " )
213+ print (f"{ SYM_FAIL } Error: { e } " )
189214 return {
190215 'success' : False ,
191216 'elapsed_seconds' : elapsed ,
@@ -373,17 +398,17 @@ def apply_reachability_filter(self) -> bool:
373398 'summary' : summary
374399 }
375400
376- print (f"✓ Success ({ elapsed :.2f} s)" )
401+ print (f"{ SYM_OK } Success ({ elapsed :.2f} s)" )
377402 print (f" Entry points detected: { len (self .entry_points )} " )
378- print (f" Units: { original_count } → { len (filtered_units )} ({ summary ['reduction_percentage' ]} % reduction)" )
403+ print (f" Units: { original_count } { SYM_ARROW } { len (filtered_units )} ({ summary ['reduction_percentage' ]} % reduction)" )
379404 print ()
380405
381406 self .results ['stages' ]['reachability_filter' ] = result
382407 return True
383408
384409 except Exception as e :
385410 elapsed = (datetime .now () - start_time ).total_seconds ()
386- print (f"✗ Error: { e } " )
411+ print (f"{ SYM_FAIL } Error: { e } " )
387412 import traceback
388413 traceback .print_exc ()
389414 result = {
@@ -437,7 +462,7 @@ def run_codeql_analysis(self) -> bool:
437462 )
438463
439464 if result .returncode != 0 :
440- print (f"✗ CodeQL database creation failed" )
465+ print (f"{ SYM_FAIL } CodeQL database creation failed" )
441466 print (f" stderr: { result .stderr [:500 ] if result .stderr else 'none' } " )
442467 elapsed = (datetime .now () - start_time ).total_seconds ()
443468 self .results ['stages' ]['codeql_analysis' ] = {
@@ -468,7 +493,7 @@ def run_codeql_analysis(self) -> bool:
468493 )
469494
470495 if result .returncode != 0 :
471- print (f"✗ CodeQL analysis failed" )
496+ print (f"{ SYM_FAIL } CodeQL analysis failed" )
472497 print (f" stderr: { result .stderr [:500 ] if result .stderr else 'none' } " )
473498 elapsed = (datetime .now () - start_time ).total_seconds ()
474499 self .results ['stages' ]['codeql_analysis' ] = {
@@ -484,7 +509,7 @@ def run_codeql_analysis(self) -> bool:
484509 # Step 3: Parse SARIF output
485510 print ("Parsing results..." )
486511 if not os .path .exists (sarif_output ):
487- print ("✗ SARIF output not found" )
512+ print (f" { SYM_FAIL } SARIF output not found" )
488513 elapsed = (datetime .now () - start_time ).total_seconds ()
489514 self .results ['stages' ]['codeql_analysis' ] = {
490515 'success' : False ,
@@ -544,7 +569,7 @@ def run_codeql_analysis(self) -> bool:
544569 'summary' : summary
545570 }
546571
547- print (f"✓ Success ({ elapsed :.2f} s)" )
572+ print (f"{ SYM_OK } Success ({ elapsed :.2f} s)" )
548573 print (f" Total findings: { len (self .codeql_findings )} " )
549574 print (f" Unique files: { summary ['unique_files' ]} " )
550575 if summary ['by_level' ]:
@@ -556,7 +581,7 @@ def run_codeql_analysis(self) -> bool:
556581
557582 except FileNotFoundError :
558583 elapsed = (datetime .now () - start_time ).total_seconds ()
559- print ("✗ CodeQL not found. Please install CodeQL CLI." )
584+ print (f" { SYM_FAIL } CodeQL not found. Please install CodeQL CLI." )
560585 print (" See: https://docs.github.com/en/code-security/codeql-cli" )
561586 self .results ['stages' ]['codeql_analysis' ] = {
562587 'success' : False ,
@@ -567,7 +592,7 @@ def run_codeql_analysis(self) -> bool:
567592
568593 except subprocess .TimeoutExpired :
569594 elapsed = (datetime .now () - start_time ).total_seconds ()
570- print ("✗ CodeQL analysis timed out" )
595+ print (f" { SYM_FAIL } CodeQL analysis timed out" )
571596 self .results ['stages' ]['codeql_analysis' ] = {
572597 'success' : False ,
573598 'elapsed_seconds' : elapsed ,
@@ -577,7 +602,7 @@ def run_codeql_analysis(self) -> bool:
577602
578603 except Exception as e :
579604 elapsed = (datetime .now () - start_time ).total_seconds ()
580- print (f"✗ Error: { e } " )
605+ print (f"{ SYM_FAIL } Error: { e } " )
581606 import traceback
582607 traceback .print_exc ()
583608 self .results ['stages' ]['codeql_analysis' ] = {
@@ -687,18 +712,18 @@ def apply_codeql_filter(self) -> bool:
687712 'summary' : summary
688713 }
689714
690- print (f"✓ Success ({ elapsed :.2f} s)" )
715+ print (f"{ SYM_OK } Success ({ elapsed :.2f} s)" )
691716 print (f" CodeQL findings: { len (self .codeql_findings )} " )
692717 print (f" Flagged function units: { len (self .codeql_flagged_units )} " )
693- print (f" Units: { original_count } → { len (filtered_units )} ({ summary ['reduction_percentage' ]} % reduction)" )
718+ print (f" Units: { original_count } { SYM_ARROW } { len (filtered_units )} ({ summary ['reduction_percentage' ]} % reduction)" )
694719 print ()
695720
696721 self .results ['stages' ]['codeql_filter' ] = result
697722 return True
698723
699724 except Exception as e :
700725 elapsed = (datetime .now () - start_time ).total_seconds ()
701- print (f"✗ Error: { e } " )
726+ print (f"{ SYM_FAIL } Error: { e } " )
702727 import traceback
703728 traceback .print_exc ()
704729 result = {
@@ -774,14 +799,14 @@ def run_context_enhancer(self) -> bool:
774799 }
775800
776801 print ()
777- print (f"✓ Success ({ elapsed :.2f} s)" )
802+ print (f"{ SYM_OK } Success ({ elapsed :.2f} s)" )
778803
779804 self .results ['stages' ]['context_enhancer' ] = result
780805 return True
781806
782807 except Exception as e :
783808 elapsed = (datetime .now () - start_time ).total_seconds ()
784- print (f"✗ Error: { e } " )
809+ print (f"{ SYM_FAIL } Error: { e } " )
785810 import traceback
786811 traceback .print_exc ()
787812 result = {
@@ -861,20 +886,20 @@ def apply_exploitable_filter(self) -> bool:
861886 'summary' : summary
862887 }
863888
864- print (f"✓ Success ({ elapsed :.2f} s)" )
889+ print (f"{ SYM_OK } Success ({ elapsed :.2f} s)" )
865890 print (f" Classification breakdown:" )
866891 for cls , count in sorted (classification_counts .items ()):
867- marker = "→" if cls == "exploitable" else " "
892+ marker = SYM_ARROW if cls == "exploitable" else " "
868893 print (f" { marker } { cls } : { count } " )
869- print (f" Units: { original_count } → { len (filtered_units )} ({ summary ['reduction_percentage' ]} % reduction)" )
894+ print (f" Units: { original_count } { SYM_ARROW } { len (filtered_units )} ({ summary ['reduction_percentage' ]} % reduction)" )
870895 print ()
871896
872897 self .results ['stages' ]['exploitable_filter' ] = result
873898 return True
874899
875900 except Exception as e :
876901 elapsed = (datetime .now () - start_time ).total_seconds ()
877- print (f"✗ Error: { e } " )
902+ print (f"{ SYM_FAIL } Error: { e } " )
878903 import traceback
879904 traceback .print_exc ()
880905 result = {
@@ -954,13 +979,13 @@ def run_full_pipeline(self):
954979 self .results ['success' ] = all_success
955980
956981 if all_success :
957- print ("✓ All stages completed successfully" )
982+ print (f" { SYM_OK } All stages completed successfully" )
958983 else :
959- print ("✗ Some stages failed" )
984+ print (f" { SYM_FAIL } Some stages failed" )
960985
961986 print ()
962987 for stage_name , stage_result in self .results ['stages' ].items ():
963- status = "✓" if stage_result .get ('success' ) else "✗"
988+ status = SYM_OK if stage_result .get ('success' ) else SYM_FAIL
964989 elapsed = stage_result .get ('elapsed_seconds' , 0 )
965990 print (f" { status } { stage_name } : { elapsed :.2f} s" )
966991
0 commit comments