<s> """<STR_LIT>""" <EOL> import logging <EOL> import os <EOL> import signal <EOL> import subprocess <EOL> import sys <EOL> import tempfile <EOL> import luigi <EOL> logger = logging . getLogger ( '<STR_LIT>' ) <EOL> class ExternalProgramTask ( luigi . Task ) : <EOL> """<STR_LIT>""" <EOL> def program_args ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def program_environment ( self ) : <EOL> """<STR_LIT>""" <EOL> env = os . environ . copy ( ) <EOL> return env <EOL> @ property <EOL> def always_log_stderr ( self ) : <EOL> """<STR_LIT>""" <EOL> return True <EOL> def _clean_output_file ( self , file_object ) : <EOL> file_object . seek ( <NUM_LIT:0> ) <EOL> return '<STR_LIT>' . join ( map ( lambda s : s . decode ( '<STR_LIT:utf-8>' ) , file_object . readlines ( ) ) ) <EOL> def run ( self ) : <EOL> args = list ( map ( str , self . program_args ( ) ) ) <EOL> logger . info ( '<STR_LIT>' , '<STR_LIT:U+0020>' . join ( args ) ) <EOL> tmp_stdout , tmp_stderr = tempfile . TemporaryFile ( ) , tempfile . TemporaryFile ( ) <EOL> env = self . program_environment ( ) <EOL> proc = subprocess . Popen ( <EOL> args , <EOL> env = env , <EOL> stdout = tmp_stdout , <EOL> stderr = tmp_stderr <EOL> ) <EOL> try : <EOL> with ExternalProgramRunContext ( proc ) : <EOL> proc . wait ( ) <EOL> success = proc . returncode == <NUM_LIT:0> <EOL> stdout = self . _clean_output_file ( tmp_stdout ) <EOL> stderr = self . _clean_output_file ( tmp_stderr ) <EOL> if stdout : <EOL> logger . info ( '<STR_LIT>' . format ( stdout ) ) <EOL> if stderr : <EOL> if self . always_log_stderr or not success : <EOL> logger . info ( '<STR_LIT>' . format ( stderr ) ) <EOL> if not success : <EOL> raise ExternalProgramRunError ( <EOL> '<STR_LIT>' . format ( proc . returncode ) , <EOL> args , env = env , stdout = stdout , stderr = stderr ) <EOL> finally : <EOL> tmp_stderr . close ( ) <EOL> tmp_stdout . close ( ) <EOL> class ExternalProgramRunContext ( object ) : <EOL> def __init__ ( self , proc ) : <EOL> self . proc = proc <EOL> def __enter__ ( self ) : <EOL> self . __old_signal = signal . getsignal ( signal . SIGTERM ) <EOL> signal . signal ( signal . SIGTERM , self . kill_job ) <EOL> return self <EOL> def __exit__ ( self , exc_type , exc_val , exc_tb ) : <EOL> if exc_type is KeyboardInterrupt : <EOL> self . kill_job ( ) <EOL> signal . signal ( signal . SIGTERM , self . __old_signal ) <EOL> def kill_job ( self , captured_signal = None , stack_frame = None ) : <EOL> self . proc . kill ( ) <EOL> if captured_signal is not None : <EOL> sys . exit ( <NUM_LIT> + captured_signal ) <EOL> class ExternalProgramRunError ( RuntimeError ) : <EOL> def __init__ ( self , message , args , env = None , stdout = None , stderr = None ) : <EOL> super ( ExternalProgramRunError , self ) . __init__ ( message , args , env , stdout , stderr ) <EOL> self . message = message <EOL> self . args = args <EOL> self . env = env <EOL> self . out = stdout <EOL> self . err = stderr <EOL> def __str__ ( self ) : <EOL> info = self . message <EOL> info += '<STR_LIT>' . format ( '<STR_LIT:U+0020>' . join ( self . args ) ) <EOL> info += '<STR_LIT>' . format ( self . out or '<STR_LIT>' ) <EOL> info += '<STR_LIT>' . format ( self . err or '<STR_LIT>' ) <EOL> env_string = None <EOL> if self . env : <EOL> env_string = '<STR_LIT:U+0020>' . join ( [ '<STR_LIT:=>' . join ( [ k , '<STR_LIT>' . format ( v ) ] ) for k , v in self . env . items ( ) ] ) <EOL> info += '<STR_LIT>' . format ( env_string or '<STR_LIT>' ) <EOL> return info <EOL> class ExternalPythonProgramTask ( ExternalProgramTask ) : <EOL> """<STR_LIT>""" <EOL> virtualenv = luigi . Parameter ( <EOL> default = None , <EOL> positional = False , <EOL> description = '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> extra_pythonpath = luigi . Parameter ( <EOL> default = None , <EOL> positional = False , <EOL> description = '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def program_environment ( self ) : <EOL> env = super ( ExternalPythonProgramTask , self ) . program_environment ( ) <EOL> if self . extra_pythonpath : <EOL> pythonpath = '<STR_LIT::>' . join ( [ self . extra_pythonpath , env . get ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> env . update ( { '<STR_LIT>' : pythonpath } ) <EOL> if self . virtualenv : <EOL> path = '<STR_LIT::>' . join ( [ '<STR_LIT>' . format ( self . virtualenv ) , env . get ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> env . update ( { <EOL> '<STR_LIT>' : path , <EOL> '<STR_LIT>' : self . virtualenv <EOL> } ) <EOL> env . pop ( '<STR_LIT>' , None ) <EOL> return env </s>
