264 lines
12 KiB
Python
264 lines
12 KiB
Python
|
from datetime import datetime
|
||
|
|
from decimal import Decimal
|
||
|
|
from typing import Any
|
||
|
|
from uuid import UUID, uuid4
|
||
|
|
|
||
|
|
from sqlalchemy.exc import IntegrityError
|
||
|
|
from sqlalchemy.orm import Session
|
||
|
|
|
||
|
|
from app.domain.enums import Availability, NormalizationStatus, OfferCondition, ValueType
|
||
|
|
from app.domain.models import AttributeDefinition, AttributeStatement, Offer, PriceObservation, Product, ProductVariant
|
||
|
|
from app.clock import now_utc
|
||
|
|
from app.domain.timeutil import require_utc
|
||
|
|
from app.domain.validation import DomainValidationError, parse_profile, validate_attribute_key, validate_currency
|
||
|
|
from app.errors import AppError
|
||
|
|
from app.normalization.manual import ManualNormalizer
|
||
|
|
from app.persistence import models as orm
|
||
|
|
from app.persistence.mapping import to_observation
|
||
|
|
from app.persistence.repositories import CatalogRepository
|
||
|
|
|
||
|
|
|
||
|
|
def _now() -> datetime:
|
||
|
|
return now_utc()
|
||
|
|
|
||
|
|
|
||
|
|
def _uuid(value: UUID | None) -> UUID:
|
||
|
|
return value or uuid4()
|
||
|
|
|
||
|
|
|
||
|
|
class IngestionService:
|
||
|
|
def __init__(self, session: Session) -> None:
|
||
|
|
self.session = session
|
||
|
|
self.repo = CatalogRepository(session)
|
||
|
|
self.normalizer = ManualNormalizer()
|
||
|
|
|
||
|
|
def create_definition(self, payload: dict[str, Any]) -> AttributeDefinition:
|
||
|
|
key = str(payload["key"])
|
||
|
|
validate_attribute_key(key)
|
||
|
|
row = orm.AttributeDefinitionRecord(
|
||
|
|
key=key,
|
||
|
|
value_type=ValueType(payload["value_type"]).value,
|
||
|
|
canonical_unit=payload.get("canonical_unit"),
|
||
|
|
minimum=payload.get("minimum"),
|
||
|
|
maximum=payload.get("maximum"),
|
||
|
|
description=payload.get("description"),
|
||
|
|
)
|
||
|
|
if self.repo.get_definition(key) is not None:
|
||
|
|
raise AppError(409, "CONFLICT", "attribute definition already exists", {"key": key})
|
||
|
|
return self.repo.add_definition(row)
|
||
|
|
|
||
|
|
def create_profile(self, payload: dict[str, Any]) -> orm.ProductProfileRecord:
|
||
|
|
try:
|
||
|
|
model_versions = self.repo.scoring_versions()
|
||
|
|
scoring_version = payload.get("scoring_version")
|
||
|
|
model = self.repo.get_scoring_model(scoring_version) if scoring_version else None
|
||
|
|
if model is None:
|
||
|
|
raise AppError(422, "UNKNOWN_SCORING_VERSION", "scoring version does not exist")
|
||
|
|
profile = parse_profile(
|
||
|
|
payload,
|
||
|
|
attribute_definitions=self.repo.all_definitions(),
|
||
|
|
scoring_versions=model_versions,
|
||
|
|
supported_strategies=set(model.supported_preference_strategies),
|
||
|
|
)
|
||
|
|
except DomainValidationError as exc:
|
||
|
|
raise AppError(422, "PROFILE_INVALID", str(exc), exc.details) from exc
|
||
|
|
except AppError:
|
||
|
|
raise
|
||
|
|
row = orm.ProductProfileRecord(
|
||
|
|
id=profile.id,
|
||
|
|
version=profile.version,
|
||
|
|
name=profile.name,
|
||
|
|
category=profile.category,
|
||
|
|
scoring_version=profile.scoring_version,
|
||
|
|
definition=profile.definition,
|
||
|
|
definition_hash=profile.definition_hash,
|
||
|
|
created_at=_now(),
|
||
|
|
)
|
||
|
|
if self.repo.get_profile(profile.id, profile.version) is not None:
|
||
|
|
raise AppError(409, "CONFLICT", "profile version already exists", {"id": profile.id, "version": profile.version})
|
||
|
|
return self.repo.add_profile(row)
|
||
|
|
|
||
|
|
def create_product(self, payload: dict[str, Any]) -> Product:
|
||
|
|
product_id = _uuid(payload.get("id"))
|
||
|
|
if self.repo.get_product(product_id) is not None:
|
||
|
|
raise AppError(409, "CONFLICT", "product already exists", {"id": str(product_id)})
|
||
|
|
row = orm.ProductRecord(
|
||
|
|
id=product_id,
|
||
|
|
category=str(payload["category"]),
|
||
|
|
brand=str(payload["brand"]),
|
||
|
|
name=str(payload["name"]),
|
||
|
|
model=payload.get("model"),
|
||
|
|
created_at=_now(),
|
||
|
|
)
|
||
|
|
return self.repo.add_product(row)
|
||
|
|
|
||
|
|
def create_variant(self, product_id: UUID, payload: dict[str, Any]) -> ProductVariant:
|
||
|
|
if self.repo.get_product(product_id) is None:
|
||
|
|
raise AppError(404, "NOT_FOUND", "product not found", {"id": str(product_id)})
|
||
|
|
variant_id = _uuid(payload.get("id"))
|
||
|
|
if self.repo.get_variant(variant_id) is not None:
|
||
|
|
raise AppError(409, "CONFLICT", "variant already exists", {"id": str(variant_id)})
|
||
|
|
row = orm.ProductVariantRecord(
|
||
|
|
id=variant_id,
|
||
|
|
product_id=product_id,
|
||
|
|
name=str(payload["name"]),
|
||
|
|
manufacturer_sku=payload.get("manufacturer_sku"),
|
||
|
|
identifiers=payload.get("identifiers") or {},
|
||
|
|
created_at=_now(),
|
||
|
|
)
|
||
|
|
return self.repo.add_variant(row)
|
||
|
|
|
||
|
|
def add_attribute(self, variant_id: UUID, payload: dict[str, Any]) -> AttributeStatement:
|
||
|
|
variant = self.repo.get_variant(variant_id)
|
||
|
|
if variant is None:
|
||
|
|
raise AppError(404, "NOT_FOUND", "variant not found", {"id": str(variant_id)})
|
||
|
|
definition = self.repo.get_definition(payload["attribute"])
|
||
|
|
if definition is None:
|
||
|
|
raise AppError(422, "UNKNOWN_ATTRIBUTE", "attribute definition does not exist")
|
||
|
|
if self.repo.get_source(payload["source_id"]) is None:
|
||
|
|
raise AppError(422, "UNKNOWN_SOURCE", "source does not exist")
|
||
|
|
observed_at = require_utc(payload["observed_at"], "observed_at")
|
||
|
|
ingested_at = _now()
|
||
|
|
if observed_at > ingested_at:
|
||
|
|
raise AppError(422, "FUTURE_OBSERVATION", "observed_at must not be after ingestion")
|
||
|
|
confidence = Decimal(str(payload["confidence"]))
|
||
|
|
if not Decimal("0") <= confidence <= Decimal("1"):
|
||
|
|
raise AppError(422, "CONFIDENCE_RANGE", "confidence must be between 0 and 1")
|
||
|
|
statement = AttributeStatement(
|
||
|
|
id=_uuid(payload.get("id")),
|
||
|
|
variant_id=variant_id,
|
||
|
|
attribute=definition.key,
|
||
|
|
raw_value=payload["raw_value"],
|
||
|
|
normalized_value=payload.get("normalized_value"),
|
||
|
|
unit=payload.get("unit"),
|
||
|
|
normalization_status=NormalizationStatus(payload["normalization_status"]),
|
||
|
|
normalizer_version=payload.get("normalizer_version") or self.normalizer.version,
|
||
|
|
confidence=confidence,
|
||
|
|
source_id=payload["source_id"],
|
||
|
|
source_reference=payload["source_reference"],
|
||
|
|
observed_at=observed_at,
|
||
|
|
ingested_at=ingested_at,
|
||
|
|
normalization_error=payload.get("normalization_error"),
|
||
|
|
)
|
||
|
|
statement = self.normalizer.validate(definition, statement)
|
||
|
|
if self.repo.get_attribute(statement.id) is not None:
|
||
|
|
raise AppError(409, "CONFLICT", "attribute statement already exists", {"id": str(statement.id)})
|
||
|
|
row = orm.VariantAttributeRecord(
|
||
|
|
id=statement.id,
|
||
|
|
variant_id=variant_id,
|
||
|
|
attribute_key=statement.attribute,
|
||
|
|
source_id=statement.source_id,
|
||
|
|
raw_value=statement.raw_value,
|
||
|
|
normalized_value=statement.normalized_value,
|
||
|
|
unit=statement.unit,
|
||
|
|
normalization_status=statement.normalization_status.value,
|
||
|
|
normalizer_version=statement.normalizer_version,
|
||
|
|
confidence=statement.confidence,
|
||
|
|
source_reference=statement.source_reference,
|
||
|
|
observed_at=statement.observed_at,
|
||
|
|
ingested_at=statement.ingested_at,
|
||
|
|
normalization_error=statement.normalization_error,
|
||
|
|
)
|
||
|
|
return self.repo.add_attribute(row)
|
||
|
|
|
||
|
|
def create_offer(self, variant_id: UUID, payload: dict[str, Any]) -> Offer:
|
||
|
|
if self.repo.get_variant(variant_id) is None:
|
||
|
|
raise AppError(404, "NOT_FOUND", "variant not found", {"id": str(variant_id)})
|
||
|
|
if self.repo.get_source(payload["source_id"]) is None:
|
||
|
|
raise AppError(422, "UNKNOWN_SOURCE", "source does not exist")
|
||
|
|
offer_id = _uuid(payload.get("id"))
|
||
|
|
if self.repo.get_offer(offer_id) is not None:
|
||
|
|
raise AppError(409, "CONFLICT", "offer already exists", {"id": str(offer_id)})
|
||
|
|
if self.repo.get_offer_by_external(payload["source_id"], str(payload["external_id"])) is not None:
|
||
|
|
raise AppError(409, "CONFLICT", "offer source/external_id already exists")
|
||
|
|
row = orm.OfferRecord(
|
||
|
|
id=offer_id,
|
||
|
|
variant_id=variant_id,
|
||
|
|
source_id=payload["source_id"],
|
||
|
|
external_id=str(payload["external_id"]),
|
||
|
|
seller=str(payload["seller"]),
|
||
|
|
url=payload.get("url"),
|
||
|
|
market=str(payload["market"]),
|
||
|
|
currency=validate_currency(str(payload["currency"])),
|
||
|
|
condition=OfferCondition(payload["condition"]).value,
|
||
|
|
created_at=_now(),
|
||
|
|
)
|
||
|
|
return self.repo.add_offer(row)
|
||
|
|
|
||
|
|
def add_price(self, offer_id: UUID, payload: dict[str, Any]) -> tuple[PriceObservation, bool]:
|
||
|
|
offer = self.repo.get_offer(offer_id)
|
||
|
|
if offer is None:
|
||
|
|
raise AppError(404, "NOT_FOUND", "offer not found", {"id": str(offer_id)})
|
||
|
|
if payload["source_id"] != offer.source_id:
|
||
|
|
raise AppError(422, "SOURCE_MISMATCH", "source_id must match the offer")
|
||
|
|
currency = validate_currency(str(payload["currency"]))
|
||
|
|
if currency != offer.currency:
|
||
|
|
raise AppError(422, "CURRENCY_MISMATCH", "currency must match the offer")
|
||
|
|
observed_at = require_utc(payload["observed_at"], "observed_at")
|
||
|
|
ingested_at = _now()
|
||
|
|
if observed_at > ingested_at:
|
||
|
|
raise AppError(422, "FUTURE_OBSERVATION", "observed_at must not be after ingestion")
|
||
|
|
item_price = Decimal(str(payload["item_price"]))
|
||
|
|
if item_price <= 0:
|
||
|
|
raise AppError(422, "ITEM_PRICE", "item_price must be > 0")
|
||
|
|
shipping = payload.get("shipping_price")
|
||
|
|
shipping_price = None if shipping is None else Decimal(str(shipping))
|
||
|
|
if shipping_price is not None and shipping_price < 0:
|
||
|
|
raise AppError(422, "SHIPPING_PRICE", "shipping_price must be >= 0")
|
||
|
|
confidence = Decimal(str(payload["confidence"]))
|
||
|
|
if not Decimal("0") <= confidence <= Decimal("1"):
|
||
|
|
raise AppError(422, "CONFIDENCE_RANGE", "confidence must be between 0 and 1")
|
||
|
|
external_id = str(payload["external_observation_id"])
|
||
|
|
existing = self.repo.get_observation_by_external(offer_id, external_id)
|
||
|
|
incoming = {
|
||
|
|
"source_id": payload["source_id"],
|
||
|
|
"observed_at": observed_at,
|
||
|
|
"item_price": item_price,
|
||
|
|
"shipping_price": shipping_price,
|
||
|
|
"currency": currency,
|
||
|
|
"availability": Availability(payload["availability"]).value,
|
||
|
|
"raw_payload": payload["raw_payload"],
|
||
|
|
"confidence": confidence,
|
||
|
|
}
|
||
|
|
if existing is not None:
|
||
|
|
if _observation_content(existing) != incoming:
|
||
|
|
raise AppError(409, "CONFLICT", "price observation content differs")
|
||
|
|
return to_observation(existing), False
|
||
|
|
row = orm.PriceObservationRecord(
|
||
|
|
id=_uuid(payload.get("id")),
|
||
|
|
offer_id=offer_id,
|
||
|
|
source_id=payload["source_id"],
|
||
|
|
external_observation_id=external_id,
|
||
|
|
observed_at=observed_at,
|
||
|
|
ingested_at=ingested_at,
|
||
|
|
item_price=item_price,
|
||
|
|
shipping_price=shipping_price,
|
||
|
|
currency=currency,
|
||
|
|
availability=incoming["availability"],
|
||
|
|
raw_payload=payload["raw_payload"],
|
||
|
|
confidence=confidence,
|
||
|
|
)
|
||
|
|
try:
|
||
|
|
return self.repo.add_observation(row), True
|
||
|
|
except IntegrityError:
|
||
|
|
self.session.rollback()
|
||
|
|
existing = self.repo.get_observation_by_external(offer_id, external_id)
|
||
|
|
if existing is None:
|
||
|
|
raise
|
||
|
|
if _observation_content(existing) != incoming:
|
||
|
|
raise AppError(409, "CONFLICT", "price observation content differs")
|
||
|
|
return to_observation(existing), False
|
||
|
|
|
||
|
|
|
||
|
|
def _observation_content(row: orm.PriceObservationRecord) -> dict[str, Any]:
|
||
|
|
return {
|
||
|
|
"source_id": row.source_id,
|
||
|
|
"observed_at": row.observed_at,
|
||
|
|
"item_price": Decimal(str(row.item_price)),
|
||
|
|
"shipping_price": None if row.shipping_price is None else Decimal(str(row.shipping_price)),
|
||
|
|
"currency": row.currency,
|
||
|
|
"availability": row.availability,
|
||
|
|
"raw_payload": row.raw_payload,
|
||
|
|
"confidence": Decimal(str(row.confidence)),
|
||
|
|
}
|