<s> '''<STR_LIT>''' <EOL> from collections import defaultdict <EOL> from framework . dependency_management . dependency_resolver import BaseComponent <EOL> from framework . dependency_management . interfaces import ShellInterface <EOL> from framework . lib . general import * <EOL> from framework . lib . formatters import LOG_LEVEL_TOOL <EOL> import signal <EOL> import subprocess <EOL> import os <EOL> import sys <EOL> import logging <EOL> class Shell ( BaseComponent , ShellInterface ) : <EOL> COMPONENT_NAME = "<STR_LIT>" <EOL> def __init__ ( self ) : <EOL> self . register_in_service_locator ( ) <EOL> self . DynamicReplacements = { } <EOL> self . command_register = self . get_component ( "<STR_LIT>" ) <EOL> self . target = self . get_component ( "<STR_LIT:target>" ) <EOL> self . error_handler = self . get_component ( "<STR_LIT>" ) <EOL> self . timer = self . get_component ( "<STR_LIT>" ) <EOL> self . CommandTimeOffset = '<STR_LIT>' <EOL> self . OldCommands = defaultdict ( list ) <EOL> self . ShellEnviron = os . environ . copy ( ) <EOL> def RefreshReplacements ( self ) : <EOL> self . DynamicReplacements [ '<STR_LIT>' ] = self . target . GetPath ( '<STR_LIT>' ) <EOL> def StartCommand ( self , OriginalCommand , ModifiedCommand ) : <EOL> if OriginalCommand == ModifiedCommand and ModifiedCommand in self . OldCommands : <EOL> OriginalCommand = self . OldCommands [ ModifiedCommand ] <EOL> self . timer . start_timer ( self . CommandTimeOffset ) <EOL> return { '<STR_LIT>' : OriginalCommand , '<STR_LIT>' : ModifiedCommand , '<STR_LIT>' : self . timer . get_start_date_time ( self . CommandTimeOffset ) } <EOL> def FinishCommand ( self , CommandInfo , WasCancelled , PluginInfo ) : <EOL> CommandInfo [ '<STR_LIT>' ] = self . timer . get_end_date_time ( self . CommandTimeOffset ) <EOL> Success = True <EOL> if WasCancelled : <EOL> Success = False <EOL> CommandInfo [ '<STR_LIT>' ] = Success <EOL> CommandInfo [ '<STR_LIT>' ] = self . timer . get_elapsed_time_as_str ( self . CommandTimeOffset ) <EOL> CommandInfo [ '<STR_LIT>' ] = self . target . GetTargetID ( ) <EOL> CommandInfo [ '<STR_LIT>' ] = PluginInfo [ "<STR_LIT:key>" ] <EOL> self . command_register . AddCommand ( CommandInfo ) <EOL> def ShellPathEscape ( self , Text ) : <EOL> return MultipleReplace ( Text , { '<STR_LIT:U+0020>' : '<STR_LIT>' , '<STR_LIT:(>' : '<STR_LIT>' , '<STR_LIT:)>' : '<STR_LIT>' } ) <EOL> def GetModifiedShellCommand ( self , Command , PluginOutputDir ) : <EOL> self . RefreshReplacements ( ) <EOL> NewCommand = "<STR_LIT>" + self . ShellPathEscape ( PluginOutputDir ) + "<STR_LIT>" + MultipleReplace ( Command , <EOL> self . DynamicReplacements ) <EOL> self . OldCommands [ NewCommand ] = Command <EOL> return NewCommand <EOL> def CanRunCommand ( self , Command ) : <EOL> Target = self . command_register . CommandAlreadyRegistered ( Command [ '<STR_LIT>' ] ) <EOL> if Target : <EOL> return [ Target , False ] <EOL> return [ None , True ] <EOL> def create_subprocess ( self , Command ) : <EOL> proc = subprocess . Popen ( <EOL> Command , <EOL> shell = True , <EOL> env = self . ShellEnviron , <EOL> preexec_fn = os . setsid , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . STDOUT , <EOL> bufsize = <NUM_LIT:1> <EOL> ) <EOL> return proc <EOL> def shell_exec_monitor ( self , Command , PluginInfo ) : <EOL> CommandInfo = self . StartCommand ( Command , Command ) <EOL> Target , CanRun = self . CanRunCommand ( CommandInfo ) <EOL> if not CanRun : <EOL> Message = "<STR_LIT>" + str ( Target ) <EOL> return Message <EOL> logging . info ( "<STR_LIT>" ) <EOL> logging . info ( "<STR_LIT>" , Command ) <EOL> logging . info ( "<STR_LIT>" ) <EOL> logging . info ( "<STR_LIT>" + self . timer . get_start_date_time_as_str ( '<STR_LIT>' ) ) <EOL> logging . info ( "<STR_LIT>" ) <EOL> Output = '<STR_LIT>' <EOL> Cancelled = False <EOL> try : <EOL> proc = self . create_subprocess ( Command ) <EOL> while True : <EOL> line = proc . stdout . readline ( ) <EOL> if not line : break <EOL> logging . warn ( line . strip ( ) ) <EOL> Output += line <EOL> except KeyboardInterrupt : <EOL> os . killpg ( proc . pid , signal . SIGINT ) <EOL> outdata , errdata = proc . communicate ( ) <EOL> logging . warn ( outdata ) <EOL> Output += outdata <EOL> try : <EOL> os . killpg ( os . getpgid ( proc . pid ) , signal . SIGTERM ) <EOL> except OSError : <EOL> pass <EOL> Cancelled = True <EOL> Output += self . error_handler . UserAbort ( '<STR_LIT>' , Output ) <EOL> finally : <EOL> self . FinishCommand ( CommandInfo , Cancelled , PluginInfo ) <EOL> return scrub_output ( Output ) <EOL> def shell_exec ( self , Command , ** kwds ) : <EOL> kwds . setdefault ( "<STR_LIT>" , subprocess . PIPE ) <EOL> kwds . setdefault ( "<STR_LIT>" , subprocess . STDOUT ) <EOL> p = subprocess . Popen ( Command , shell = True , ** kwds ) <EOL> return p . communicate ( ) [ <NUM_LIT:0> ] </s>
