101 lines
3.5 KiB
Python
101 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
from uuid import UUID, uuid4
|
|
|
|
from app.clock import now_utc
|
|
from app.domain.attributes import select_attribute_snapshot
|
|
from app.domain.constraints import evaluate_constraint
|
|
from app.domain.enums import Eligibility, NormalizationStatus
|
|
from app.domain.models import AttributeStatement, Product
|
|
from app.scoring.engine import _product_eligibility
|
|
from app.scoring.mathutil import round_score
|
|
from app.scoring.product import score_product
|
|
|
|
|
|
def rank_candidates(
|
|
*,
|
|
profile,
|
|
scoring_model,
|
|
definitions,
|
|
candidates: list[dict[str, Any]],
|
|
shortlist_size: int,
|
|
as_of: datetime | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
scored: list[dict[str, Any]] = []
|
|
when = as_of or now_utc()
|
|
for candidate in candidates:
|
|
attrs = _statements(candidate, when)
|
|
snapshot = select_attribute_snapshot(
|
|
attrs,
|
|
as_of=when,
|
|
snapshot_cutoff=when,
|
|
)
|
|
constraint_results = tuple(
|
|
evaluate_constraint(
|
|
constraint,
|
|
snapshot.get(constraint.attribute),
|
|
definitions[constraint.attribute],
|
|
scoring_model.minimum_confidence,
|
|
)
|
|
for constraint in profile.constraints
|
|
if constraint.attribute in definitions
|
|
)
|
|
product = Product(
|
|
id=uuid4(),
|
|
category=profile.category,
|
|
brand=str(candidate.get("manufacturer") or "unknown"),
|
|
name=str(candidate.get("source_title") or candidate.get("name") or "candidate"),
|
|
model=candidate.get("model"),
|
|
created_at=when,
|
|
)
|
|
_gates, product_eligibility = _product_eligibility(profile, product, constraint_results)
|
|
product_score, coverage, _components = score_product(
|
|
profile.preferences,
|
|
snapshot,
|
|
constraint_results,
|
|
scoring_model,
|
|
)
|
|
scored.append(
|
|
{
|
|
**candidate,
|
|
"product_eligibility": product_eligibility.value,
|
|
"product_score": round_score(product_score),
|
|
"product_coverage": coverage,
|
|
"eligible": product_eligibility == Eligibility.ELIGIBLE,
|
|
}
|
|
)
|
|
scored.sort(key=lambda item: (not item["eligible"], -(item["product_score"] or Decimal("0"))))
|
|
for index, item in enumerate(scored):
|
|
item["candidate_status"] = "SHORTLISTED" if index < shortlist_size and item["eligible"] else "CANDIDATE"
|
|
item["rank"] = index + 1
|
|
return scored
|
|
|
|
|
|
def _statements(candidate: dict[str, Any], when: datetime) -> tuple[AttributeStatement, ...]:
|
|
raw = candidate.get("normalized_attributes") or {}
|
|
now = when
|
|
items = []
|
|
variant_id = candidate.get("variant_id") or uuid4()
|
|
for key, value in raw.items():
|
|
items.append(
|
|
AttributeStatement(
|
|
id=uuid4(),
|
|
variant_id=variant_id if isinstance(variant_id, UUID) else uuid4(),
|
|
attribute=key,
|
|
raw_value=value,
|
|
normalized_value=value,
|
|
unit=None,
|
|
normalization_status=NormalizationStatus.NORMALIZED,
|
|
normalizer_version="discovery-v1",
|
|
confidence=Decimal("1"),
|
|
source_id="geizhals",
|
|
source_reference="discovery",
|
|
observed_at=now,
|
|
ingested_at=now,
|
|
)
|
|
)
|
|
return tuple(items)
|