52 lines
1.6 KiB
Python
52 lines
1.6 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
SPEC_ALIASES = {
|
|
"ram_gb": ("ram", "arbeitsspeicher"),
|
|
"storage_gb": ("speicher", "ssd", "festplatte"),
|
|
"weight_kg": ("gewicht",),
|
|
}
|
|
|
|
|
|
def merge_observation_attributes(
|
|
raw_attributes: dict[str, Any],
|
|
normalized_attributes: dict[str, Any],
|
|
) -> list[dict[str, Any]]:
|
|
items: list[dict[str, Any]] = []
|
|
used_norm: set[str] = set()
|
|
for name, raw_value in raw_attributes.items():
|
|
if raw_value is None or str(raw_value).strip() == "":
|
|
continue
|
|
fold = str(name).casefold().rstrip(":")
|
|
norm_key = next((key for key, aliases in SPEC_ALIASES.items() if fold in aliases), None)
|
|
norm_value = normalized_attributes.get(norm_key) if norm_key else None
|
|
if norm_key:
|
|
used_norm.add(norm_key)
|
|
items.append(
|
|
{
|
|
"attribute_key": norm_key or fold.replace(" ", "_"),
|
|
"raw_name": str(name).rstrip(":"),
|
|
"raw_value": raw_value,
|
|
"normalized_value": norm_value,
|
|
}
|
|
)
|
|
for key, value in normalized_attributes.items():
|
|
if key in used_norm or value is None:
|
|
continue
|
|
items.append(
|
|
{
|
|
"attribute_key": key,
|
|
"raw_name": key,
|
|
"raw_value": None,
|
|
"normalized_value": value,
|
|
}
|
|
)
|
|
return items
|
|
|
|
|
|
def is_derived_duplicate(name: str, present: set[str]) -> bool:
|
|
fold = name.casefold()
|
|
aliases = SPEC_ALIASES.get(fold)
|
|
return bool(aliases and present.intersection(aliases))
|