Python subprocess 模块，mswindows() 实例源码

def stdout_encode(data):
    try:
        data = data or ""

        # Reference: http://bugs.python.org/issue1602
        if mswindows:
            output = data.encode(sys.stdout.encoding, "replace")

            if '?' in output and '?' not in data:
                warn = "cannot properly display Unicode characters "
                warn += "inside Windows OS command prompt "
                warn += "(http://bugs.python.org/issue1602). All "
                warn += "unhandled occurances will result in "
                warn += "replacement with '?' character. Please, find "
                warn += "proper character representation inside "
                warn += "corresponding output files. "
                single_time_warn_message(warn)

            ret = output
        else:
            ret = data.encode(sys.stdout.encoding)
    except Exception as e:
        ret = data.encode(UNICODE_ENCODING) if isinstance(data, unicode) else data

    return ret

def check_sudo():
    """
    Checks for sudo/Administrator privileges
    """

    check = None

    if not subprocess.mswindows:
        if getattr(os, "geteuid"):
            check = os.geteuid() == 0
    else:
        import ctypes
        check = ctypes.windll.shell32.IsUserAnAdmin()

    return check

def _chown(filepath):
    if not subprocess.mswindows and os.path.exists(filepath):
        try:
            os.chown(filepath, int(os.environ.get("SUDO_UID", -1)), int(os.environ.get("SUDO_GID", -1)))
        except Exception, ex:
            print "[x] '%s'" % ex

def beep():
    try:
        if subprocess.mswindows:
            _win_wav_play(BEEP_WAV_FILENAME)
        elif sys.platform == "darwin":
            _mac_beep()
        elif sys.platform == "linux2":
            _linux_wav_play(BEEP_WAV_FILENAME)
        else:
            _speaker_beep()
    except:
        _speaker_beep()

def startElastix(self, cmdLineArguments):
    self.addLog("Register volumes...")
    import subprocess
    executableFilePath = os.path.join(self.getElastixBinDir(),self.elastixFilename)
    logging.info("Register volumes using: "+executableFilePath+": "+repr(cmdLineArguments))
    if subprocess.mswindows:
      return subprocess.Popen([executableFilePath] + cmdLineArguments, env=self.getElastixEnv(),
                            stdout=subprocess.PIPE, universal_newlines=True, startupinfo=self.getStartupInfo())
    else:
      return subprocess.Popen([executableFilePath] + cmdLineArguments, env=self.getElastixEnv(),
                            stdout=subprocess.PIPE, universal_newlines=True)

def startTransformix(self, cmdLineArguments):
    self.addLog("Generate output...")
    import subprocess
    executableFilePath = os.path.join(self.getElastixBinDir(), self.transformixFilename)
    logging.info("Generate output using: " + executableFilePath + ": " + repr(cmdLineArguments))    
    if subprocess.mswindows:
      return subprocess.Popen([os.path.join(self.getElastixBinDir(),self.transformixFilename)] + cmdLineArguments, env=self.getElastixEnv(),
                            stdout=subprocess.PIPE, universal_newlines = True, startupinfo=self.getStartupInfo())
    else:
      return subprocess.Popen([os.path.join(self.getElastixBinDir(),self.transformixFilename)] + cmdLineArguments, env=self.getElastixEnv(),
                            stdout=subprocess.PIPE, universal_newlines = True)

def __init__(self, *args, **kwargs):
    if subprocess.mswindows:
      kwargs = inspect.getcallargs(
          subprocess.Popen.__init__, self, *args, **kwargs)
      args = []
      kwargs.pop('self')
      startup_info = kwargs.pop('startupinfo')
      if startup_info is None:
        startup_info = subprocess.STARTUPINFO()
      startup_info.dwFlags |= subprocess.STARTF_USESHOWWINDOW
      kwargs['startupinfo'] = startup_info
    return super(PopenBg, self).__init__(*args, **kwargs)

def beep():
    try:
        if subprocess.mswindows:
            _win_wav_play(BEEP_WAV_FILENAME)
        elif sys.platform == "darwin":
            _mac_beep()
        elif sys.platform == "linux2":
            _linux_wav_play(BEEP_WAV_FILENAME)
        else:
            _speaker_beep()
    except:
        _speaker_beep()

def beep():
    try:
        if subprocess.mswindows:
            _win_wav_play(BEEP_WAV_FILENAME)
        elif sys.platform == "darwin":
            _mac_beep()
        elif sys.platform == "linux2":
            _linux_wav_play(BEEP_WAV_FILENAME)
        else:
            _speaker_beep()
    except:
        _speaker_beep()

def beep():
    try:
        if subprocess.mswindows:
            _win_wav_play(BEEP_WAV_FILENAME)
        elif sys.platform == "darwin":
            _mac_beep()
        elif sys.platform == "linux2":
            _linux_wav_play(BEEP_WAV_FILENAME)
        else:
            _speaker_beep()
    except:
        _speaker_beep()

def main():

    print("%s (server) #v%s\n" % (NAME, VERSION))

    parser = optparse.OptionParser(version=VERSION)
    parser.add_option("-c", dest="config_file", default=CONFIG_FILE, help="Configuration file (default: '%s')" % os.path.split(CONFIG_FILE)[-1])
    options, _ = parser.parse_args()

    read_config(options.config_file)

    if config.USE_SSL:
        try:
            import OpenSSL
        except ImportError:
            if subprocess.mswindows:
                exit("[!] please install 'pyopenssl' (e.g. 'pip install pyopenssl')")
            else:
                msg, _ = "[!] please install 'pyopenssl'", platform.linux_distribution()[0].lower()
                for distro, install in {("fedora", "centos"): "sudo yum install pyOpenSSL", ("debian", "ubuntu"): "sudo apt-get install python-openssl"}.items():
                    if _ in distro:
                        msg += " (e.g. '%s')" % install
                        break
                exit(msg)

        if not config.SSL_PEM or not os.path.isfile(config.SSL_PEM):
            hint = "openssl req -new -x509 -keyout %s -out %s -days 365 -nodes -subj '/O=%s CA/C=EU'" % (config.SSL_PEM or "server.pem", config.SSL_PEM or "server.pem", NAME)
            exit("[!] invalid configuration value for 'SSL_PEM' ('%s')\n[?] (hint: \"%s\")" % (config.SSL_PEM, hint))

    def update_timer():
        if config.USE_SERVER_UPDATE_TRAILS:
            update_trails()

        update_ipcat()

        thread = threading.Timer(config.UPDATE_PERIOD, update_timer)
        thread.daemon = True
        thread.start()

    if config.UDP_ADDRESS and config.UDP_PORT:
        if check_sudo() is False:
            exit("[!] please run '%s' with sudo/Administrator privileges when using 'UDP_ADDRESS' configuration value" % __file__)

        start_logd(address=config.UDP_ADDRESS, port=config.UDP_PORT, join=False)

    try:
        update_timer()
        start_httpd(address=config.HTTP_ADDRESS, port=config.HTTP_PORT, pem=config.SSL_PEM if config.USE_SSL else None, join=True)
    except KeyboardInterrupt:
        print("\r[x] stopping (Ctrl-C pressed)")