<commit_before>import os
import logging
from jsongraph import Graph

from datamapper.sinks.base import Sink
from datamapper.util import ConfigException

log = logging.getLogger(__name__)


class RDFSink(Sink):
    """ Generate RDF files from the incoming records. """

    def make_context(self, record):
        graph = Graph(self.config.base_uri, resolver=self.config.resolver)
        graph.register(self.config.get_alias(record.schema), record.schema)
        for alias, schema in self.config.schemas.items():
            graph.register(alias, schema)
        return graph.context(meta=record.source.to_dict())

    def get_path(self, record, section):
        path = self.config.get('rdf_path')
        if path is None:
            raise ConfigException("No 'rdf_path' is configured.")
        doc_type = self.config.get_alias(record.schema)
        file_name = '%s.%s.nq' % (doc_type, section)
        return os.path.join(path, record.source.slug, file_name)

    def store(self, context, record, section):
        path = self.get_path(record, section)
        try:
            os.makedirs(os.path.dirname(path))
        except:
            pass
        context.save()
        log.info("Storing data to: %s", path)
        with open(path, 'w') as fh:
            graph = context.parent
            graph.graph.serialize(fh, format='nquads')

    def load(self):
        section = 1
        for mapping in self.generator.mappings:
            context = None
            for i, record in enumerate(self.generator.generate(mapping)):
                if context is None:
                    context = self.make_context(record)
                doc_type = self.config.get_alias(record.schema)
                context.add(doc_type, record.entity)
                if i > 0 and i % 10000 == 0:
                    log.info("Generaring %r RDF: %s records", doc_type, i)
                if i > 0 and i % 100000 == 0:
                    self.store(context, record, section)
                    context = None
                    section += 1
            self.store(context, record, section)<commit_msg>Write Turtle instead of NQuads. <commit_after>import os
import logging

from jsongraph.binding import Binding
from jsongraph.triplify import triplify

from datamapper.sinks.base import Sink
from datamapper.util import ConfigException

log = logging.getLogger(__name__)


class RDFSink(Sink):
    """ Generate RDF files from the incoming records. """

    def get_path(self, mapping, record):
        path = self.config.get('rdf_path')
        if path is None:
            raise ConfigException("No 'rdf_path' is configured.")
        file_name = '%s.ttl' % mapping
        return os.path.join(path, record.source.slug, file_name)

    def emit_to_file(self, fh, triples):
        output = []
        for t in triples:
            output.extend((t[0].n3(), t[1].n3(), t[2].n3(), '.\n'))
        fh.write(' '.join(output).encode('utf-8'))

    def load_mapping(self, mapping):
        outfh = None
        path = None
        schema = None
        import time
        begin = time.time()
        for i, record in enumerate(self.generator.generate(mapping)):
            if path is None:
                path = self.get_path(mapping, record)
                try:
                    os.makedirs(os.path.dirname(path))
                except:
                    pass
                outfh = open(path, 'w')
                _, schema = self.config.resolver.resolve(record.schema)

            data = record.entity
            binding = Binding(schema, self.config.resolver, data=data)
            _, triples = triplify(binding)

            if i > 0 and i % 10000 == 0:
                elapsed = time.time() - begin
                per_record = float(elapsed) / float(i)
                speed = per_record * 1000
                log.info("Generaring %r RDF: %s records (speed: %s)", mapping, i, speed)

        outfh.close()

    def load(self):
        for mapping in self.generator.mappings:
            self.load_mapping(mapping)
{'fraction_non_alphanumeric': 0.1075776397515528, 'fraction_numerical': 0.008695652173913044, 'mean_word_length': 2.008968609865471, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 6, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5288909', 'n_tokens_mistral': 1210, 'n_tokens_neox': 1131, 'n_words': 300}