## $Id: helpers.py 8466 2012-05-17 16:30:02Z uli $ ## ## Copyright (C) 2011 Uli Fouquet & Henrik Bettermann ## This program is free software; you can redistribute it and/or modify ## it under the terms of the GNU General Public License as published by ## the Free Software Foundation; either version 2 of the License, or ## (at your option) any later version. ## ## This program is distributed in the hope that it will be useful, ## but WITHOUT ANY WARRANTY; without even the implied warranty of ## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the ## GNU General Public License for more details. ## ## You should have received a copy of the GNU General Public License ## along with this program; if not, write to the Free Software ## Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA ## """General helper functions for Kofa. """ import datetime import imghdr import os import pytz import re import shutil import grok from cStringIO import StringIO from docutils.core import publish_string from zope.component import getUtility from zope.component.interfaces import IFactory from zope.interface import implementedBy from zope.interface.interface import Method, Attribute from zope.schema import getFieldNames from zope.schema.fieldproperty import FieldProperty from zope.security.interfaces import NoInteraction from zope.security.management import getInteraction from zope.pluggableauth.interfaces import IAuthenticatorPlugin BUFSIZE = 8 * 1024 def remove_file_or_directory(filepath): """Remove a file or directory. Different to :func:`shutil.rmtree` we also accept not existing paths (returning silently) and if a dir turns out to be a regular file, we remove that. """ filepath = os.path.abspath(filepath) if not os.path.exists(filepath): return if os.path.isdir(filepath): shutil.rmtree(filepath) else: os.unlink(filepath) return def copy_filesystem_tree(src, dst, overwrite=False, del_old=False): """Copy contents of directory src to directory dst. Both directories must exists. If `overwrite` is true, any same named objects will be overwritten. Otherwise these files will not be touched. If `del_old` is true, copied files and directories will be removed from the src directory. This functions returns a list of non-copied files. Unix hidden files and directories (starting with '.') are not processed by this function. """ if not os.path.exists(src): raise ValueError('source path does not exist: %s' % src) if not os.path.exists(dst): raise ValueError('destination path does not exist: %s' % dst) if not os.path.isdir(src): raise ValueError('source path is not a directory: %s' % src) if not os.path.isdir(dst): raise ValueError('destination path is not a directory: %s' % dst) not_copied = [] for item in os.listdir(src): if item.startswith('.'): continue # We do not copy hidden stuff... itemsrc = os.path.join(src, item) itemdst = os.path.join(dst, item) if os.path.exists(itemdst): if overwrite is True: remove_file_or_directory(itemdst) else: not_copied.append(item) continue if os.path.isdir(itemsrc): shutil.copytree(itemsrc, itemdst) else: shutil.copy2(itemsrc, itemdst) if del_old: remove_file_or_directory(itemsrc) return not_copied def get_inner_HTML_part(html_code): """Return the 'inner' part of a complete HTML snippet. If there is a form part, get this. If there is no form part, try to return the body part contents. If there is no body, return as-is. Let's see how that works. If we deliver some doc with form, we will get that form only: >>> doc = '
My Form
Outside the form' >>> get_inner_HTML_part(doc) '
My Form
' No form? Then seek for a body part and get the contents: >>> doc = 'My BodyTrailing Trash' >>> get_inner_HTML_part(doc) 'My Body' If none of these is included, return what we got: >>> doc = 'without body nor form' >>> get_inner_HTML_part(doc) 'without body nor form' """ try: result = re.match('^.+(]*>.*).+$', html_code, re.DOTALL).groups()[0] return result except AttributeError: # No
part included try: result = re.match('^.+]*>(.*).*$', html_code, re.DOTALL).groups()[0] return result except AttributeError: # No and no tag... pass return html_code class FactoryBase(grok.GlobalUtility): """A factory for things. This is a baseclass for easier creation of factories. Factories are utilities that are registered under a certain name and return instances of certain classes when called. In :mod:`waeup.kofa` we use factories extensively for batching. While processing a batch some processors looks up a factory to create real-world instances that then get filled with data from imported CSV files. To get rid of reimplementing the same stuff over and over again, most notably the methods defined here, we offer this base class (which will *not* be registered as a factory itself). Real factories can then be created like this: >>> import grok >>> from waeup.kofa.utils.helpers import FactoryBase >>> class MyObject(object): ... # Some class we want to get instances of. ... pass >>> class MyObjectFactory(FactoryBase): ... # This is the factory for MyObject instances ... grok.name(u'waeup.kofa.factory.MyObject') ... factory = MyObject That's it. It is essential to set the ``factory`` attribute, which will determine the class of which instances should be created when called. The given name must even be unique amongst all utilities registered during runtime. While you can pick any name you like you might want to prepend ``waeup.kofa.factory.`` to the name string to make sure it does not clash with names of other utilities one day. Before all this works we have to grok the baseclass once and our freshly defined factory. This executes all the component registration stuff we don't want to do ourselves. In daily use this is done automatically on startup of a :mod:`waeup.kofa` system. >>> grok.testing.grok('waeup.kofa.utils.helpers') >>> grok.testing.grok_component( ... 'MyObjectFactory', MyObjectFactory ... ) True After grokking we (and processors) can create objects without knowing about the location of the real class definition, just by the factory name: >>> from zope.component import createObject >>> obj = createObject('waeup.kofa.factory.MyObject') >>> isinstance(obj, MyObject) True We can also use the regular utility lookups to find our new factory: >>> from zope.component import getUtility >>> from zope.component.interfaces import IFactory >>> factory = getUtility( ... IFactory, name='waeup.kofa.factory.MyObject' ... ) >>> isinstance(factory, MyObjectFactory) True And this factory generates `MyObject` instances: >>> obj = factory() >>> isinstance(obj, MyObject) True """ grok.baseclass() # Do not grok this class, do not register us. grok.implements(IFactory) # You can override any of the following attributes in derived # classes. The `grok.name` setting *must* even be set to some # unique value. grok.name(u'waeup.Factory') title = u"Create instances of ``factory``.", description = u"This factory instantiates new applicant instances." factory = None def __call__(self, *args, **kw): """The main factory function. Returns an instance of the requested object. """ return self.factory() def getInterfaces(self): # Required by IFactory return implementedBy(self.factory) def ReST2HTML_w_warnings(source_string): """Convert a reStructuredText string to HTML preserving warnings. Returns a tuple ``(, )``, both being strings. Where ```` is the HTML code generated from the source string (in unicode), ```` is a string containing any warning messages or ``None``. Regular multi-line ReStructuredText strings will be returned as HTML code: >>> from waeup.kofa.utils.helpers import ReST2HTML >>> source = ''' ... Headline ... ======== ... ... - A list item ... - Another item ... ... Thanks for watching! ... ''' >>> html, warnings = ReST2HTML_w_warnings(source) >>> print html

