Source code for swh.indexer.metadata_dictionary.python

# Copyright (C) 2018-2021  The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information

import email.parser
import email.policy

from rdflib import BNode, Literal, URIRef

from swh.indexer.codemeta import CROSSWALK_TABLE
from swh.indexer.namespaces import RDF, SCHEMA

from .base import DictMapping, SingleFileIntrinsicMapping
from .utils import add_list

_normalize_pkginfo_key = str.lower

[docs]class LinebreakPreservingEmailPolicy(email.policy.EmailPolicy):
[docs] def header_fetch_parse(self, name, value): if hasattr(value, "name"): return value value = value.replace("\n ", "\n") return self.header_factory(name, value)
[docs]class PythonPkginfoMapping(DictMapping, SingleFileIntrinsicMapping): """Dedicated class for Python's PKG-INFO mapping and translation.""" name = "pkg-info" filename = b"PKG-INFO" mapping = { _normalize_pkginfo_key(k): v for (k, v) in CROSSWALK_TABLE["Python PKG-INFO"].items() } string_fields = [ "name", "version", "description", "summary", "author", "author-email", ] _parser = email.parser.BytesHeaderParser(policy=LinebreakPreservingEmailPolicy())
[docs] def translate(self, content): msg = self._parser.parsebytes(content) d = {} for (key, value) in msg.items(): key = _normalize_pkginfo_key(key) if value != "UNKNOWN": d.setdefault(key, []).append(value) return self._translate_dict(d)
[docs] def extra_translation(self, graph, root, d): author_names = list(graph.triples((root,, None))) author_emails = list(graph.triples((root,, None))) graph.remove((root,, None)) graph.remove((root,, None)) if author_names or author_emails: author = BNode() graph.add((author, RDF.type, SCHEMA.Person)) for (_, _, author_name) in author_names: graph.add((author,, author_name)) for (_, _, author_email) in author_emails: graph.add((author,, author_email)) add_list(graph, root,, [author])
[docs] def normalize_home_page(self, urls): return [URIRef(url) for url in urls]
[docs] def normalize_keywords(self, keywords): return [Literal(keyword) for s in keywords for keyword in s.split(" ")]
[docs] def normalize_license(self, licenses): return [URIRef("" + license) for license in licenses]