| from os.path import dirname, join |
| |
| from collections import OrderedDict |
| |
| from xml.parsers import expat |
| import xml.etree.ElementTree as etree |
| |
| _catalog = join(dirname(__file__), "catalog") |
| |
| def _wrap_error(e): |
| err = etree.ParseError(e) |
| err.code = e.code |
| err.position = e.lineno, e.offset |
| raise err |
| |
| _names = {} |
| def _fixname(key): |
| try: |
| name = _names[key] |
| except KeyError: |
| name = key |
| if "}" in name: |
| name = "{" + name |
| _names[key] = name |
| return name |
| |
| |
| class XMLParser(object): |
| """ |
| An XML parser with support for XHTML DTDs and all Python-supported encodings |
| |
| This implements the API defined by |
| xml.etree.ElementTree.XMLParser, but supports XHTML DTDs |
| (therefore allowing XHTML entities) and supports all encodings |
| Python does, rather than just those supported by expat. |
| """ |
| def __init__(self, encoding=None): |
| self._parser = expat.ParserCreate(encoding, "}") |
| self._target = etree.TreeBuilder() |
| # parser settings |
| self._parser.buffer_text = 1 |
| self._parser.ordered_attributes = 1 |
| self._parser.SetParamEntityParsing(expat.XML_PARAM_ENTITY_PARSING_UNLESS_STANDALONE) |
| # parser callbacks |
| self._parser.XmlDeclHandler = self._xml_decl |
| self._parser.StartElementHandler = self._start |
| self._parser.EndElementHandler = self._end |
| self._parser.CharacterDataHandler = self._data |
| self._parser.ExternalEntityRefHandler = self._external |
| self._parser.SkippedEntityHandler = self._skipped |
| # used for our horrible re-encoding hack |
| self._fed_data = [] |
| self._read_encoding = None |
| |
| def _xml_decl(self, version, encoding, standalone): |
| self._read_encoding = encoding |
| |
| def _start(self, tag, attrib_in): |
| self._fed_data = None |
| tag = _fixname(tag) |
| attrib = OrderedDict() |
| if attrib_in: |
| for i in range(0, len(attrib_in), 2): |
| attrib[_fixname(attrib_in[i])] = attrib_in[i+1] |
| return self._target.start(tag, attrib) |
| |
| def _data(self, text): |
| return self._target.data(text) |
| |
| def _end(self, tag): |
| return self._target.end(_fixname(tag)) |
| |
| def _external(self, context, base, systemId, publicId): |
| if publicId in { |
| "-//W3C//DTD XHTML 1.0 Transitional//EN", |
| "-//W3C//DTD XHTML 1.1//EN", |
| "-//W3C//DTD XHTML 1.0 Strict//EN", |
| "-//W3C//DTD XHTML 1.0 Frameset//EN", |
| "-//W3C//DTD XHTML Basic 1.0//EN", |
| "-//W3C//DTD XHTML 1.1 plus MathML 2.0//EN", |
| "-//W3C//DTD XHTML 1.1 plus MathML 2.0 plus SVG 1.1//EN", |
| "-//W3C//DTD MathML 2.0//EN", |
| "-//WAPFORUM//DTD XHTML Mobile 1.0//EN" |
| }: |
| parser = self._parser.ExternalEntityParserCreate(context) |
| with open(join(_catalog, "xhtml.dtd"), "rb") as fp: |
| try: |
| parser.ParseFile(fp) |
| except expat.error: |
| return False |
| |
| return True |
| |
| def _skipped(self, name, is_parameter_entity): |
| err = expat.error("undefined entity %s: line %d, column %d" % |
| (name, self._parser.ErrorLineNumber, |
| self._parser.ErrorColumnNumber)) |
| err.code = expat.errors.XML_ERROR_UNDEFINED_ENTITY |
| err.lineno = self._parser.ErrorLineNumber |
| err.offset = self._parser.ErrorColumnNumber |
| raise err |
| |
| def feed(self, data): |
| if self._fed_data is not None: |
| self._fed_data.append(data) |
| try: |
| self._parser.Parse(data, False) |
| except expat.error as v: |
| _wrap_error(v) |
| except ValueError as e: |
| if e.args[0] == 'multi-byte encodings are not supported': |
| assert self._read_encoding is not None |
| xml = b"".join(self._fed_data).decode(self._read_encoding).encode("utf-8") |
| new_parser = XMLParser("utf-8") |
| self._parser = new_parser._parser |
| self._target = new_parser._target |
| self._fed_data = None |
| self.feed(xml) |
| |
| def close(self): |
| try: |
| self._parser.Parse("", True) |
| except expat.error as v: |
| _wrap_error(v) |
| tree = self._target.close() |
| return tree |