Headline

  • A list item
  • Another item

Thanks for watching!

Here no warnings happened, so the `warnings` are ``None``: >>> warnings is None True If warnings happen then they can be retrieved in the returned ``warnings``. We try to render an erraneous document: >>> source = ''' ... Headline ... ====== ... ... Thanks for watching! ... ''' >>> html, warnings = ReST2HTML_w_warnings(source) >>> print html

Headline

Thanks for watching!

>>> print warnings :3: (WARNING/2) Title underline too short. Headline ====== As you can see, the warnings are not displayed inline the document but can be retrieved from the returned warnings, which is a string or ``None``. """ warnings = StringIO() fulldoc = publish_string( source_string, writer_name='html4css1', settings_overrides={ 'report_level': 0, 'warning_stream': warnings, }) warnings.seek(0) warning_msgs = warnings.read() if warning_msgs: # Render again, this time with no warnings inline... fulldoc = publish_string( source_string, writer_name='html4css1', settings_overrides={ 'report_level': 10000, 'halt_level': 10000, 'warning_stream': warnings, }) if warning_msgs == '': warning_msgs = None result = get_inner_HTML_part(fulldoc).strip() if not isinstance(result, unicode): result = result.decode('utf-8') return result, warning_msgs def ReST2HTML(source_string): """Render a string containing ReStructuredText to HTML. Any warnings about too short headings, etc. are silently discarded. Use :func:`ReST2HTML_w_warnings` if you want to get any warnings. The returned string will be unicode. A regular document will be rendered like this: >>> source = ''' ... Headline ... ======== ... ... Thanks for watching! ... ''' >>> html = ReST2HTML(source) >>> print html

