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
python类mswindows()的实例源码
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)")
