"""\ N-Triples Parser License: GPL 2, W3C, BSD, or MIT Author: Sean B. Palmer, inamidst.com """ from __future__ import annotations import codecs import re from io import BytesIO, StringIO, TextIOBase from typing import ( IO, TYPE_CHECKING, Any, Match, MutableMapping, Optional, Pattern, TextIO, Union, ) from rdflib.compat import _string_escape_map, decodeUnicodeEscape from rdflib.exceptions import ParserError as ParseError from rdflib.parser import InputSource, Parser from rdflib.term import BNode as bNode from rdflib.term import Literal, URIRef from rdflib.term import URIRef as URI # noqa: N814 if TYPE_CHECKING: import typing_extensions as te from rdflib.graph import Graph, _ObjectType, _PredicateType, _SubjectType __all__ = [ "unquote", "uriquote", "W3CNTriplesParser", "NTGraphSink", "NTParser", "DummySink", ] uriref = r'<([^:]+:[^\s"<>]*)>' literal = r'"([^"\\]*(?:\\.[^"\\]*)*)"' litinfo = r"(?:@([a-zA-Z]+(?:-[a-zA-Z0-9]+)*)|\^\^" + uriref + r")?" r_line = re.compile(r"([^\r\n]*)(?:\r\n|\r|\n)") r_wspace = re.compile(r"[ \t]*") r_wspaces = re.compile(r"[ \t]+") r_tail = re.compile(r"[ \t]*\.[ \t]*(#.*)?") r_uriref = re.compile(uriref) r_nodeid = re.compile(r"_:([A-Za-z0-9_:]([-A-Za-z0-9_:\.]*[-A-Za-z0-9_:])?)") r_literal = re.compile(literal + litinfo) bufsiz = 2048 validate = False class DummySink: def __init__(self): self.length = 0 def triple(self, s, p, o): self.length += 1 print(s, p, o) r_safe = re.compile(r"([\x20\x21\x23-\x5B\x5D-\x7E]+)") r_quot = re.compile(r"""\\([tbnrf"'\\])""") r_uniquot = re.compile(r"\\u([0-9A-Fa-f]{4})|\\U([0-9A-Fa-f]{8})") def unquote(s: str) -> str: """Unquote an N-Triples string.""" if not validate: if isinstance(s, str): # nquads s = decodeUnicodeEscape(s) else: s = s.decode("unicode-escape") # type: ignore[unreachable] return s else: result = [] while s: m = r_safe.match(s) if m: s = s[m.end() :] result.append(m.group(1)) continue m = r_quot.match(s) if m: s = s[2:] result.append(_string_escape_map[m.group(1)]) continue m = r_uniquot.match(s) if m: s = s[m.end() :] u, U = m.groups() # noqa: N806 codepoint = int(u or U, 16) if codepoint > 0x10FFFF: raise ParseError("Disallowed codepoint: %08X" % codepoint) result.append(chr(codepoint)) elif s.startswith("\\"): raise ParseError("Illegal escape at: %s..." % s[:10]) else: raise ParseError("Illegal literal character: %r" % s[0]) return "".join(result) r_hibyte = re.compile(r"([\x80-\xFF])") def uriquote(uri: str) -> str: if not validate: return uri else: return r_hibyte.sub(lambda m: "%%%02X" % ord(m.group(1)), uri) _BNodeContextType = MutableMapping[str, bNode] class W3CNTriplesParser: """An N-Triples Parser. This is a legacy-style Triples parser for NTriples provided by W3C Usage:: p = W3CNTriplesParser(sink=MySink()) sink = p.parse(f) # file; use parsestring for a string To define a context in which blank node identifiers refer to the same blank node across instances of NTriplesParser, pass the same dict as ``bnode_context`` to each instance. By default, a new blank node context is created for each instance of `W3CNTriplesParser`. """ __slots__ = ("_bnode_ids", "sink", "buffer", "file", "line", "skolemize") def __init__( self, sink: Optional[Union[DummySink, NTGraphSink]] = None, bnode_context: Optional[_BNodeContextType] = None, ): self.skolemize = False if bnode_context is not None: self._bnode_ids = bnode_context else: self._bnode_ids = {} self.sink: Union[DummySink, NTGraphSink] if sink is not None: self.sink = sink else: self.sink = DummySink() self.buffer: Optional[str] = None self.file: Optional[Union[TextIO, codecs.StreamReader]] = None self.line: Optional[str] = "" def parse( self, f: Union[TextIO, IO[bytes], codecs.StreamReader], bnode_context: Optional[_BNodeContextType] = None, skolemize: bool = False, ) -> Union[DummySink, NTGraphSink]: """ Parse f as an N-Triples file. :type f: :term:`file object` :param f: the N-Triples source :type bnode_context: `dict`, optional :param bnode_context: a dict mapping blank node identifiers (e.g., ``a`` in ``_:a``) to `~rdflib.term.BNode` instances. An empty dict can be passed in to define a distinct context for a given call to `parse`. """ if not hasattr(f, "read"): raise ParseError("Item to parse must be a file-like object.") if not hasattr(f, "encoding") and not hasattr(f, "charbuffer"): # someone still using a bytestream here? f = codecs.getreader("utf-8")(f) self.skolemize = skolemize self.file = f # type: ignore[assignment] self.buffer = "" while True: self.line = self.readline() if self.line is None: break try: self.parseline(bnode_context=bnode_context) except ParseError: raise ParseError("Invalid line: {}".format(self.line)) return self.sink def parsestring(self, s: Union[bytes, bytearray, str], **kwargs) -> None: """Parse s as an N-Triples string.""" if not isinstance(s, (str, bytes, bytearray)): raise ParseError("Item to parse must be a string instance.") f: Union[codecs.StreamReader, StringIO] if isinstance(s, (bytes, bytearray)): f = codecs.getreader("utf-8")(BytesIO(s)) else: f = StringIO(s) self.parse(f, **kwargs) def readline(self) -> Optional[str]: """Read an N-Triples line from buffered input.""" # N-Triples lines end in either CRLF, CR, or LF # Therefore, we can't just use f.readline() if not self.buffer: # type error: Item "None" of "Union[TextIO, StreamReader, None]" has no attribute "read" buffer = self.file.read(bufsiz) # type: ignore[union-attr] if not buffer: return None self.buffer = buffer while True: m = r_line.match(self.buffer) if m: # the more likely prospect self.buffer = self.buffer[m.end() :] return m.group(1) else: # type error: Item "None" of "Union[TextIO, StreamReader, None]" has no attribute "read" buffer = self.file.read(bufsiz) # type: ignore[union-attr] if not buffer and not self.buffer.isspace(): # Last line does not need to be terminated with a newline buffer += "\n" elif not buffer: return None self.buffer += buffer def parseline(self, bnode_context: Optional[_BNodeContextType] = None) -> None: self.eat(r_wspace) if (not self.line) or self.line.startswith("#"): return # The line is empty or a comment subject = self.subject(bnode_context) self.eat(r_wspaces) predicate = self.predicate() self.eat(r_wspaces) object_ = self.object(bnode_context) self.eat(r_tail) if self.line: raise ParseError("Trailing garbage: {}".format(self.line)) self.sink.triple(subject, predicate, object_) def peek(self, token: str) -> bool: return self.line.startswith(token) # type: ignore[union-attr] def eat(self, pattern: Pattern[str]) -> Match[str]: m = pattern.match(self.line) # type: ignore[arg-type] if not m: # @@ Why can't we get the original pattern? # print(dir(pattern)) # print repr(self.line), type(self.line) raise ParseError("Failed to eat %s at %s" % (pattern.pattern, self.line)) self.line = self.line[m.end() :] # type: ignore[index] return m def subject(self, bnode_context=None) -> Union[bNode, URIRef]: # @@ Consider using dictionary cases subj = self.uriref() or self.nodeid(bnode_context) if not subj: raise ParseError("Subject must be uriref or nodeID") return subj def predicate(self) -> Union[bNode, URIRef]: pred = self.uriref() if not pred: raise ParseError("Predicate must be uriref") return pred def object( self, bnode_context: Optional[_BNodeContextType] = None ) -> Union[URI, bNode, Literal]: objt = self.uriref() or self.nodeid(bnode_context) or self.literal() if objt is False: raise ParseError("Unrecognised object type") return objt def uriref(self) -> Union[te.Literal[False], URI]: if self.peek("<"): uri = self.eat(r_uriref).group(1) uri = unquote(uri) uri = uriquote(uri) return URI(uri) return False def nodeid( self, bnode_context: Optional[_BNodeContextType] = None ) -> Union[te.Literal[False], bNode, URI]: if self.peek("_"): if self.skolemize: bnode_id = self.eat(r_nodeid).group(1) return bNode(bnode_id).skolemize() else: # Fix for https://github.com/RDFLib/rdflib/issues/204 if bnode_context is None: bnode_context = self._bnode_ids bnode_id = self.eat(r_nodeid).group(1) new_id = bnode_context.get(bnode_id, None) if new_id is not None: # Re-map to id specific to this doc return bNode(new_id) else: # Replace with freshly-generated document-specific BNode id bnode = bNode() # Store the mapping bnode_context[bnode_id] = bnode return bnode return False def literal(self) -> Union[te.Literal[False], Literal]: if self.peek('"'): lit, lang, dtype = self.eat(r_literal).groups() if lang: lang = lang else: lang = None if dtype: dtype = unquote(dtype) dtype = uriquote(dtype) dtype = URI(dtype) else: dtype = None if lang and dtype: raise ParseError("Can't have both a language and a datatype") lit = unquote(lit) return Literal(lit, lang, dtype) return False class NTGraphSink: __slots__ = ("g",) def __init__(self, graph: Graph): self.g = graph def triple(self, s: _SubjectType, p: _PredicateType, o: _ObjectType) -> None: self.g.add((s, p, o)) class NTParser(Parser): """parser for the ntriples format, often stored with the .nt extension See http://www.w3.org/TR/rdf-testcases/#ntriples""" __slots__ = () @classmethod def parse(cls, source: InputSource, sink: Graph, **kwargs: Any) -> None: """ Parse the NT format :type source: `rdflib.parser.InputSource` :param source: the source of NT-formatted data :type sink: `rdflib.graph.Graph` :param sink: where to send parsed triples :param kwargs: Additional arguments to pass to `.W3CNTriplesParser.parse` """ f: Union[TextIO, IO[bytes], codecs.StreamReader] f = source.getCharacterStream() if not f: b = source.getByteStream() # TextIOBase includes: StringIO and TextIOWrapper if isinstance(b, TextIOBase): # f is not really a ByteStream, but a CharacterStream f = b # type: ignore[assignment] else: # since N-Triples 1.1 files can and should be utf-8 encoded f = codecs.getreader("utf-8")(b) parser = W3CNTriplesParser(NTGraphSink(sink)) parser.parse(f, **kwargs) f.close()