utils.py

   1 # -*- coding: utf-8 -*-
   2 #######################################################################################
   3 #   Plinn - http://plinn.org                                                          #
   4 #   Copyright (C) 2005-2007  Benoît PIN <benoit.pin@ensmp.fr>                         #
   5 #                                                                                     #
   6 #   This program is free software; you can redistribute it and/or                     #
   7 #   modify it under the terms of the GNU General Public License                       #
   8 #   as published by the Free Software Foundation; either version 2                    #
   9 #   of the License, or (at your option) any later version.                            #
  10 #                                                                                     #
  11 #   This program is distributed in the hope that it will be useful,                   #
  12 #   but WITHOUT ANY WARRANTY; without even the implied warranty of                    #
  13 #   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the                     #
  14 #   GNU General Public License for more details.                                      #
  15 #                                                                                     #
  16 #   You should have received a copy of the GNU General Public License                 #
  17 #   along with this program; if not, write to the Free Software                       #
  18 #   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.   #
  19 #######################################################################################
  20 """ Plinn public utilities
  21
  22
  23 """
  24
  25 import string
  26 import re
  27 from types import StringType
  28 from random import randrange
  29 from Acquisition import aq_base
  30 from quopri import encodestring
  31 from zope.globalrequest import getRequest
  32 from AccessControl.PermissionRole import rolesForPermissionOn
  33 from AccessControl import ModuleSecurityInfo
  34 from AccessControl import getSecurityManager
  35 from AccessControl.User import UnrestrictedUser
  36 from OFS.CopySupport import _cb_decode, _cb_encode, cookie_path
  37 from Products.CMFCore.utils import getToolByName, getUtilityByInterfaceName
  38 from Products.CMFCore.exceptions import BadRequest
  39 from Products.Utf8Splitter.Utf8Splitter import Utf8Utils
  40 from Globals import REPLACEABLE, NOT_REPLACEABLE, UNIQUE
  41 from zope.i18n import translate as i18ntranslate
  42 from zope.i18n.interfaces import IUserPreferredLanguages
  43 from zope.i18nmessageid import MessageFactory
  44 from zope.component.interfaces import ComponentLookupError
  45 from zope.dottedname.resolve import resolve as resolve_dotted_name
  46 from zope.component import queryAdapter
  47
  48 _marker = []
  49
  50 security = ModuleSecurityInfo( 'Products.Plinn.utils' )
  51
  52 security.declarePublic('thisObjectComeFromPortalSkin')
  53 def thisObjectComeFromPortalSkin(ob, portal=None):
  54     """ check if ob comes from portal_skins """
  55     if not portal :
  56         portal = getToolByName(ob, 'portal_url')
  57         portal = portal.getPortalObject()
  58
  59     if ob.aq_self == portal.aq_self :
  60         return False
  61
  62     obId = ob.id
  63     if callable(obId) :
  64         obId = obId()
  65
  66     sob = getattr(portal, obId, None)
  67
  68     if sob is None :
  69         return False
  70     elif not(sob.aq_inner.aq_self is ob.aq_inner.aq_self) :
  71         return False
  72     else :
  73         try :
  74             portal._checkId(obId)
  75             return True
  76         except BadRequest :
  77             return False
  78
  79 security.declarePublic('listActionProviders_')
  80 def listActionProviders_(context) :
  81     atool = getToolByName(context, 'portal_actions')
  82     return atool.listActionProviders()
  83
  84 def capitalizeCompoundGivenName(givenName) :
  85     givenName = givenName.strip()
  86     givenNames = ' '.join(givenName.split('-')).split()
  87     givenNameCapitalized = '-'.join(map(string.capitalize, givenNames))
  88     return givenNameCapitalized
  89
  90
  91 def formatFullName(memberName, memberGivenName, memberId, nameBefore=1) :
  92     memberName = memberName.decode('utf-8')
  93     memberGivenName = memberGivenName.decode('utf-8')
  94     memberFullName = u''
  95     if memberName and memberGivenName :
  96         if nameBefore :
  97             memberFullName = memberName.capitalize() + ' ' + capitalizeCompoundGivenName(memberGivenName)
  98         else :
  99             memberFullName = capitalizeCompoundGivenName(memberGivenName) + ' ' + memberName.capitalize()
 100
 101     elif memberName and not memberGivenName :
 102         memberFullName = memberName.capitalize()
 103
 104     elif not memberName and memberGivenName :
 105         memberFullName = capitalizeCompoundGivenName(memberGivenName)
 106
 107     else :
 108         memberFullName = memberId
 109
 110     return memberFullName.encode('utf-8')
 111
 112 # from OFS.ObjectManager #63
 113 bad_url_chars = re.compile(r'[^a-zA-Z0-9-_~,.$\(\)@]')
 114
 115 security.declarePublic('makeValidId')
 116 def makeValidId(self, id, allow_dup=0):
 117     id = Utf8Utils.desacc(id)
 118     id = bad_url_chars.sub('-', id)
 119     # If allow_dup is false, an error will be raised if an object
 120     # with the given id already exists. If allow_dup is true,
 121     # only check that the id string contains no illegal chars;
 122     # check_valid_id() will be called again later with allow_dup
 123     # set to false before the object is added.
 124
 125     makeRandomId = False
 126     if id in ('.', '..'):
 127         makeRandomId = True
 128     if id.startswith('_'):
 129         id = id.lstrip('_')
 130     if id.startswith('aq_'):
 131         id = id[3:]
 132
 133     while id.endswith('__') :
 134         id = id[:-1]
 135     if not allow_dup:
 136         obj = getattr(self, id, None)
 137         if obj is not None:
 138             # An object by the given id exists either in this
 139             # ObjectManager or in the acquisition path.
 140             flags = getattr(obj, '__replaceable__', NOT_REPLACEABLE)
 141             if hasattr(aq_base(self), id):
 142                 # The object is located in this ObjectManager.
 143                 if not flags & REPLACEABLE:
 144                     makeRandomId = True
 145                 # else the object is replaceable even if the UNIQUE
 146                 # flag is set.
 147             elif flags & UNIQUE:
 148                 makeRandomId = True
 149     if id == 'REQUEST':
 150         makeRandomId = True
 151
 152     if makeRandomId is True :
 153         id = str(randrange(2,10000)) + id
 154     return id
 155
 156
 157
 158 def _checkMemberPermission(userid, permission, obj, StringType = type('')):
 159     user = obj.aq_inner.acl_users.getUser(userid)
 160     roles = rolesForPermissionOn(permission, obj)
 161     if type(roles) is StringType:
 162         roles=[roles]
 163     if user.allowed( obj, roles ):
 164         return 1
 165     return 0
 166
 167 def getCPInfo(self) :
 168     if self.REQUEST.RESPONSE.cookies.has_key('__cp') :
 169         cp = self.REQUEST.RESPONSE.cookies['__cp']['value']
 170     else :
 171         cp = self.REQUEST.get('__cp')
 172     try: cp = _cb_decode(cp)
 173     except: return None
 174     return cp
 175
 176
 177 def popCP(self, indexes=None) :
 178     try: cp = _cb_decode(self.REQUEST['__cp'])
 179     except: return
 180
 181     paths = list(cp[1])
 182     if indexes is not None :
 183         indexes = list(indexes)
 184         indexes.sort()
 185         indexes.reverse()
 186         for index in indexes :
 187             paths.pop(index)
 188     else :
 189         paths.pop()
 190
 191     if not paths :
 192         self.REQUEST.RESPONSE.expireCookie('__cp', path=self.REQUEST['BASEPATH1'] or "/")
 193     else :
 194         paths = tuple(paths)
 195         cp = _cb_encode( (cp[0], paths) )
 196         resp = self.REQUEST['RESPONSE']
 197         resp.setCookie('__cp', cp, path='%s' % cookie_path(self.REQUEST))
 198
 199 security.declarePublic('Message')
 200 Message = MessageFactory('plinn')
 201
 202 security.declarePublic('translate')
 203 def translate(message, context=None):
 204     """ Translate i18n message.
 205     """
 206     if isinstance(message, Exception):
 207         try:
 208             message = message[0]
 209         except (TypeError, IndexError):
 210             pass
 211     if not context :
 212         request = getRequest()
 213     else :
 214         request = context.REQUEST
 215     return i18ntranslate(message, domain='plinn', context=request)
 216
 217 security.declarePublic('desacc')
 218 desacc = Utf8Utils.desacc
 219
 220 security.declarePublic('getPreferredLanguages')
 221 def getPreferredLanguages(context):
 222     """ returns browser prefered languages"""
 223     request = getattr(context, 'REQUEST', None)
 224     if request is not None :
 225         adapter = IUserPreferredLanguages(request, None)
 226         if adapter is not None :
 227             return adapter.getPreferredLanguages()
 228     return []
 229
 230 security.declarePublic('getBestTranslationLanguage')
 231 def getBestTranslationLanguage(langs, context):
 232     """ returns best translation language according
 233         to available languages (param langs)
 234         and user preferences (retrieves by context)
 235     """
 236     request = getattr(context, 'REQUEST', None)
 237     if request :
 238         negociator = getUtilityByInterfaceName('zope.i18n.interfaces.INegotiator')
 239         return negociator.getLanguage(langs, request) or langs[0]
 240     else :
 241         return langs[0]
 242
 243 security.declarePublic('getAdapterByInterface')
 244 def getAdapterByInterface(ob, dotted_name, default=_marker) :
 245     """ Get the adapter which provides the interface on the given object.
 246     """
 247     try:
 248         iface = resolve_dotted_name(dotted_name)
 249     except ImportError:
 250         if default is _marker:
 251             raise ComponentLookupError, dotted_name
 252         return default
 253
 254     adapter = queryAdapter(ob, iface, default=default)
 255     if adapter is _marker :
 256         raise ComponentLookupError, "no adapter providing %r found on %r" % (dotted_name, ob)
 257
 258     # the adapter must be wrapped to allow security mahinery to work.
 259     if adapter != default :
 260         return adapter.__of__(ob)
 261     else :
 262         return default
 263
 264 security.declarePublic('encodeQuopriEmail')
 265 def encodeQuopriEmail(name, email) :
 266     qpName = encodestring(name).replace('=\n', '')
 267     return '''"=?utf-8?q?%s?=" <%s>''' % (qpName, email)
 268
 269 security.declarePublic('encodeMailHeader')
 270 def encodeMailHeader(content) :
 271     s = encodestring(content).replace('=\n', '')
 272     s = s.replace('_', '=5F')
 273     s = s.replace(' ', '_')
 274
 275     lines = []
 276     STEP = 50
 277     start = 0
 278     stop = STEP
 279     part = s[start:stop]
 280     lines.append(part)
 281
 282     while len(part) == STEP:
 283         start = start + STEP
 284         stop = stop + STEP
 285         part = s[start:stop]
 286         lines.append(part)
 287
 288     lines = [' =?utf-8?Q?%s?=' % part for part in lines]
 289     s = '\n'.join(lines)
 290     s = s.strip()
 291     return s
 292
 293
 294 def _sudo(func, userid=None) :
 295     """
 296     execute func or any callable object
 297     without restriction. Used to switch off
 298     security assertions (eg. checkPermission) encountered
 299     during the execution.
 300     """
 301
 302     sm = getSecurityManager()
 303     restrictedUser = sm.getUser()
 304
 305     if not userid :
 306         userid = restrictedUser.getId()
 307
 308     sm._context.user = UnrestrictedUser(userid, '', (), ())
 309
 310     deferedEx = None
 311     try :
 312         ret = func()
 313     except Exception, e :
 314         deferedEx = e
 315
 316     sm._context.user = restrictedUser
 317
 318     if deferedEx is not None :
 319         raise e
 320
 321     return ret
 322
 323 security.declarePublic('searchContentsWithLocalRolesForAuthenticatedUser')
 324 def searchContentsWithLocalRolesForAuthenticatedUser(**kw):
 325     mtool = getUtilityByInterfaceName('Products.CMFCore.interfaces.IMembershipTool')
 326     ctool = getUtilityByInterfaceName('Products.CMFCore.interfaces.ICatalogTool')
 327     member = mtool.getAuthenticatedMember()
 328     userid = member.getId()
 329     userAndGroups = ['user:%s' % userid]
 330
 331     getGroups = getattr(member, 'getGroups', None)
 332     if getGroups is not None :
 333         for group in getGroups():
 334             userAndGroups.append('user:'+group)
 335
 336     kw[ 'allowedRolesAndUsers' ] = userAndGroups
 337
 338     return ctool.unrestrictedSearchResults(**kw)