| 210 | |
| 211 | |
| 212 | class DOMEntityResolver(object): |
| 213 | __slots__ = '_opener', |
| 214 | |
| 215 | def resolveEntity(self, publicId, systemId): |
| 216 | assert systemId is not None |
| 217 | source = DOMInputSource() |
| 218 | source.publicId = publicId |
| 219 | source.systemId = systemId |
| 220 | source.byteStream = self._get_opener().open(systemId) |
| 221 | |
| 222 | # determine the encoding if the transport provided it |
| 223 | source.encoding = self._guess_media_encoding(source) |
| 224 | |
| 225 | # determine the base URI is we can |
| 226 | import posixpath, urllib.parse |
| 227 | parts = urllib.parse.urlparse(systemId) |
| 228 | scheme, netloc, path, params, query, fragment = parts |
| 229 | # XXX should we check the scheme here as well? |
| 230 | if path and not path.endswith("/"): |
| 231 | path = posixpath.dirname(path) + "/" |
| 232 | parts = scheme, netloc, path, params, query, fragment |
| 233 | source.baseURI = urllib.parse.urlunparse(parts) |
| 234 | |
| 235 | return source |
| 236 | |
| 237 | def _get_opener(self): |
| 238 | try: |
| 239 | return self._opener |
| 240 | except AttributeError: |
| 241 | self._opener = self._create_opener() |
| 242 | return self._opener |
| 243 | |
| 244 | def _create_opener(self): |
| 245 | import urllib.request |
| 246 | return urllib.request.build_opener() |
| 247 | |
| 248 | def _guess_media_encoding(self, source): |
| 249 | info = source.byteStream.info() |
| 250 | # import email.message |
| 251 | # assert isinstance(info, email.message.Message) |
| 252 | charset = info.get_param('charset') |
| 253 | if charset is not None: |
| 254 | return charset.lower() |
| 255 | return None |
| 256 | |
| 257 | |
| 258 | class DOMInputSource(object): |
no outgoing calls
no test coverage detected
searching dependent graphs…