Changeset View
Changeset View
Standalone View
Standalone View
swh/indexer/metadata_dictionary/ruby.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 ast | import ast | ||||
| import itertools | import itertools | ||||
| import re | import re | ||||
| from typing import Any, Dict, List, Optional | from typing import Any, Dict, List, Optional, Tuple, Union | ||||
| from swh.indexer.codemeta import CROSSWALK_TABLE, SCHEMA_URI | from swh.indexer.codemeta import CROSSWALK_TABLE, SCHEMA_URI | ||||
| from .base import DictMapping | from .base import DictMapping, File_entries | ||||
| def name_to_person(name: str) -> Dict[str, str]: | def name_to_person(name: str) -> Dict[str, str]: | ||||
| return { | return { | ||||
| "@type": SCHEMA_URI + "Person", | "@type": SCHEMA_URI + "Person", | ||||
| SCHEMA_URI + "name": name, | SCHEMA_URI + "name": name, | ||||
| } | } | ||||
| class GemspecMapping(DictMapping): | class GemspecMapping(DictMapping): | ||||
| name = "gemspec" | name = "gemspec" | ||||
| mapping = CROSSWALK_TABLE["Ruby Gem"] | mapping = CROSSWALK_TABLE["Ruby Gem"] | ||||
| string_fields = ["name", "version", "description", "summary", "email"] | string_fields = ["name", "version", "description", "summary", "email"] | ||||
| _re_spec_new = re.compile(r".*Gem::Specification.new +(do|\{) +\|.*\|.*") | _re_spec_new = re.compile(r".*Gem::Specification.new +(do|\{) +\|.*\|.*") | ||||
| _re_spec_entry = re.compile(r"\s*\w+\.(?P<key>\w+)\s*=\s*(?P<expr>.*)") | _re_spec_entry = re.compile(r"\s*\w+\.(?P<key>\w+)\s*=\s*(?P<expr>.*)") | ||||
| @classmethod | @classmethod | ||||
| def detect_metadata_files(cls: Any, file_entries: Any) -> List[str]: | def detect_metadata_files( | ||||
| cls: Any, file_entries: List[File_entries] | |||||
| ) -> List[bytes]: | |||||
| for entry in file_entries: | for entry in file_entries: | ||||
| if isinstance(entry["name"], bytes): | |||||
| if entry["name"].endswith(b".gemspec"): | if entry["name"].endswith(b".gemspec"): | ||||
| return [entry["sha1"]] | return [entry["sha1"]] | ||||
| return [] | return [] | ||||
| def translate(self, raw_content: Any) -> Optional[Dict[str, str]]: | def translate( | ||||
| self, raw_content_bytes: bytes | |||||
| ) -> Optional[Dict[str, Union[str, List[Any], Dict[str, Any], Tuple[str]]]]: | |||||
| try: | try: | ||||
| raw_content = raw_content.decode() | raw_content = raw_content_bytes.decode() | ||||
| except UnicodeDecodeError: | except UnicodeDecodeError: | ||||
| self.log.warning("Error unidecoding from %s", self.log_suffix) | self.log.warning("Error unidecoding from %s", self.log_suffix) | ||||
| return None | return None | ||||
| # Skip lines before 'Gem::Specification.new' | # Skip lines before 'Gem::Specification.new' | ||||
| lines = itertools.dropwhile( | lines = itertools.dropwhile( | ||||
| lambda x: not self._re_spec_new.match(x), raw_content.split("\n") | lambda x: not self._re_spec_new.match(x), raw_content.split("\n") | ||||
| ) | ) | ||||
| try: | try: | ||||
| next(lines) # Consume 'Gem::Specification.new' | next(lines) # Consume 'Gem::Specification.new' | ||||
| except StopIteration: | except StopIteration: | ||||
| self.log.warning("Could not find Gem::Specification in %s", self.log_suffix) | self.log.warning("Could not find Gem::Specification in %s", self.log_suffix) | ||||
| return None | return None | ||||
| content_dict = {} | content_dict: Dict[str, Union[str, List[str]]] = {} | ||||
| for line in lines: | for line in lines: | ||||
| match = self._re_spec_entry.match(line) | match = self._re_spec_entry.match(line) | ||||
| if match: | if match: | ||||
| value = self.eval_ruby_expression(match.group("expr")) | value = self.eval_ruby_expression(match.group("expr")) | ||||
| if value: | if value: | ||||
| content_dict[match.group("key")] = value | content_dict[match.group("key")] = value | ||||
| return self._translate_dict(content_dict) | return self._translate_dict(content_dict) | ||||
| def eval_ruby_expression(self, expr: str) -> Any: | def eval_ruby_expression(self, expr: str) -> Optional[Union[str, List[str]]]: | ||||
| """Very simple evaluator of Ruby expressions. | """Very simple evaluator of Ruby expressions. | ||||
| >>> GemspecMapping().eval_ruby_expression('"Foo bar"') | >>> GemspecMapping().eval_ruby_expression('"Foo bar"') | ||||
| 'Foo bar' | 'Foo bar' | ||||
| >>> GemspecMapping().eval_ruby_expression("'Foo bar'") | >>> GemspecMapping().eval_ruby_expression("'Foo bar'") | ||||
| 'Foo bar' | 'Foo bar' | ||||
| >>> GemspecMapping().eval_ruby_expression("['Foo', 'bar']") | >>> GemspecMapping().eval_ruby_expression("['Foo', 'bar']") | ||||
| ['Foo', 'bar'] | ['Foo', 'bar'] | ||||
| Show All 22 Lines | def eval_ruby_expression(self, expr: str) -> Optional[Union[str, List[str]]]: | ||||
| # ast.parse works for very simple Ruby expressions | # ast.parse works for very simple Ruby expressions | ||||
| # (mainly strings delimited with " or ', and lists | # (mainly strings delimited with " or ', and lists | ||||
| # of such strings). | # of such strings). | ||||
| tree = ast.parse(expr, mode="eval") | tree = ast.parse(expr, mode="eval") | ||||
| except (SyntaxError, ValueError): | except (SyntaxError, ValueError): | ||||
| return None | return None | ||||
| if isinstance(tree, ast.Expression): | if isinstance(tree, ast.Expression): | ||||
| return evaluator(tree.body) | return evaluator(tree.body) | ||||
| else: | |||||
| return None | |||||
| def normalize_homepage(self, s: str) -> Dict[str, str]: | def normalize_homepage(self, s: str) -> Dict[str, str]: | ||||
| if isinstance(s, str): | if isinstance(s, str): | ||||
| return {"@id": s} | return {"@id": s} | ||||
| def normalize_license(self, s: str) -> List[Dict[str, str]]: | def normalize_license(self, s: str) -> List[Dict[str, str]]: | ||||
| if isinstance(s, str): | if isinstance(s, str): | ||||
| return [{"@id": "https://spdx.org/licenses/" + s}] | return [{"@id": "https://spdx.org/licenses/" + s}] | ||||
| def normalize_licenses(self, licenses: List[str]) -> Any: | def normalize_licenses(self, licenses: List[str]) -> List[Dict[str, str]]: | ||||
| if isinstance(licenses, list): | if isinstance(licenses, list): | ||||
| return [ | return [ | ||||
| {"@id": "https://spdx.org/licenses/" + license} | {"@id": "https://spdx.org/licenses/" + license} | ||||
| for license in licenses | for license in licenses | ||||
| if isinstance(license, str) | if isinstance(license, str) | ||||
| ] | ] | ||||
| return [] | |||||
| def normalize_author(self, author: str) -> Any: | def normalize_author(self, author: str) -> Dict[str, List[Dict[str, str]]]: | ||||
| if isinstance(author, str): | if isinstance(author, str): | ||||
| return {"@list": [name_to_person(author)]} | return {"@list": [name_to_person(author)]} | ||||
| def normalize_authors(self, authors: List[str]) -> Any: | def normalize_authors( | ||||
| self, authors: List[str] | |||||
| ) -> Optional[Dict[str, List[Dict[str, str]]]]: | |||||
| if isinstance(authors, list): | if isinstance(authors, list): | ||||
| return { | return { | ||||
| "@list": [ | "@list": [ | ||||
| name_to_person(author) | name_to_person(author) | ||||
| for author in authors | for author in authors | ||||
| if isinstance(author, str) | if isinstance(author, str) | ||||
| ] | ] | ||||
| } | } | ||||
| else: | |||||
| return None | |||||