← Back
Editing: drv_libxml2.py
# -*- coding: iso-8859-1 -*- """ A SAX2 driver for libxml2, on top of it's XmlReader API USAGE # put this file (drv_libxml2.py) in PYTHONPATH import xml.sax reader = xml.sax.make_parser(["drv_libxml2"]) # ...and the rest is standard python sax. CAVEATS - Lexical handlers are supported, except for start/endEntity (waiting for XmlReader.ResolveEntity) and start/endDTD - Error callbacks are not exactly synchronous, they tend to be invoked before the corresponding content callback, because the underlying reader interface parses data by chunks of 512 bytes TODO - search for TODO - some ErrorHandler events (warning) - some ContentHandler events (setDocumentLocator, skippedEntity) - EntityResolver (using libxml2.?) - DTDHandler (if/when libxml2 exposes such node types) - DeclHandler (if/when libxml2 exposes such node types) - property_xml_string? - feature_string_interning? - Incremental parser - additional performance tuning: - one might cache callbacks to avoid some name lookups - one might implement a smarter way to pass attributes to startElement (some kind of lazy evaluation?) - there might be room for improvement in start/endPrefixMapping - other? """ __author__ = "St�phane Bidoul <sbi@skynet.be>" __version__ = "0.3" import sys import codecs if sys.version_info[0] < 3: __author__ = codecs.unicode_escape_decode(__author__)[0] StringTypes = (str, unicode) # libxml2 returns strings as UTF8 _decoder = codecs.lookup("utf8")[1] def _d(s): if s is None: return s else: return _decoder(s)[0] else: StringTypes = str # s is Unicode `str` already def _d(s): return s from xml.sax._exceptions import * from xml.sax import xmlreader, saxutils from xml.sax.handler import \ feature_namespaces, \ feature_namespace_prefixes, \ feature_string_interning, \ feature_validation, \ feature_external_ges, \ feature_external_pes, \ property_lexical_handler, \ property_declaration_handler, \ property_dom_node, \ property_xml_string try: import libxml2 except ImportError: raise SAXReaderNotAvailable("libxml2 not available: " \ "import error was: %s" % sys.exc_info()[1]) class Locator(xmlreader.Locator): """SAX Locator adapter for libxml2.xmlTextReaderLocator""" def __init__(self,locator): self.__locator = locator def getColumnNumber(self): "Return the column number where the current event ends." return -1 def getLineNumber(self): "Return the line number where the current event ends." return self.__locator.LineNumber() def getPublicId(self): "Return the public identifier for the current event." return None def getSystemId(self): "Return the system identifier for the current event." return self.__locator.BaseURI() class LibXml2Reader(xmlreader.XMLReader): def __init__(self): xmlreader.XMLReader.__init__(self) # features self.__ns = 0 self.__nspfx = 0 self.__validate = 0 self.__extparams = 1 # parsing flag self.__parsing = 0 # additional handlers self.__lex_handler = None self.__decl_handler = None # error messages accumulator self.__errors = None def _errorHandler(self,arg,msg,severity,locator): if self.__errors is None: self.__errors = [] self.__errors.append((severity, SAXParseException(msg,None, Locator(locator)))) def _reportErrors(self,fatal): for severity,exception in self.__errors: if severity in (libxml2.PARSER_SEVERITY_VALIDITY_WARNING, libxml2.PARSER_SEVERITY_WARNING): self._err_handler.warning(exception) else: # when fatal is set, the parse will stop; # we consider that the last error reported # is the fatal one. if fatal and exception is self.__errors[-1][1]: self._err_handler.fatalError(exception) else: self._err_handler.error(exception) self.__errors = None def parse(self, source): self.__parsing = 1 try: # prepare source and create reader if isinstance(source, StringTypes): reader = libxml2.newTextReaderFilename(source) else: source = saxutils.prepare_input_source(source) input = libxml2.inputBuffer(source.getByteStream()) reader = input.newTextReader(source.getSystemId()) reader.SetErrorHandler(self._errorHandler,None) # configure reader if self.__extparams: reader.SetParserProp(libxml2.PARSER_LOADDTD,1) reader.SetParserProp(libxml2.PARSER_DEFAULTATTRS,1) reader.SetParserProp(libxml2.PARSER_SUBST_ENTITIES,1) reader.SetParserProp(libxml2.PARSER_VALIDATE,self.__validate) else: reader.SetParserProp(libxml2.PARSER_LOADDTD, 0) # we reuse attribute maps (for a slight performance gain) if self.__ns: attributesNSImpl = xmlreader.AttributesNSImpl({},{}) else: attributesImpl = xmlreader.AttributesImpl({}) # prefixes to pop (for endPrefixMapping) prefixes = [] # start loop self._cont_handler.startDocument() while 1: r = reader.Read() # check for errors if r == 1: if not self.__errors is None: self._reportErrors(0) elif r == 0: if not self.__errors is None: self._reportErrors(0) break # end of parse else: if not self.__errors is None: self._reportErrors(1) else: self._err_handler.fatalError(\ SAXException("Read failed (no details available)")) break # fatal parse error # get node type nodeType = reader.NodeType() # Element if nodeType == 1: if self.__ns: eltName = (_d(reader.NamespaceUri()),\ _d(reader.LocalName())) eltQName = _d(reader.Name()) attributesNSImpl._attrs = attrs = {} attributesNSImpl._qnames = qnames = {} newPrefixes = [] while reader.MoveToNextAttribute(): qname = _d(reader.Name()) value = _d(reader.Value()) if qname.startswith("xmlns"): if len(qname) > 5: newPrefix = qname[6:] else: newPrefix = None newPrefixes.append(newPrefix) self._cont_handler.startPrefixMapping(\ newPrefix,value) if not self.__nspfx: continue # don't report xmlns attribute attName = (_d(reader.NamespaceUri()), _d(reader.LocalName())) qnames[attName] = qname attrs[attName] = value reader.MoveToElement() self._cont_handler.startElementNS( \ eltName,eltQName,attributesNSImpl) if reader.IsEmptyElement(): self._cont_handler.endElementNS(eltName,eltQName) for newPrefix in newPrefixes: self._cont_handler.endPrefixMapping(newPrefix) else: prefixes.append(newPrefixes) else: eltName = _d(reader.Name()) attributesImpl._attrs = attrs = {} while reader.MoveToNextAttribute(): attName = _d(reader.Name()) attrs[attName] = _d(reader.Value()) reader.MoveToElement() self._cont_handler.startElement( \ eltName,attributesImpl) if reader.IsEmptyElement(): self._cont_handler.endElement(eltName) # EndElement elif nodeType == 15: if self.__ns: self._cont_handler.endElementNS( \ (_d(reader.NamespaceUri()),_d(reader.LocalName())), _d(reader.Name())) for prefix in prefixes.pop(): self._cont_handler.endPrefixMapping(prefix) else: self._cont_handler.endElement(_d(reader.Name())) # Text elif nodeType == 3: self._cont_handler.characters(_d(reader.Value())) # Whitespace elif nodeType == 13: self._cont_handler.ignorableWhitespace(_d(reader.Value())) # SignificantWhitespace elif nodeType == 14: self._cont_handler.characters(_d(reader.Value())) # CDATA elif nodeType == 4: if not self.__lex_handler is None: self.__lex_handler.startCDATA() self._cont_handler.characters(_d(reader.Value())) if not self.__lex_handler is None: self.__lex_handler.endCDATA() # EntityReference elif nodeType == 5: if not self.__lex_handler is None: self.startEntity(_d(reader.Name())) reader.ResolveEntity() # EndEntity elif nodeType == 16: if not self.__lex_handler is None: self.endEntity(_d(reader.Name())) # ProcessingInstruction elif nodeType == 7: self._cont_handler.processingInstruction( \ _d(reader.Name()),_d(reader.Value())) # Comment elif nodeType == 8: if not self.__lex_handler is None: self.__lex_handler.comment(_d(reader.Value())) # DocumentType elif nodeType == 10: #if not self.__lex_handler is None: # self.__lex_handler.startDTD() pass # TODO (how to detect endDTD? on first non-dtd event?) # XmlDeclaration elif nodeType == 17: pass # TODO # Entity elif nodeType == 6: pass # TODO (entity decl) # Notation (decl) elif nodeType == 12: pass # TODO # Attribute (never in this loop) #elif nodeType == 2: # pass # Document (not exposed) #elif nodeType == 9: # pass # DocumentFragment (never returned by XmlReader) #elif nodeType == 11: # pass # None #elif nodeType == 0: # pass # - else: raise SAXException("Unexpected node type %d" % nodeType) if r == 0: self._cont_handler.endDocument() reader.Close() finally: self.__parsing = 0 def setDTDHandler(self, handler): # TODO (when supported, the inherited method works just fine) raise SAXNotSupportedException("DTDHandler not supported") def setEntityResolver(self, resolver): # TODO (when supported, the inherited method works just fine) raise SAXNotSupportedException("EntityResolver not supported") def getFeature(self, name): if name == feature_namespaces: return self.__ns elif name == feature_namespace_prefixes: return self.__nspfx elif name == feature_validation: return self.__validate elif name == feature_external_ges: return 1 # TODO (does that relate to PARSER_LOADDTD)? elif name == feature_external_pes: return self.__extparams else: raise SAXNotRecognizedException("Feature '%s' not recognized" % \ name) def setFeature(self, name, state): if self.__parsing: raise SAXNotSupportedException("Cannot set feature %s " \ "while parsing" % name) if name == feature_namespaces: self.__ns = state elif name == feature_namespace_prefixes: self.__nspfx = state elif name == feature_validation: self.__validate = state elif name == feature_external_ges: if state == 0: # TODO (does that relate to PARSER_LOADDTD)? raise SAXNotSupportedException("Feature '%s' not supported" % \ name) elif name == feature_external_pes: self.__extparams = state else: raise SAXNotRecognizedException("Feature '%s' not recognized" % \ name) def getProperty(self, name): if name == property_lexical_handler: return self.__lex_handler elif name == property_declaration_handler: return self.__decl_handler else: raise SAXNotRecognizedException("Property '%s' not recognized" % \ name) def setProperty(self, name, value): if name == property_lexical_handler: self.__lex_handler = value elif name == property_declaration_handler: # TODO: remove if/when libxml2 supports dtd events raise SAXNotSupportedException("Property '%s' not supported" % \ name) self.__decl_handler = value else: raise SAXNotRecognizedException("Property '%s' not recognized" % \ name) def create_parser(): return LibXml2Reader()
Save File
Cancel