Changeset View
Changeset View
Standalone View
Standalone View
swh/indexer/metadata_dictionary/python.py
| # Copyright (C) 2018-2019 The Software Heritage developers | # Copyright (C) 2018-2019 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| import email.parser | import email.parser | ||||
| import email.policy | import email.policy | ||||
| import itertools | import itertools | ||||
| from swh.indexer.codemeta import CROSSWALK_TABLE, SCHEMA_URI | from swh.indexer.codemeta import CROSSWALK_TABLE | ||||
| from swh.indexer.namespaces import SCHEMA | |||||
| from .base import DictMapping, SingleFileIntrinsicMapping | from .base import DictMapping, SingleFileIntrinsicMapping | ||||
| _normalize_pkginfo_key = str.lower | _normalize_pkginfo_key = str.lower | ||||
| class LinebreakPreservingEmailPolicy(email.policy.EmailPolicy): | class LinebreakPreservingEmailPolicy(email.policy.EmailPolicy): | ||||
| def header_fetch_parse(self, name, value): | def header_fetch_parse(self, name, value): | ||||
| Show All 28 Lines | class PythonPkginfoMapping(DictMapping, SingleFileIntrinsicMapping): | ||||
| def translate(self, content): | def translate(self, content): | ||||
| msg = self._parser.parsebytes(content) | msg = self._parser.parsebytes(content) | ||||
| d = {} | d = {} | ||||
| for (key, value) in msg.items(): | for (key, value) in msg.items(): | ||||
| key = _normalize_pkginfo_key(key) | key = _normalize_pkginfo_key(key) | ||||
| if value != "UNKNOWN": | if value != "UNKNOWN": | ||||
| d.setdefault(key, []).append(value) | d.setdefault(key, []).append(value) | ||||
| metadata = self._translate_dict(d, normalize=False) | metadata = self._translate_dict(d, normalize=False) | ||||
| if SCHEMA_URI + "author" in metadata or SCHEMA_URI + "email" in metadata: | if SCHEMA.author in metadata or SCHEMA.email in metadata: | ||||
| metadata[SCHEMA_URI + "author"] = { | metadata[SCHEMA.author] = { | ||||
| "@list": [ | "@list": [ | ||||
| { | { | ||||
| "@type": SCHEMA_URI + "Person", | "@type": SCHEMA.Person, | ||||
| SCHEMA_URI | SCHEMA.name: metadata.pop(SCHEMA.author, [None])[0], | ||||
| + "name": metadata.pop(SCHEMA_URI + "author", [None])[0], | SCHEMA.email: metadata.pop(SCHEMA.email, [None])[0], | ||||
| SCHEMA_URI | |||||
| + "email": metadata.pop(SCHEMA_URI + "email", [None])[0], | |||||
| } | } | ||||
| ] | ] | ||||
| } | } | ||||
| return self.normalize_translation(metadata) | return self.normalize_translation(metadata) | ||||
| def normalize_home_page(self, urls): | def normalize_home_page(self, urls): | ||||
| return [{"@id": url} for url in urls] | return [{"@id": url} for url in urls] | ||||
| def normalize_keywords(self, keywords): | def normalize_keywords(self, keywords): | ||||
| return list(itertools.chain.from_iterable(s.split(" ") for s in keywords)) | return list(itertools.chain.from_iterable(s.split(" ") for s in keywords)) | ||||
| def normalize_license(self, licenses): | def normalize_license(self, licenses): | ||||
| return [{"@id": license} for license in licenses] | return [{"@id": license} for license in licenses] | ||||