Changeset View
Changeset View
Standalone View
Standalone View
swh/scanner/model.py
# Copyright (C) 2020 The Software Heritage developers | # Copyright (C) 2020 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 | ||||
from __future__ import annotations | from __future__ import annotations | ||||
from enum import Enum | from enum import Enum | ||||
import json | import json | ||||
from pathlib import Path | from pathlib import Path | ||||
import sys | import sys | ||||
from typing import Any, Dict, Iterable, List, Tuple | from typing import Any, Dict, Iterator, List, Tuple | ||||
import ndjson | import ndjson | ||||
from swh.model.identifiers import CONTENT, DIRECTORY | from swh.model.identifiers import CONTENT, DIRECTORY | ||||
from .exceptions import InvalidDirectoryPath, InvalidObjectType | from .exceptions import InvalidDirectoryPath, InvalidObjectType | ||||
from .plot import generate_sunburst, offline_plot | from .plot import generate_sunburst, offline_plot | ||||
▲ Show 20 Lines • Show All 74 Lines • ▼ Show 20 Lines | def printNode(self, node: Any, isatty: bool, inc: int) -> None: | ||||
elif node.otype == DIRECTORY: | elif node.otype == DIRECTORY: | ||||
rel_path = colorize(rel_path, Color.blue) | rel_path = colorize(rel_path, Color.blue) | ||||
elif node.otype == CONTENT: | elif node.otype == CONTENT: | ||||
rel_path = colorize(rel_path, Color.green) | rel_path = colorize(rel_path, Color.green) | ||||
print(f"{begin}{rel_path}{end}") | print(f"{begin}{rel_path}{end}") | ||||
@property | @property | ||||
def attributes(self): | def attributes(self) -> Dict[str, Dict[str, Any]]: | ||||
""" | """ | ||||
Get the attributes of the current node grouped by the relative path. | Get the attributes of the current node grouped by the relative path. | ||||
Returns: | Returns: | ||||
a dictionary containing a path as key and its known/unknown status and the | a dictionary containing a path as key and its known/unknown status and the | ||||
SWHID as values. | SWHID as values. | ||||
""" | """ | ||||
return {str(self.path): {"swhid": self.swhid, "known": self.known,}} | return {str(self.path): {"swhid": self.swhid, "known": self.known,}} | ||||
def toDict(self, dict_nodes={}) -> Dict[str, Dict[str, Dict]]: | def toDict(self) -> Dict[str, Dict[str, Any]]: | ||||
""" | """ | ||||
Recursively groups the current child nodes inside a dictionary. | Recursively flatten the current tree nodes into a dictionary. | ||||
For example, if you have the following structure: | For example, if you have the following structure: | ||||
.. code-block:: none | .. code-block:: none | ||||
root { | root { | ||||
subdir: { | subdir: { | ||||
file.txt | file.txt | ||||
Show All 16 Lines | def toDict(self) -> Dict[str, Dict[str, Any]]: | ||||
"root/subdir/file.txt": { | "root/subdir/file.txt": { | ||||
"swhid": "...", | "swhid": "...", | ||||
"known": True/False | "known": True/False | ||||
} | } | ||||
} | } | ||||
""" | """ | ||||
for node_dict in self.__iterNodesAttr(): | return {k: v for d in self.__iterNodesAttr() for k, v in d.items()} | ||||
dict_nodes.update(node_dict) | |||||
return dict_nodes | |||||
def iterate(self) -> Iterable[Tree]: | def iterate(self) -> Iterator[Tree]: | ||||
""" | """ | ||||
Recursively iterate through the children of the current node | Recursively iterate through the children of the current node | ||||
""" | """ | ||||
for _, child_node in self.children.items(): | for _, child_node in self.children.items(): | ||||
yield child_node | yield child_node | ||||
if child_node.otype == DIRECTORY: | if child_node.otype == DIRECTORY: | ||||
yield from child_node.iterate() | yield from child_node.iterate() | ||||
def __iterNodesAttr(self) -> Iterable[Dict[str, Dict]]: | def __iterNodesAttr(self) -> Iterator[Dict[str, Dict[str, Any]]]: | ||||
""" | """ | ||||
Recursively iterate through the children of the current node returning | Recursively iterate through the children of the current node returning | ||||
an iterable of the children nodes attributes | an iterable of the children nodes attributes | ||||
Yields: | Yields: | ||||
a dictionary containing a path with its known/unknown status and the | a dictionary containing a path with its known/unknown status and the | ||||
SWHID | SWHID | ||||
""" | """ | ||||
▲ Show 20 Lines • Show All 96 Lines • Show Last 20 Lines |