279 lines
11 KiB
Python
279 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.discovery.enums import FilterType, MappingStatus, TaxonomyStatus
|
|
from app.domain.serialize import jsonable
|
|
from app.errors import AppError
|
|
from app.persistence.discovery_models import (
|
|
CriterionDefinitionRecord,
|
|
CriterionMappingRecord,
|
|
ProductCategoryRecord,
|
|
SourceCategoryRecord,
|
|
SourceFilterDefinitionRecord,
|
|
SourceFilterOptionRecord,
|
|
SourceFilterRangeRecord,
|
|
)
|
|
from app.discovery.definitions import upsert_definition
|
|
from app.persistence.repositories import CatalogRepository
|
|
|
|
|
|
FILTER_OPERATORS = {
|
|
FilterType.BOOLEAN.value: ["=", "!="],
|
|
FilterType.ENUM.value: ["=", "!="],
|
|
FilterType.MULTI_ENUM.value: ["in", "not_in"],
|
|
FilterType.NUMERIC_RANGE.value: [">=", "<=", "between"],
|
|
FilterType.NUMERIC_THRESHOLD.value: [">=", ">", "<=", "<"],
|
|
FilterType.TEXT.value: ["=", "contains"],
|
|
}
|
|
|
|
|
|
class CatalogService:
|
|
def __init__(self, session: Session) -> None:
|
|
self.session = session
|
|
self.catalog = CatalogRepository(session)
|
|
|
|
def list_product_categories(self) -> list[dict[str, Any]]:
|
|
rows = self.session.scalars(select(ProductCategoryRecord).order_by(ProductCategoryRecord.name)).all()
|
|
return [{"id": row.id, "name": row.name, "status": row.status} for row in rows]
|
|
|
|
def get_product_category(self, category_id: str) -> ProductCategoryRecord:
|
|
row = self.session.get(ProductCategoryRecord, category_id)
|
|
if row is None:
|
|
raise AppError(404, "NOT_FOUND", "product category not found", {"id": category_id})
|
|
return row
|
|
|
|
def sources_for_category(self, product_category_id: str) -> list[dict[str, Any]]:
|
|
self.get_product_category(product_category_id)
|
|
rows = self.session.scalars(
|
|
select(SourceCategoryRecord)
|
|
.where(SourceCategoryRecord.product_category_id == product_category_id)
|
|
.order_by(SourceCategoryRecord.source_id)
|
|
).all()
|
|
sources = {item.id: item for item in self.catalog.list_sources()}
|
|
return [
|
|
{
|
|
"source": row.source_id,
|
|
"source_category_id": row.id,
|
|
"display_name": sources[row.source_id].name if row.source_id in sources else row.name,
|
|
"external_category_key": row.external_category_key,
|
|
"status": row.status,
|
|
"taxonomy_status": row.taxonomy_status,
|
|
}
|
|
for row in rows
|
|
]
|
|
|
|
def source_categories(self, source_id: str) -> list[dict[str, Any]]:
|
|
if self.catalog.get_source(source_id) is None:
|
|
raise AppError(404, "NOT_FOUND", "source not found", {"source_id": source_id})
|
|
rows = self.session.scalars(
|
|
select(SourceCategoryRecord)
|
|
.where(SourceCategoryRecord.source_id == source_id)
|
|
.order_by(SourceCategoryRecord.external_category_key)
|
|
).all()
|
|
return [
|
|
{
|
|
"id": row.id,
|
|
"source": row.source_id,
|
|
"source_category_id": row.id,
|
|
"external_category_key": row.external_category_key,
|
|
"name": row.name,
|
|
"product_category_id": row.product_category_id,
|
|
"status": row.status,
|
|
"taxonomy_status": row.taxonomy_status,
|
|
"source_url": row.source_url,
|
|
}
|
|
for row in rows
|
|
]
|
|
|
|
def available_criteria(self, product_category_id: str, source_id: str) -> dict[str, Any]:
|
|
self.get_product_category(product_category_id)
|
|
if self.catalog.get_source(source_id) is None:
|
|
raise AppError(404, "NOT_FOUND", "source not found", {"source_id": source_id})
|
|
source_category = self.session.scalars(
|
|
select(SourceCategoryRecord).where(
|
|
SourceCategoryRecord.source_id == source_id,
|
|
SourceCategoryRecord.product_category_id == product_category_id,
|
|
)
|
|
).first()
|
|
if source_category is not None and source_category.taxonomy_status == TaxonomyStatus.NOT_LEARNED.value:
|
|
from app.services.discovery_service import TaxonomyService
|
|
|
|
TaxonomyService(self.session).discover(source_id, source_category.external_category_key)
|
|
source_category = self.session.scalars(
|
|
select(SourceCategoryRecord).where(
|
|
SourceCategoryRecord.source_id == source_id,
|
|
SourceCategoryRecord.product_category_id == product_category_id,
|
|
)
|
|
).first()
|
|
criteria: list[dict[str, Any]] = []
|
|
if source_category is not None:
|
|
filters = self.session.scalars(
|
|
select(SourceFilterDefinitionRecord)
|
|
.where(SourceFilterDefinitionRecord.source_category_id == source_category.id)
|
|
.order_by(SourceFilterDefinitionRecord.label)
|
|
).all()
|
|
mappings = {
|
|
row.external_filter_key: row
|
|
for row in self.session.scalars(
|
|
select(CriterionMappingRecord).where(
|
|
CriterionMappingRecord.source_category_id == source_category.id
|
|
)
|
|
)
|
|
}
|
|
for filt in filters:
|
|
mapping = mappings.get(filt.external_filter_key)
|
|
definition = None if mapping is None else self.session.get(CriterionDefinitionRecord, mapping.criterion_definition_id)
|
|
if definition is None:
|
|
definition = upsert_definition(
|
|
self.session,
|
|
{
|
|
"key": f"{source_id}:{source_category.external_category_key}:{filt.external_filter_key}",
|
|
"origin": "SOURCE",
|
|
"source": source_id,
|
|
"product_category_id": product_category_id,
|
|
"label": filt.label,
|
|
"value_type": "number"
|
|
if filt.filter_type in {FilterType.NUMERIC_RANGE.value, FilterType.NUMERIC_THRESHOLD.value}
|
|
else "string",
|
|
"unit": filt.canonical_unit,
|
|
"category": product_category_id,
|
|
},
|
|
)
|
|
self.session.flush()
|
|
options = self.session.scalars(
|
|
select(SourceFilterOptionRecord).where(SourceFilterOptionRecord.filter_id == filt.id)
|
|
).all()
|
|
rng = self.session.get(SourceFilterRangeRecord, filt.id)
|
|
criteria.append(
|
|
self._criterion_out(
|
|
definition,
|
|
filt,
|
|
mapping,
|
|
options,
|
|
rng,
|
|
)
|
|
)
|
|
internals = self.session.scalars(
|
|
select(CriterionDefinitionRecord)
|
|
.where(
|
|
CriterionDefinitionRecord.origin == "INTERNAL",
|
|
(CriterionDefinitionRecord.product_category_id == product_category_id)
|
|
| (CriterionDefinitionRecord.category == product_category_id),
|
|
)
|
|
.order_by(CriterionDefinitionRecord.label)
|
|
).all()
|
|
seen = {item["criterion_definition_id"] for item in criteria}
|
|
for definition in internals:
|
|
if definition.id in seen:
|
|
continue
|
|
criteria.append(self._criterion_out(definition, None, None, [], None))
|
|
return jsonable({"product_category_id": product_category_id, "source": source_id, "criteria": criteria})
|
|
|
|
def list_templates(self) -> list[dict[str, Any]]:
|
|
rows, _total = self.catalog.list_profiles(200, 0)
|
|
return [
|
|
{
|
|
"id": row.id,
|
|
"version": row.version,
|
|
"name": row.definition.get("name") or row.id,
|
|
"product_category_id": row.definition.get("category"),
|
|
"market": row.definition.get("market"),
|
|
"currency": row.definition.get("currency"),
|
|
"description": row.definition.get("description"),
|
|
}
|
|
for row in rows
|
|
]
|
|
|
|
def _criterion_out(
|
|
self,
|
|
definition: CriterionDefinitionRecord,
|
|
filt: SourceFilterDefinitionRecord | None,
|
|
mapping: CriterionMappingRecord | None,
|
|
options: list[SourceFilterOptionRecord],
|
|
rng: SourceFilterRangeRecord | None,
|
|
) -> dict[str, Any]:
|
|
filter_type = None if filt is None else filt.filter_type
|
|
return {
|
|
"criterion_definition_id": definition.id,
|
|
"label": definition.label,
|
|
"description": definition.description,
|
|
"origin": definition.origin,
|
|
"source": definition.source,
|
|
"product_category_id": definition.product_category_id,
|
|
"filter_type": filter_type,
|
|
"unit": definition.unit or (None if filt is None else filt.canonical_unit),
|
|
"operators": FILTER_OPERATORS.get(filter_type or "", ["=", ">=", "<="]),
|
|
"options": [
|
|
{
|
|
"source_value": option.source_value,
|
|
"normalized_value": option.normalized_value,
|
|
"operator": option.operator,
|
|
"unit": option.unit,
|
|
}
|
|
for option in options
|
|
],
|
|
"range": None
|
|
if rng is None
|
|
else {"minimum": rng.minimum, "maximum": rng.maximum, "step": rng.step, "unit": rng.unit},
|
|
"mapping_status": None if mapping is None else mapping.status,
|
|
"supports_discovery": mapping is not None
|
|
and mapping.status == MappingStatus.VERIFIED.value
|
|
and filt is not None,
|
|
"scoring_attribute_key": definition.scoring_attribute_key,
|
|
}
|
|
|
|
|
|
def ensure_product_category(session: Session, category_id: str, name: str | None = None) -> ProductCategoryRecord:
|
|
row = session.get(ProductCategoryRecord, category_id)
|
|
if row is not None:
|
|
return row
|
|
row = ProductCategoryRecord(id=category_id, name=name or category_id.title(), status="ACTIVE")
|
|
session.add(row)
|
|
session.flush()
|
|
return row
|
|
|
|
|
|
def ensure_source_category(
|
|
session: Session,
|
|
*,
|
|
source_id: str,
|
|
external_category_key: str,
|
|
product_category_id: str,
|
|
name: str,
|
|
source_url: str,
|
|
) -> SourceCategoryRecord:
|
|
row = session.scalars(
|
|
select(SourceCategoryRecord).where(
|
|
SourceCategoryRecord.source_id == source_id,
|
|
SourceCategoryRecord.external_category_key == external_category_key,
|
|
)
|
|
).first()
|
|
if row is None:
|
|
row = SourceCategoryRecord(
|
|
id=uuid4(),
|
|
source_id=source_id,
|
|
external_category_key=external_category_key,
|
|
name=name,
|
|
source_url=source_url,
|
|
product_category_id=product_category_id,
|
|
canonical_category=product_category_id,
|
|
status="ACTIVE",
|
|
taxonomy_status=TaxonomyStatus.NOT_LEARNED.value,
|
|
last_taxonomy_scan_at=None,
|
|
last_verified_at=None,
|
|
)
|
|
session.add(row)
|
|
session.flush()
|
|
return row
|
|
if row.product_category_id is None:
|
|
row.product_category_id = product_category_id
|
|
if not row.source_url:
|
|
row.source_url = source_url
|
|
session.flush()
|
|
return row
|