Skip to content

Commit

Permalink
refactor: review -- static only classes to methods
Browse files Browse the repository at this point in the history
  • Loading branch information
Hellgartner committed Jan 30, 2025
1 parent ea536f1 commit 30bd42a
Show file tree
Hide file tree
Showing 19 changed files with 513 additions and 559 deletions.
13 changes: 7 additions & 6 deletions src/opossum_lib/cli.py
Original file line number Diff line number Diff line change
@@ -1,19 +1,20 @@
#!/usr/bin/env python3
import logging
import sys
from pathlib import Path

# SPDX-FileCopyrightText: TNG Technology Consulting GmbH <https://www.tngtech.com>
#
# SPDX-License-Identifier: Apache-2.0
import logging
import sys
from pathlib import Path

import click

from opossum_lib.core.services.input_reader import InputReader
from opossum_lib.core.services.opossum_generator import (
generate_impl,
)
from opossum_lib.input_formats.opossum.services.opossum_format_reader import (
OpossumFormatReader,
from opossum_lib.input_formats.opossum.services.opossum_file_reader import (
OpossumFileReader,
)
from opossum_lib.input_formats.scancode.services.scancode_format_reader import (
ScancodeFormatReader,
Expand Down Expand Up @@ -74,7 +75,7 @@ def generate(
sys.exit(1)
input_readers: list[InputReader] = []
input_readers += [ScancodeFormatReader(path=path) for path in scancode_json_files]
input_readers += [OpossumFormatReader(path=path) for path in opossum_files]
input_readers += [OpossumFileReader(path=path) for path in opossum_files]

generate_impl(input_readers=input_readers, output_file=Path(outfile))

Expand Down
66 changes: 30 additions & 36 deletions src/opossum_lib/core/services/opossum_file_writer.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,43 +12,37 @@
from opossum_lib.shared.entities.opossum_file_model import OpossumFileModel


class OpossumFileWriter:
@staticmethod
def write(opossum_file_model: OpossumFileModel, file_path: Path) -> None:
file_path = OpossumFileWriter._ensure_outfile_suffix(file_path)
with ZipFile(
file_path, "w", compression=ZIP_DEFLATED, compresslevel=COMPRESSION_LEVEL
) as zip_file:
OpossumFileWriter._write_input_json(opossum_file_model, zip_file)
OpossumFileWriter._write_output_json_if_existing(
opossum_file_model, zip_file
)

@staticmethod
def _write_output_json_if_existing(
opossum_file_content: OpossumFileModel, zip_file: ZipFile
) -> None:
if opossum_file_content.output_file:
zip_file.writestr(
OUTPUT_JSON_NAME,
opossum_file_content.output_file.model_dump_json(
exclude_none=True, indent=4, by_alias=True
),
)

@staticmethod
def _write_input_json(
opossum_file_model: OpossumFileModel, zip_file: ZipFile
) -> None:
def write_opossum_file(opossum_file_model: OpossumFileModel, file_path: Path) -> None:
file_path = _ensure_outfile_suffix(file_path)
with ZipFile(
file_path, "w", compression=ZIP_DEFLATED, compresslevel=COMPRESSION_LEVEL
) as zip_file:
_write_input_json(opossum_file_model, zip_file)
_write_output_json_if_existing(opossum_file_model, zip_file)


def _write_output_json_if_existing(
opossum_file_model: OpossumFileModel, zip_file: ZipFile
) -> None:
if opossum_file_model.output_file:
zip_file.writestr(
INPUT_JSON_NAME,
opossum_file_model.input_file.model_dump_json(
indent=4,
exclude_none=True,
by_alias=True,
OUTPUT_JSON_NAME,
opossum_file_model.output_file.model_dump_json(
exclude_none=True, indent=4, by_alias=True
),
)

@staticmethod
def _ensure_outfile_suffix(outfile_path: Path) -> Path:
return outfile_path.with_suffix(".opossum")

def _write_input_json(opossum_file_model: OpossumFileModel, zip_file: ZipFile) -> None:
zip_file.writestr(
INPUT_JSON_NAME,
opossum_file_model.input_file.model_dump_json(
indent=4,
exclude_none=True,
by_alias=True,
),
)


def _ensure_outfile_suffix(outfile_path: Path) -> Path:
return outfile_path.with_suffix(".opossum")
4 changes: 2 additions & 2 deletions src/opossum_lib/core/services/opossum_generator.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
from pathlib import Path

from opossum_lib.core.services.input_reader import InputReader
from opossum_lib.core.services.opossum_file_writer import OpossumFileWriter
from opossum_lib.core.services.opossum_file_writer import write_opossum_file


def generate_impl(input_readers: list[InputReader], output_file: Path) -> None:
Expand All @@ -13,4 +13,4 @@ def generate_impl(input_readers: list[InputReader], output_file: Path) -> None:
opossum = input_readers[0].read()

opossum_file_content = opossum.to_opossum_model()
OpossumFileWriter.write(opossum_file_content, output_file)
write_opossum_file(opossum_file_content, output_file)
267 changes: 267 additions & 0 deletions src/opossum_lib/input_formats/opossum/services/convert_to_opossum.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,267 @@
# SPDX-FileCopyrightText: TNG Technology Consulting GmbH <https://www.tngtech.com>
#
# SPDX-License-Identifier: Apache-2.0

from copy import deepcopy
from pathlib import PurePath

from opossum_lib.core.entities.opossum import (
BaseUrlsForSources,
ExternalAttributionSource,
FrequentLicense,
Metadata,
Opossum,
OpossumPackage,
OpossumPackageIdentifierModel,
Resource,
ResourceType,
ScanResults,
SourceInfo,
_convert_path_to_str,
)
from opossum_lib.shared.entities.opossum_file_model import OpossumFileModel
from opossum_lib.shared.entities.opossum_input_file_model import (
ExternalAttributionSourceModel as FileExternalAttributionSource,
)
from opossum_lib.shared.entities.opossum_input_file_model import (
FrequentLicenseModel,
MetadataModel,
OpossumInputFileModel,
OpossumPackageModel,
ResourceInFileModel,
ResourcePathModel,
SourceInfoModel,
)


def convert_to_opossum(opossum_file_model: OpossumFileModel) -> Opossum:
opossum = Opossum(
scan_results=_convert_to_scan_results(opossum_file_model.input_file),
review_results=opossum_file_model.output_file,
)
return opossum


def _convert_to_scan_results(
opossum_input_file_model: OpossumInputFileModel,
) -> ScanResults:
resources, used_attribution_ids = _convert_to_resource_tree(
resources=opossum_input_file_model.resources,
external_attributions=opossum_input_file_model.external_attributions,
resources_to_attributions=opossum_input_file_model.resources_to_attributions,
)
# fmt: off
frequent_licenses = (
opossum_input_file_model.frequent_licenses
and _convert_frequent_licenses(opossum_input_file_model.frequent_licenses)
)
# fmt: on

base_urls_for_sources = (
opossum_input_file_model.base_urls_for_sources
and BaseUrlsForSources(
**(opossum_input_file_model.base_urls_for_sources.model_dump())
)
)

file_attribution_sources = opossum_input_file_model.external_attribution_sources
external_attribution_sources = {
# noqa required due to clash between linter and formatter
name: _convert_external_attribution_source( # noqa: E501
attribution_source
)
for name, attribution_source in file_attribution_sources.items()
}

attribution_with_id = _convert_to_attribution_with_id(
opossum_input_file_model.external_attributions
)
return ScanResults(
metadata=_convert_to_metadata(opossum_input_file_model.metadata),
resources=resources,
attribution_breakpoints=deepcopy(
opossum_input_file_model.attribution_breakpoints
),
external_attribution_sources=external_attribution_sources,
frequent_licenses=frequent_licenses,
files_with_children=deepcopy(opossum_input_file_model.files_with_children),
base_urls_for_sources=base_urls_for_sources,
attribution_to_id=attribution_with_id,
unassigned_attributions=_get_unassigned_attributions(
used_attribution_ids, opossum_input_file_model.external_attributions
),
)


def _get_unassigned_attributions(
used_attribution_ids: set[OpossumPackageIdentifierModel],
external_attributions: dict[
OpossumPackageIdentifierModel,
OpossumPackageModel,
],
) -> list[OpossumPackage] | None:
available_attribution_ids = external_attributions.keys()
unused_attributions_ids = set(available_attribution_ids) - used_attribution_ids
unused_attributions = [
_convert_package(external_attributions[id]) for id in unused_attributions_ids
]
return unused_attributions


def _convert_external_attribution_source(
external_attribution_source: FileExternalAttributionSource,
) -> ExternalAttributionSource:
return ExternalAttributionSource(
name=external_attribution_source.name,
priority=external_attribution_source.priority,
is_relevant_for_preferred=external_attribution_source.is_relevant_for_preferred,
)


def _convert_frequent_licenses(
frequent_licenses_infile: list[FrequentLicenseModel],
) -> list[FrequentLicense]:
frequent_licenses: list[FrequentLicense] = [
_convert_frequent_license(license) for license in frequent_licenses_infile
]
return frequent_licenses


def _convert_to_metadata(
infile_metadata: MetadataModel,
) -> Metadata:
return Metadata(**infile_metadata.model_dump())


def _convert_to_resource_tree(
resources: ResourceInFileModel,
external_attributions: dict[
OpossumPackageIdentifierModel,
OpossumPackageModel,
],
resources_to_attributions: dict[
ResourcePathModel,
list[OpossumPackageIdentifierModel],
],
) -> tuple[list[Resource], set[OpossumPackageIdentifierModel]]:
used_attribution_ids = set()

def generate_child_resource(
current_path: PurePath,
to_insert: ResourceInFileModel,
) -> Resource:
path = current_path
current_path_as_string = _convert_path_to_str(current_path)
if not current_path_as_string.startswith("/"):
current_path_as_string = "/" + current_path_as_string
attributions, attribution_ids = _get_applicable_attributions(
current_path_as_string
)
used_attribution_ids.update(attribution_ids)
if isinstance(to_insert, int):
resource_type = ResourceType.FILE
return Resource(
type=resource_type,
path=path,
attributions=attributions,
)
else:
resource_type = ResourceType.FOLDER
return Resource(
type=resource_type,
path=path,
attributions=attributions,
children={
relative_path: generate_child_resource(
current_path / relative_path, child
)
for relative_path, child in to_insert.items()
},
)

def _get_applicable_attributions(
current_path_as_string: str,
) -> tuple[list[OpossumPackage], set[OpossumPackageIdentifierModel]]:
attributions = []
attribution_ids: list[str] = []
if current_path_as_string in resources_to_attributions:
attribution_ids = resources_to_attributions[current_path_as_string]
attributions = [
_convert_package(external_attributions[id]) for id in attribution_ids
]
return attributions, set(attribution_ids)

root_path = PurePath("")

if isinstance(resources, dict):
return [
generate_child_resource(root_path / relative_path, child)
for relative_path, child in resources.items()
], used_attribution_ids
else:
raise RuntimeError("Root node must not be of file type")


def _convert_to_attribution_with_id(
external_attributions: dict[
OpossumPackageIdentifierModel,
OpossumPackageModel,
],
) -> dict[OpossumPackage, str]:
result = {}
for package_identifier, package in external_attributions.items():
converted_package = _convert_package(package)
if converted_package not in result:
result[converted_package] = package_identifier
else:
raise RuntimeError(
"An attribution was duplicated in the scan breaking internal assertions"
)
return result


def _convert_frequent_license(
infile_frequent_license: FrequentLicenseModel,
) -> FrequentLicense:
return FrequentLicense(
full_name=infile_frequent_license.full_name,
short_name=infile_frequent_license.short_name,
default_text=infile_frequent_license.default_text,
)


def _convert_package(
infile_package: OpossumPackageModel,
) -> OpossumPackage:
return OpossumPackage(
source=_convert_source(infile_package.source),
attribution_confidence=infile_package.attribution_confidence,
comment=infile_package.comment,
package_name=infile_package.package_name,
package_version=infile_package.package_version,
package_namespace=infile_package.package_namespace,
package_type=infile_package.package_type,
package_purl_appendix=infile_package.package_p_u_r_l_appendix,
copyright=infile_package.copyright,
license_name=infile_package.license_name,
license_text=infile_package.license_text,
url=infile_package.url,
first_party=infile_package.first_party,
exclude_from_notice=infile_package.exclude_from_notice,
pre_selected=infile_package.pre_selected,
follow_up=infile_package.follow_up,
origin_id=infile_package.origin_id,
origin_ids=infile_package.origin_ids,
criticality=infile_package.criticality,
was_preferred=infile_package.was_preferred,
)


def _convert_source(
infile_source_info: SourceInfoModel,
) -> SourceInfo:
return SourceInfo(
name=infile_source_info.name,
document_confidence=infile_source_info.document_confidence,
additional_name=infile_source_info.additional_name,
)
Loading

0 comments on commit 30bd42a

Please sign in to comment.