Headline

Thanks for watching!

A document with markup problems (here: the underline is too short) will look similar: >>> source = ''' ... Headline ... ====== ... ... Thanks for watching! ... ''' >>> html = ReST2HTML(source) >>> print html

Headline

Thanks for watching!

""" html, warnings = ReST2HTML_w_warnings(source_string) return html def attrs_to_fields(cls): """Turn the attributes of a class into FieldProperty instances. With Python >= 2.6 we can even use this function as a class decorator. """ iface = list(implementedBy(cls))[0] for field_name in getFieldNames(iface): setattr(cls, field_name, FieldProperty(iface[field_name])) return cls def get_current_principal(): """Get the 'current' principal. This method works without a request. Examining a request is the regular (and recommended) way to get a principal involved 'currently'. Use this method only if you really have no access to the current request. Returns ``None`` when no principal is involved (for instance during tests). """ try: principal = getInteraction().participations[0].principal except NoInteraction: return None except IndexError: # No participations present return None return principal def cmp_files(file_descr1, file_descr2): """Compare two files by their file descriptors. Returns ``True`` if both are equal, ``False`` otherwise. """ file_descr1.seek(0) file_descr2.seek(0) while True: b1 = file_descr1.read(BUFSIZE) b2 = file_descr2.read(BUFSIZE) if b1 != b2: return False if not b1: return True def string_from_bytes(number): """Turn a number into some textual representation. Examples: >>> string_from_bytes(1) u'1 byte(s)' >>> string_from_bytes(1025) u'1 KB' >>> string_from_bytes(1.5 * 1024*1024) u'1.50 MB' >>> string_from_bytes(673.286 * 1024**3) u'673.29 GB' """ if number < 1024: return u'%s byte(s)' % (str(number),) elif number < 1024**2: return u'%s KB' % (number / 1024,) elif number < 1024**3: return u'%.2f MB' % (number / 1024**2,) return u'%.2f GB' % (number / 1024**3,) def file_size(file_like_obj): """Determine file size in most effective manner. Returns the number of bytes in a file. This function works for both, real files as well as file-like objects like cStringIO based 'files'. Example: >>> from cStringIO import StringIO >>> file_size(StringIO('my file content')) 15 Please note that this function expects the file-like object passed in to be at first reading position (it does no seek(0)) and that when finished the file pointer might be at end of file. """ if hasattr(file_like_obj, 'fileno'): return os.fstat(file_like_obj.fileno())[6] file_like_obj.seek(0, 2) # seek to last position in file return file_like_obj.tell() def get_user_account(request): """Return local user account. """ principal_id = request.principal.id authenticator = getUtility(IAuthenticatorPlugin, name='users') account = authenticator.getAccount(principal_id) return account def iface_names(iface, omit=[], exclude_attribs=True, exclude_methods=True): """Get all attribute names of an interface. Searches also base interfaces. Names of fields that are pure attributes (i.e. zope.interface.Attribute) or methods are excluded by default. Names of typical fields derived from zope.schema are included. The `omit` paramter can give a list of names to exclude. Returns an unsorted list of strings. """ ifaces = set((iface,)) # Collect all interfaces (also bases) recursively while True: ext_ifaces = set(ifaces) for iface in ext_ifaces: ext_ifaces = set.union(ext_ifaces, set(iface.getBases())) if ext_ifaces == ifaces: # No new interfaces found, list complete break ifaces = ext_ifaces # Collect (filtered) names of collected interfaces result = [] for iface in ifaces: for name, descr in iface.namesAndDescriptions(): if name in omit: continue if exclude_attribs and descr.__class__ is Attribute: continue if exclude_methods and isinstance(descr, Method): continue result.append(name) return result def get_sorted_preferred(tuples_iterable, preferred_list): """Get a list of tuples (,<TOKEN>) with values in `preferred_list` put in front. The rest of the tuples iterable is returned in orginal order. This is useful for putting default entries on top of (already sorted) lists of choice values, for instance when sorting countries and their code. Sample: We have a list of tuples with uppercase 'titles' and lowercase 'tokens'. This list is already sorted but we want certain values of this list to show up before other values. For instance we want to see the 'C' entry to come first. >>> get_sorted_preferred([('A','a'), ('B','b'), ('C','c')], ... ['c']) (('C', 'c'), ('A', 'a'), ('B', 'b')) i.e. the entry with 'c' as second value moved to head of result. We can also require multiple entries at head of list: >>> get_sorted_preferred([('A','a'), ('B','b'), ('C','c')], ... ['b', 'c']) (('B', 'b'), ('C', 'c'), ('A', 'a')) We required the 'b' entry to come before the 'c' entry and then the rest of the input list. That's what we got. The result is returned as a tuple of tuples to keep order of values. """ result = [None for x in preferred_list] for title, code in tuples_iterable: if code in preferred_list: index = preferred_list.index(code) result[index] = (title, code) else: result.append((title, code)) return tuple(result) def now(tz=None): """Get current datetime in timezone of `tz`. If `tz`, a `tzinfo` instance, is None, UTC time is returned. `tz` should be a timezone as defined in pytz. """ return to_timezone(datetime.datetime.utcnow(), tz=tz) def to_timezone(dt, tz=None): """Shift datetime into timezone `tz`. If datetime `dt` contains no `tzinfo` (i.e. it is 'naive'), it is assumed to be UTC. If no `tz` is given, shift to UTC is performed. If `dt` is not a datetime.datetime, the input value is returned unchanged. """ if not isinstance(dt, datetime.datetime): return dt if tz is None: tz = pytz.utc if dt.tzinfo is None: dt = pytz.utc.localize(dt) return tz.normalize(dt.tzinfo.normalize(dt).astimezone(tz)) def get_fileformat(path, bytestream=None): """Try to determine the file format of a given media file. Although checks done here are not done very thoroughly, they make no assumptions about the filetype by looking at its filename extension or similar. Instead they check header data to comply with common known rules (Magic Words). If bytestream is not `None` the `path` is ignored. Returns filetype as string (something like ``'jpg'``) if file-format can be recognized, ``None`` else. Tested recognized filetypes currently are `jpg`, `png`, and `pdf`. More filetypes (though untested in waeup.kofa) are automatically recognized because we deploy the stdlib `imghdr` library. See this module's docs for a complete list of filetypes recognized. """ if path is None and bytestream is None: return None img_type = None if bytestream is not None: img_type = imghdr.what(path, bytestream) else: img_type = imghdr.what(path) for name, replacement in (('jpeg', 'jpg'), ('tiff', 'tif')): if img_type == name: img_type = replacement return img_type def check_pdf(bytestream, file): """Tell whether a file or bytestream is a PDF file. Works as a test/plugin for the stdlib `imghdr` library. """ if file is not None: file.seek(0) bytestream = file.read(4) file.seek(0) if bytestream.startswith('%PDF'): return 'pdf' return None # register check_pdf as header check function with `imghdr` if check_pdf not in imghdr.tests: imghdr.tests.append(check_pdf)