75 lines
2.4 KiB
Python
75 lines
2.4 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from decimal import Decimal
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
from app.discovery.enums import DiffStatus, LifecycleStatus, RunStatus
|
||
|
|
|
||
|
|
|
||
|
|
RELEVANT_ATTR_IGNORE = {"title", "html", "presentation", "raw_title"}
|
||
|
|
|
||
|
|
|
||
|
|
def observations_changed(previous: dict[str, Any] | None, current: dict[str, Any] | None) -> bool:
|
||
|
|
if previous is None or current is None:
|
||
|
|
return previous != current
|
||
|
|
if _money(previous.get("best_price")) != _money(current.get("best_price")):
|
||
|
|
return True
|
||
|
|
if previous.get("availability") != current.get("availability"):
|
||
|
|
return True
|
||
|
|
if previous.get("seller_count") != current.get("seller_count"):
|
||
|
|
return True
|
||
|
|
prev_attrs = _technical(previous.get("normalized_attributes") or {})
|
||
|
|
curr_attrs = _technical(current.get("normalized_attributes") or {})
|
||
|
|
return prev_attrs != curr_attrs
|
||
|
|
|
||
|
|
|
||
|
|
def _money(value: Any) -> Decimal | None:
|
||
|
|
if value is None:
|
||
|
|
return None
|
||
|
|
return Decimal(str(value))
|
||
|
|
|
||
|
|
|
||
|
|
def _technical(attrs: dict[str, Any]) -> dict[str, Any]:
|
||
|
|
return {key: attrs[key] for key in sorted(attrs) if key not in RELEVANT_ATTR_IGNORE}
|
||
|
|
|
||
|
|
|
||
|
|
def diff_run(
|
||
|
|
*,
|
||
|
|
previous_ids: set[Any],
|
||
|
|
current_ids: set[Any],
|
||
|
|
previous_obs: dict[Any, dict[str, Any]],
|
||
|
|
current_obs: dict[Any, dict[str, Any]],
|
||
|
|
run_status: str,
|
||
|
|
) -> dict[Any, DiffStatus]:
|
||
|
|
result: dict[Any, DiffStatus] = {}
|
||
|
|
for ref_id in current_ids:
|
||
|
|
if ref_id not in previous_ids:
|
||
|
|
result[ref_id] = DiffStatus.NEW
|
||
|
|
elif observations_changed(previous_obs.get(ref_id), current_obs.get(ref_id)):
|
||
|
|
result[ref_id] = DiffStatus.CHANGED
|
||
|
|
else:
|
||
|
|
result[ref_id] = DiffStatus.UNCHANGED
|
||
|
|
if run_status == RunStatus.SUCCESS.value:
|
||
|
|
for ref_id in previous_ids - current_ids:
|
||
|
|
result[ref_id] = DiffStatus.MISSING
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
def next_lifecycle(
|
||
|
|
*,
|
||
|
|
current: str,
|
||
|
|
missing_in_success_run: bool,
|
||
|
|
consecutive_missing: int,
|
||
|
|
reappeared: bool,
|
||
|
|
) -> tuple[str, int]:
|
||
|
|
if current == LifecycleStatus.ARCHIVED.value:
|
||
|
|
return current, consecutive_missing
|
||
|
|
if reappeared:
|
||
|
|
return LifecycleStatus.ACTIVE.value, 0
|
||
|
|
if missing_in_success_run:
|
||
|
|
streak = consecutive_missing + 1
|
||
|
|
if streak >= 3:
|
||
|
|
return LifecycleStatus.DISAPPEARED.value, streak
|
||
|
|
return LifecycleStatus.ACTIVE.value, streak
|
||
|
|
return current, consecutive_missing
|