# ------------------------------------------------------------------------------
import os.path, time
import appy
from appy.px import Px
from appy.gen.mail import sendMail
from appy.gen.wrappers import AbstractWrapper
from appy.shared.utils import executeCommand
from appy.shared.ldap_connector import LdapConnector
# ------------------------------------------------------------------------------
class ToolWrapper(AbstractWrapper):
# --------------------------------------------------------------------------
# Navigation-related PXs
# --------------------------------------------------------------------------
# Icon for hiding/showing details below the title of an object shown in a
# list of objects.
pxShowDetails = Px('''
''')
# Displays up/down arrows in a table header column for sorting a given
# column. Requires variables "sortable", 'filterable' and 'field'.
pxSortAndFilter = Px('''
''')
# Buttons for navigating among a list of objects (from a Ref field or a
# query): next,back,first,last...
pxNavigate = Px('''
''')
# --------------------------------------------------------------------------
# PXs for graphical elements shown on every page
# --------------------------------------------------------------------------
# Global elements included in every page.
pxPagePrologue = Px('''
''')
pxPageBottom = Px('''
''')
pxLiveSearchResults = Px('''
''')
# Hook for defining a PX that proposes additional links, after the links
# corresponding to top-level pages.
pxLinks = Px('')
# Hook for defining a PX that proposes additional icons after standard
# icons in the user strip.
pxIcons = Px('')
# Displays the content of a layouted object (a page or a field). If the
# layouted object is a page, the "layout target" (where to look for PXs)
# will be the object whose page is shown; if the layouted object is a field,
# the layout target will be this field.
pxLayoutedObject = Px('''
:getattr(layoutTarget, px)
''')
pxHome = Px('''
::_('front_page_text')
''', template=AbstractWrapper.pxTemplate, hook='content')
# Show on query list or grid, the field content for a given object.
pxQueryField = Px('''
::sup::zobj.getListTitle(mode=titleMode, nav=navInfo, target=target, \
page=pageName, inPopup=inPopup, selectJs=selectJs, highlight=True)::zobj.highlight(sub)
:targetObj.appy().pxTransitions:field.pxView
:_('unauthorized'):field.pxRender''')
# Show query results as a list.
pxQueryResultList = Px('''
''', template=AbstractWrapper.pxTemplate, hook='content')
pxBack = Px('''
''')
def isManager(self):
'''Some pages on the tool can only be accessed by managers.'''
if self.user.has_role('Manager'): return 'view'
def isManagerEdit(self):
'''Some pages on the tool can only be accessed by managers, also in
edit mode.'''
if self.user.has_role('Manager'): return True
def computeConnectedUsers(self):
'''Computes a table showing users that are currently connected.'''
res = '
' \
'
%s
' % \
self.translate('last_user_access')
rows = []
for userId, lastAccess in self.o.loggedUsers.items():
user = self.search1('User', noSecurity=True, login=userId)
if not user: continue # Could have been deleted in the meanwhile
fmt = '%s (%s)' % (self.dateFormat, self.hourFormat)
access = time.strftime(fmt, time.localtime(lastAccess))
rows.append('
'
def getObject(self, uid):
'''Allow to retrieve an object from its unique identifier p_uid.'''
return self.o.getObject(uid, appy=True)
def getDiskFolder(self):
'''Returns the disk folder where the Appy application is stored.'''
return self.o.config.diskFolder
def getClass(self, zopeName):
'''Gets the Appy class corresponding to technical p_zopeName.'''
return self.o.getAppyClass(zopeName)
def getAvailableLanguages(self):
'''Returns the list of available languages for this application.'''
return [(t.id, t.title) for t in self.translations]
def convert(self, fileName, format):
'''Launches a UNO-enabled Python interpreter as defined in the self for
converting, using OpenOffice in server mode, a file named p_fileName
into an output p_format.'''
convScript = '%s/pod/converter.py' % os.path.dirname(appy.__file__)
cmd = '%s %s "%s" %s -p%d' % (self.unoEnabledPython, convScript,
fileName, format, self.openOfficePort)
self.log('executing %s...' % cmd)
return executeCommand(cmd) # The result can contain an error message
def sendMail(self, to, subject, body, attachments=None):
'''Sends a mail. See doc for appy.gen.mail.sendMail.'''
mailConfig = self.o.getProductConfig(True).mail
sendMail(mailConfig, to, subject, body, attachments=attachments,
log=self.log)
def formatDate(self, date, format=None, withHour=True, language=None):
'''Check doc @ToolMixin::formatDate.'''
if not date: return
return self.o.formatDate(date, format, withHour, language)
def getUserName(self, login=None, normalized=False):
return self.o.getUserName(login=login, normalized=normalized)
def refreshCatalog(self, startObject=None):
'''Reindex all Appy objects. For some unknown reason, method
catalog.refreshCatalog is not able to recatalog Appy objects.'''
if not startObject:
# This is a global refresh. Clear the catalog completely, and then
# reindex all Appy-managed objects, ie those in folders "config"
# and "data".
# First, clear the catalog.
self.log('recomputing the whole catalog...')
app = self.o.getParentNode()
app.catalog._catalog.clear()
nb = 1
failed = []
for obj in app.config.objectValues():
subNb, subFailed = self.refreshCatalog(startObject=obj)
nb += subNb
failed += subFailed
try:
app.config.reindex()
except:
failed.append(app.config)
# Then, refresh objects in the "data" folder.
for obj in app.data.objectValues():
subNb, subFailed = self.refreshCatalog(startObject=obj)
nb += subNb
failed += subFailed
# Re-try to index all objects for which reindexation has failed.
for obj in failed: obj.reindex()
if failed:
failMsg = ' (%d retried)' % len(failed)
else:
failMsg = ''
self.log('%d object(s) reindexed%s.' % (nb, failMsg))
else:
nb = 1
failed = []
for obj in startObject.objectValues():
subNb, subFailed = self.refreshCatalog(startObject=obj)
nb += subNb
failed += subFailed
try:
startObject.reindex()
except Exception, e:
failed.append(startObject)
return nb, failed
def _login(self, login):
'''Performs a login programmatically. Used by the test system.'''
self.request.user = self.search1('User', noSecurity=True, login=login)
def doSynchronizeExternalUsers(self):
'''Synchronizes the local User copies with a distant LDAP user base.'''
cfg = self.o.getProductConfig(True).ldap
if not cfg: raise Exception('LDAP config not found.')
counts = cfg.synchronizeUsers(self)
msg = 'LDAP users: %d created, %d updated, %d untouched.' % counts
return True, msg
def showSynchronizeUsers(self):
'''Show this button only if a LDAP connection exists and is enabled.'''
cfg = self.o.getProductConfig(True).ldap
if cfg and cfg.enabled: return 'view'
def mayDelete(self):
'''No one can delete the tool.'''
return
# ------------------------------------------------------------------------------
)
)
)