def GetDomainName():
try:
tok = win32security.OpenThreadToken(win32api.GetCurrentThread(),
TOKEN_QUERY, 1)
except win32api.error, details:
if details[0] != winerror.ERROR_NO_TOKEN:
raise
# attempt to open the process token, since no thread token
# exists
tok = win32security.OpenProcessToken(win32api.GetCurrentProcess(),
TOKEN_QUERY)
sid, attr = win32security.GetTokenInformation(tok, TokenUser)
win32api.CloseHandle(tok)
name, dom, typ = win32security.LookupAccountSid(None, sid)
return dom
python类error()的实例源码
def __init__ (self):
wx.Frame.__init__ (self, None, title="Clipboard viewer", size=(250,150))
self.first = True
self.nextWnd = None
# Get native window handle of this wxWidget Frame.
self.hwnd = self.GetHandle ()
# Set the WndProc to our function.
self.oldWndProc = win32gui.SetWindowLong (self.hwnd,
win32con.GWL_WNDPROC,
self.MyWndProc)
try:
self.nextWnd = win32clipboard.SetClipboardViewer (self.hwnd)
except win32api.error:
if win32api.GetLastError () == 0:
# information that there is no other window in chain
pass
else:
raise
def MyWndProc (self, hWnd, msg, wParam, lParam):
if msg == win32con.WM_CHANGECBCHAIN:
self.OnChangeCBChain (msg, wParam, lParam)
elif msg == win32con.WM_DRAWCLIPBOARD:
self.OnDrawClipboard (msg, wParam, lParam)
# Restore the old WndProc. Notice the use of win32api
# instead of win32gui here. This is to avoid an error due to
# not passing a callable object.
if msg == win32con.WM_DESTROY:
if self.nextWnd:
win32clipboard.ChangeClipboardChain (self.hwnd, self.nextWnd)
else:
win32clipboard.ChangeClipboardChain (self.hwnd, 0)
win32api.SetWindowLong (self.hwnd,
win32con.GWL_WNDPROC,
self.oldWndProc)
# Pass all messages (in this case, yours may be different) on
# to the original WndProc
return win32gui.CallWindowProc (self.oldWndProc,
hWnd, msg, wParam, lParam)
def checkWork(self):
finished = 0
fullDataRead = []
while 1:
try:
buffer, bytesToRead, result = win32pipe.PeekNamedPipe(self.pipe, 1)
# finished = (result == -1)
if not bytesToRead:
break
hr, data = win32file.ReadFile(self.pipe, bytesToRead, None)
fullDataRead.append(data)
except win32api.error:
finished = 1
break
dataBuf = ''.join(fullDataRead)
if dataBuf:
self.receivedCallback(dataBuf)
if finished:
self.cleanup()
return len(dataBuf)
def RegisterCoreDLL(coredllName = None):
"""Registers the core DLL in the registry.
If no params are passed, the name of the Python DLL used in
the current process is used and registered.
"""
if coredllName is None:
coredllName = win32api.GetModuleFileName(sys.dllhandle)
# must exist!
else:
try:
os.stat(coredllName)
except os.error:
print "Warning: Registering non-existant core DLL %s" % coredllName
hKey = win32api.RegCreateKey(GetRootKey() , BuildDefaultPythonKey())
try:
win32api.RegSetValue(hKey, "Dll", win32con.REG_SZ, coredllName)
finally:
win32api.RegCloseKey(hKey)
# Lastly, setup the current version to point to me.
win32api.RegSetValue(GetRootKey(), "Software\\Python\\PythonCore\\CurrentVersion", win32con.REG_SZ, sys.winver)
def __init__(self, writePipe, lostCallback):
self.disconnecting = False
self.producer = None
self.producerPaused = 0
self.streamingProducer = 0
self.outQueue = []
self.writePipe = writePipe
self.lostCallback = lostCallback
try:
win32pipe.SetNamedPipeHandleState(writePipe,
win32pipe.PIPE_NOWAIT,
None,
None)
except pywintypes.error:
# Maybe it's an invalid handle. Who knows.
pass
def RegisterPythonExe(exeFullPath, exeAlias = None, exeAppPath = None):
"""Register a .exe file that uses Python.
Registers the .exe with the OS. This allows the specified .exe to
be run from the command-line or start button without using the full path,
and also to setup application specific path (ie, os.environ['PATH']).
Currently the exeAppPath is not supported, so this function is general
purpose, and not specific to Python at all. Later, exeAppPath may provide
a reasonable default that is used.
exeFullPath -- The full path to the .exe
exeAlias = None -- An alias for the exe - if none, the base portion
of the filename is used.
exeAppPath -- Not supported.
"""
# Note - Dont work on win32s (but we dont care anymore!)
if exeAppPath:
raise error("Do not support exeAppPath argument currently")
if exeAlias is None:
exeAlias = os.path.basename(exeFullPath)
win32api.RegSetValue(GetRootKey(), GetAppPathsKey() + "\\" + exeAlias, win32con.REG_SZ, exeFullPath)
def RegisterModule(modName, modPath):
"""Register an explicit module in the registry. This forces the Python import
mechanism to locate this module directly, without a sys.path search. Thus
a registered module need not appear in sys.path at all.
modName -- The name of the module, as used by import.
modPath -- The full path and file name of the module.
"""
try:
import os
os.stat(modPath)
except os.error:
print "Warning: Registering non-existant module %s" % modPath
win32api.RegSetValue(GetRootKey(),
BuildDefaultPythonKey() + "\\Modules\\%s" % modName,
win32con.REG_SZ, modPath)
def addcounter(self,object, counter, instance = None, inum=-1, machine=None):
'''
Adds a single counter path to the paths attribute. Normally
this will be called by a child class' speciality functions,
rather than being called directly by the user. (Though it isn't
hard to call manually, since almost everything is given a default)
This method is only functional when the query is closed (or hasn't
yet been opened). This is to prevent conflict in multi-threaded
query applications).
e.g.:
query.addcounter('Memory','Available Bytes')
'''
if not self.active:
try:
self.rawaddcounter(object, counter, instance, inum, machine)
return 0
except win32api.error:
return -1
else:
return -1
def collectdataslave(self,format = win32pdh.PDH_FMT_LONG):
'''
### Not a public method
Called only when the Query is known to be open, runs over
the whole set of counters, appending results to the temp,
returns the values as a list.
'''
try:
win32pdh.CollectQueryData(self._base)
temp = []
for counter in self.counters:
ok = 0
try:
if counter:
temp.append(win32pdh.GetFormattedCounterValue(counter, format)[1])
ok = 1
except win32api.error:
pass
if not ok:
temp.append(-1) # a better way to signal failure???
return temp
except win32api.error: # will happen if, for instance, no counters are part of the query and we attempt to collect data for it.
return [-1] * len(self.counters)
# pickle functions
def open(self,*args,**namedargs):
'''
Explicitly open a query:
When you are needing to make multiple calls to the same query,
it is most efficient to open the query, run all of the calls,
then close the query, instead of having the collectdata method
automatically open and close the query each time it runs.
There are currently no arguments to open.
'''
# do all the normal opening stuff, self._base is now the query object
BaseQuery.open(*(self,)+args, **namedargs)
# should rewrite getinstpaths to take a single tuple
paths = []
for tup in self.volatilecounters:
paths[len(paths):] = self.getinstpaths(*tup)
for path in paths:
try:
self.counters.append(win32pdh.AddCounter(self._base, path))
self.curpaths.append(path) # if we fail on the line above, this path won't be in the table or the counters
except win32api.error:
pass # again, what to do with a malformed path???
def _GetServiceShortName(longName):
# looks up a services name
# from the display name
# Thanks to Andy McKay for this code.
access = win32con.KEY_READ | win32con.KEY_ENUMERATE_SUB_KEYS | win32con.KEY_QUERY_VALUE
hkey = win32api.RegOpenKey(win32con.HKEY_LOCAL_MACHINE, "SYSTEM\\CurrentControlSet\\Services", 0, access)
num = win32api.RegQueryInfoKey(hkey)[0]
longName = longName.lower()
# loop through number of subkeys
for x in range(0, num):
# find service name, open subkey
svc = win32api.RegEnumKey(hkey, x)
skey = win32api.RegOpenKey(hkey, svc, 0, access)
try:
# find display name
thisName = str(win32api.RegQueryValueEx(skey, "DisplayName")[0])
if thisName.lower() == longName:
return svc
except win32api.error:
# in case there is no key called DisplayName
pass
return None
# Open a service given either it's long or short name.
def RemoveService(serviceName):
try:
import perfmon
perfmon.UnloadPerfCounterTextStrings("python.exe "+serviceName)
except (ImportError, win32api.error):
pass
hscm = win32service.OpenSCManager(None,None,win32service.SC_MANAGER_ALL_ACCESS)
try:
hs = SmartOpenService(hscm, serviceName, win32service.SERVICE_ALL_ACCESS)
win32service.DeleteService(hs)
win32service.CloseServiceHandle(hs)
finally:
win32service.CloseServiceHandle(hscm)
import win32evtlogutil
try:
win32evtlogutil.RemoveSourceFromRegistry(serviceName)
except win32api.error:
pass
def __FindSvcDeps(findName):
if type(findName) is pywintypes.UnicodeType: findName = str(findName)
dict = {}
k = win32api.RegOpenKey(win32con.HKEY_LOCAL_MACHINE, "SYSTEM\\CurrentControlSet\\Services")
num = 0
while 1:
try:
svc = win32api.RegEnumKey(k, num)
except win32api.error:
break
num = num + 1
sk = win32api.RegOpenKey(k, svc)
try:
deps, typ = win32api.RegQueryValueEx(sk, "DependOnService")
except win32api.error:
deps = ()
for dep in deps:
dep = dep.lower()
dep_on = dict.get(dep, [])
dep_on.append(svc)
dict[dep]=dep_on
return __ResolveDeps(findName, dict)
def RestartService(serviceName, args = None, waitSeconds = 30, machine = None):
"Stop the service, and then start it again (with some tolerance for allowing it to stop.)"
try:
StopService(serviceName, machine)
except pywintypes.error, exc:
# Allow only "service not running" error
if exc.winerror!=winerror.ERROR_SERVICE_NOT_ACTIVE:
raise
# Give it a few goes, as the service may take time to stop
for i in range(waitSeconds):
try:
StartService(serviceName, args, machine)
break
except pywintypes.error, exc:
if exc.winerror!=winerror.ERROR_SERVICE_ALREADY_RUNNING:
raise
win32api.Sleep(1000)
else:
print "Gave up waiting for the old service to stop!"
def GetServiceClassString(cls, argv = None):
if argv is None:
argv = sys.argv
import pickle
modName = pickle.whichmodule(cls, cls.__name__)
if modName == '__main__':
try:
fname = win32api.GetFullPathName(argv[0])
path = os.path.split(fname)[0]
# Eaaaahhhh - sometimes this will be a short filename, which causes
# problems with 1.5.1 and the silly filename case rule.
# Get the long name
fname = os.path.join(path, win32api.FindFiles(fname)[0][8])
except win32api.error:
raise error("Could not resolve the path name '%s' to a full path" % (argv[0]))
modName = os.path.splitext(fname)[0]
return modName + "." + cls.__name__
def _find_localserver_module():
import win32com.server
path = win32com.server.__path__[0]
baseName = "localserver"
pyfile = os.path.join(path, baseName + ".py")
try:
os.stat(pyfile)
except os.error:
# See if we have a compiled extension
if __debug__:
ext = ".pyc"
else:
ext = ".pyo"
pyfile = os.path.join(path, baseName + ext)
try:
os.stat(pyfile)
except os.error:
raise RuntimeError("Can not locate the Python module 'win32com.server.%s'" % baseName)
return pyfile
def EnumKeys(root):
index = 0
ret = []
while 1:
try:
item = win32api.RegEnumKey(root, index)
except win32api.error:
break
try:
# Note this doesn't handle REG_EXPAND_SZ, but the implementation
# here doesn't need to - that is handled as the data is read.
val = win32api.RegQueryValue(root, item)
except win32api.error:
val = "" # code using this assumes a string.
ret.append((item, val))
index = index + 1
return ret
def GetDefaultProfileName():
import win32api, win32con
try:
key = win32api.RegOpenKey(win32con.HKEY_CURRENT_USER, "Software\\Microsoft\\Windows NT\\CurrentVersion\\Windows Messaging Subsystem\\Profiles")
try:
return win32api.RegQueryValueEx(key, "DefaultProfile")[0]
finally:
key.Close()
except win32api.error:
return None
#
# Recursive dump of folders.
#
def test():
import win32com.client
oldcwd = os.getcwd()
try:
session = gencache.EnsureDispatch("MAPI.Session")
try:
session.Logon(GetDefaultProfileName())
except pythoncom.com_error, details:
print "Could not log on to MAPI:", details
return
except pythoncom.error:
# no mapi.session - let's try outlook
app = gencache.EnsureDispatch("Outlook.Application")
session = app.Session
try:
TestUser(session)
TestAddress(session)
DumpFolders(session)
finally:
session.Logoff()
# It appears Exchange will change the cwd on us :(
os.chdir(oldcwd)
def _find_localserver_module():
import win32com.server
path = win32com.server.__path__[0]
baseName = "localserver"
pyfile = os.path.join(path, baseName + ".py")
try:
os.stat(pyfile)
except os.error:
# See if we have a compiled extension
if __debug__:
ext = ".pyc"
else:
ext = ".pyo"
pyfile = os.path.join(path, baseName + ext)
try:
os.stat(pyfile)
except os.error:
raise RuntimeError("Can not locate the Python module 'win32com.server.%s'" % baseName)
return pyfile
def EnumKeys(root):
index = 0
ret = []
while 1:
try:
item = win32api.RegEnumKey(root, index)
except win32api.error:
break
try:
# Note this doesn't handle REG_EXPAND_SZ, but the implementation
# here doesn't need to - that is handled as the data is read.
val = win32api.RegQueryValue(root, item)
except win32api.error:
val = "" # code using this assumes a string.
ret.append((item, val))
index = index + 1
return ret
def test():
import win32com.client
oldcwd = os.getcwd()
try:
session = gencache.EnsureDispatch("MAPI.Session")
try:
session.Logon(GetDefaultProfileName())
except pythoncom.com_error, details:
print "Could not log on to MAPI:", details
return
except pythoncom.error:
# no mapi.session - let's try outlook
app = gencache.EnsureDispatch("Outlook.Application")
session = app.Session
try:
TestUser(session)
TestAddress(session)
DumpFolders(session)
finally:
session.Logoff()
# It appears Exchange will change the cwd on us :(
os.chdir(oldcwd)
def checkWork(self):
finished = 0
fullDataRead = []
while 1:
try:
buffer, bytesToRead, result = win32pipe.PeekNamedPipe(self.pipe, 1)
# finished = (result == -1)
if not bytesToRead:
break
hr, data = win32file.ReadFile(self.pipe, bytesToRead, None)
fullDataRead.append(data)
except win32api.error:
finished = 1
break
dataBuf = ''.join(fullDataRead)
if dataBuf:
self.receivedCallback(dataBuf)
if finished:
self.cleanup()
return len(dataBuf)
def __init__(self, writePipe, lostCallback):
self.disconnecting = False
self.producer = None
self.producerPaused = 0
self.streamingProducer = 0
self.outQueue = []
self.writePipe = writePipe
self.lostCallback = lostCallback
try:
win32pipe.SetNamedPipeHandleState(writePipe,
win32pipe.PIPE_NOWAIT,
None,
None)
except pywintypes.error:
# Maybe it's an invalid handle. Who knows.
pass
def JobError(self, job, error):
print 'Job Error', job, error
f = error.GetFile()
print 'While downloading', f.GetRemoteName()
print 'To', f.GetLocalName()
print 'The following error happened:'
self._print_error(error)
if f.GetRemoteName().endswith('missing-favicon.ico'):
print 'Changing to point to correct file'
f2 = f.QueryInterface(bits.IID_IBackgroundCopyFile2)
favicon = 'http://www.python.org/favicon.ico'
print 'Changing RemoteName from', f2.GetRemoteName(), 'to', favicon
f2.SetRemoteName(favicon)
job.Resume()
else:
job.Cancel()
def __init__(self, module):
self.module = module
if hasattr(module, '__file__'):
fname = self.module.__file__
# Check for .pyc or .pyo or even .pys!
if fname[-1] in ['O','o','C','c', 'S', 's']: fname = fname[:-1]
try:
fname = win32api.GetFullPathName(fname)
except win32api.error:
pass
else:
if module.__name__=='__main__' and len(sys.argv)>0:
fname = sys.argv[0]
else:
fname = "<Unknown!>"
SourceCodeContainer.__init__(self, None, fname)
def FindPackagePath(packageName, knownFileName, searchPaths):
"""Find a package.
Given a ni style package name, check the package is registered.
First place looked is the registry for an existing entry. Then
the searchPaths are searched.
"""
import regutil, os
pathLook = regutil.GetRegisteredNamedPath(packageName)
if pathLook and IsPackageDir(pathLook, packageName, knownFileName):
return pathLook, None # The currently registered one is good.
# Search down the search paths.
for pathLook in searchPaths:
if IsPackageDir(pathLook, packageName, knownFileName):
# Found it
ret = os.path.abspath(pathLook)
return ret, ret
raise error("The package %s can not be located" % packageName)
def FindHelpPath(helpFile, helpDesc, searchPaths):
# See if the current registry entry is OK
import os, win32api, win32con
try:
key = win32api.RegOpenKey(win32con.HKEY_LOCAL_MACHINE, "Software\\Microsoft\\Windows\\Help", 0, win32con.KEY_ALL_ACCESS)
try:
try:
path = win32api.RegQueryValueEx(key, helpDesc)[0]
if FileExists(os.path.join(path, helpFile)):
return os.path.abspath(path)
except win32api.error:
pass # no registry entry.
finally:
key.Close()
except win32api.error:
pass
for pathLook in searchPaths:
if FileExists(os.path.join(pathLook, helpFile)):
return os.path.abspath(pathLook)
pathLook = os.path.join(pathLook, "Help")
if FileExists(os.path.join( pathLook, helpFile)):
return os.path.abspath(pathLook)
raise error("The help file %s can not be located" % helpFile)
def FindAppPath(appName, knownFileName, searchPaths):
"""Find an application.
First place looked is the registry for an existing entry. Then
the searchPaths are searched.
"""
# Look in the first path.
import regutil, string, os
regPath = regutil.GetRegisteredNamedPath(appName)
if regPath:
pathLook = regPath.split(";")[0]
if regPath and FileExists(os.path.join(pathLook, knownFileName)):
return None # The currently registered one is good.
# Search down the search paths.
for pathLook in searchPaths:
if FileExists(os.path.join(pathLook, knownFileName)):
# Found it
return os.path.abspath(pathLook)
raise error("The file %s can not be located for application %s" % (knownFileName, appName))
