347 lines
16 KiB
Python
347 lines
16 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from decimal import Decimal
|
||
|
|
from typing import Any
|
||
|
|
from uuid import UUID, uuid4
|
||
|
|
|
||
|
|
from sqlalchemy import func, select
|
||
|
|
from sqlalchemy.orm import Session
|
||
|
|
|
||
|
|
from app.clock import now_utc
|
||
|
|
from app.discovery.criteria import criterion_payload, parse_project_criteria
|
||
|
|
from app.discovery.definitions import get_definition, get_definition_by_key, upsert_definition
|
||
|
|
from app.discovery.enums import CandidateStatus, DiffStatus, ProjectStatus, RankingStatus, SearchStatus
|
||
|
|
from app.domain.serialize import jsonable
|
||
|
|
from app.domain.validation import DomainValidationError
|
||
|
|
from app.errors import AppError
|
||
|
|
from app.persistence.discovery_models import (
|
||
|
|
BuyingProjectCriteriaRecord,
|
||
|
|
BuyingProjectRecord,
|
||
|
|
DiscoverySearchRecord,
|
||
|
|
ProductCategoryRecord,
|
||
|
|
ProjectCandidateRecord,
|
||
|
|
ProjectCriterionRecord,
|
||
|
|
SearchRunItemRecord,
|
||
|
|
SearchRunRecord,
|
||
|
|
)
|
||
|
|
from app.persistence.repositories import CatalogRepository
|
||
|
|
from app.services.catalog_service import CatalogService, ensure_product_category
|
||
|
|
|
||
|
|
|
||
|
|
INACTIVE_SEARCH = {SearchStatus.BLOCKED.value, SearchStatus.DISABLED.value, SearchStatus.PAUSED.value}
|
||
|
|
|
||
|
|
|
||
|
|
def _uuid(value: UUID | None) -> UUID:
|
||
|
|
return value or uuid4()
|
||
|
|
|
||
|
|
|
||
|
|
class ProjectService:
|
||
|
|
def __init__(self, session: Session) -> None:
|
||
|
|
self.session = session
|
||
|
|
self.catalog = CatalogRepository(session)
|
||
|
|
self.categories = CatalogService(session)
|
||
|
|
|
||
|
|
def create(self, payload: dict[str, Any]) -> dict[str, Any]:
|
||
|
|
template_id = payload.get("template_id") or payload.get("profile_id")
|
||
|
|
template_version = payload.get("template_version") or payload.get("profile_version")
|
||
|
|
profile = None
|
||
|
|
if template_id is not None:
|
||
|
|
if template_version is None:
|
||
|
|
raise AppError(422, "VALIDATION_ERROR", "template_version is required when template_id is set")
|
||
|
|
profile = self.catalog.get_profile(str(template_id), int(template_version))
|
||
|
|
if profile is None:
|
||
|
|
raise AppError(404, "NOT_FOUND", "template version not found")
|
||
|
|
product_category_id = payload.get("product_category_id") or (None if profile is None else profile.category)
|
||
|
|
if not product_category_id:
|
||
|
|
raise AppError(422, "VALIDATION_ERROR", "product_category_id is required")
|
||
|
|
if self.session.get(ProductCategoryRecord, product_category_id) is None:
|
||
|
|
ensure_product_category(self.session, str(product_category_id))
|
||
|
|
market = payload.get("market") or (None if profile is None else profile.market)
|
||
|
|
currency = payload.get("currency") or (None if profile is None else profile.currency)
|
||
|
|
if not market or not currency:
|
||
|
|
raise AppError(422, "VALIDATION_ERROR", "market and currency are required")
|
||
|
|
target_budget = payload.get("target_budget")
|
||
|
|
if target_budget is None and profile is not None and getattr(profile, "budget", None) is not None:
|
||
|
|
target_budget = getattr(profile.budget, "target", None)
|
||
|
|
criteria_payload = payload.get("criteria")
|
||
|
|
if not criteria_payload:
|
||
|
|
raise AppError(422, "CRITERIA_REQUIRED", "criteria are required")
|
||
|
|
try:
|
||
|
|
parsed = parse_project_criteria(criteria_payload)
|
||
|
|
except DomainValidationError as exc:
|
||
|
|
raise AppError(422, "CRITERIA_INVALID", str(exc), exc.details) from exc
|
||
|
|
shortlist_size = int(payload.get("shortlist_size") or 5)
|
||
|
|
if shortlist_size < 1:
|
||
|
|
raise AppError(422, "VALIDATION_ERROR", "shortlist_size must be > 0")
|
||
|
|
now = now_utc()
|
||
|
|
project = BuyingProjectRecord(
|
||
|
|
id=_uuid(payload.get("id")),
|
||
|
|
name=str(payload["name"]),
|
||
|
|
status=ProjectStatus.ACTIVE.value,
|
||
|
|
workspace_id=payload.get("workspace_id"),
|
||
|
|
product_category_id=str(product_category_id),
|
||
|
|
market=str(market),
|
||
|
|
currency=str(currency),
|
||
|
|
target_budget=None if target_budget is None else Decimal(str(target_budget)),
|
||
|
|
template_id=None if profile is None else profile.id,
|
||
|
|
template_version=None if profile is None else profile.version,
|
||
|
|
shortlist_size=shortlist_size,
|
||
|
|
criteria_version=1,
|
||
|
|
created_at=now,
|
||
|
|
updated_at=now,
|
||
|
|
archived_at=None,
|
||
|
|
)
|
||
|
|
self.session.add(project)
|
||
|
|
self.session.flush()
|
||
|
|
self._store_criteria(project.id, 1, parsed, str(product_category_id), now)
|
||
|
|
return self.get(project.id)
|
||
|
|
|
||
|
|
def replace_criteria(self, project_id: UUID, payload: dict[str, Any]) -> dict[str, Any]:
|
||
|
|
project = self._require_project(project_id)
|
||
|
|
try:
|
||
|
|
parsed = parse_project_criteria(payload)
|
||
|
|
except DomainValidationError as exc:
|
||
|
|
raise AppError(422, "CRITERIA_INVALID", str(exc), exc.details) from exc
|
||
|
|
current = self.latest_criteria(project_id)
|
||
|
|
version = (current.version if current is not None else 0) + 1
|
||
|
|
now = now_utc()
|
||
|
|
self._store_criteria(project.id, version, parsed, project.product_category_id, now)
|
||
|
|
project.criteria_version = version
|
||
|
|
project.updated_at = now
|
||
|
|
for search in self.session.scalars(
|
||
|
|
select(DiscoverySearchRecord).where(DiscoverySearchRecord.buying_project_id == project.id)
|
||
|
|
):
|
||
|
|
if search.status not in INACTIVE_SEARCH:
|
||
|
|
search.status = SearchStatus.OUT_OF_SYNC.value
|
||
|
|
search.updated_at = now
|
||
|
|
self.session.flush()
|
||
|
|
return self.get(project.id)
|
||
|
|
|
||
|
|
def add_criteria(self, project_id: UUID, payload: dict[str, Any]) -> dict[str, Any]:
|
||
|
|
return self.replace_criteria(project_id, payload)
|
||
|
|
|
||
|
|
def patch(self, project_id: UUID, payload: dict[str, Any]) -> dict[str, Any]:
|
||
|
|
project = self._require_project(project_id)
|
||
|
|
allowed = {"name", "target_budget", "shortlist_size", "status"}
|
||
|
|
unknown = set(payload) - allowed
|
||
|
|
if unknown:
|
||
|
|
raise AppError(422, "VALIDATION_ERROR", "field cannot be patched", {"fields": sorted(unknown)})
|
||
|
|
if "name" in payload:
|
||
|
|
project.name = str(payload["name"])
|
||
|
|
if "target_budget" in payload:
|
||
|
|
project.target_budget = None if payload["target_budget"] is None else Decimal(str(payload["target_budget"]))
|
||
|
|
if "shortlist_size" in payload:
|
||
|
|
size = int(payload["shortlist_size"])
|
||
|
|
if size < 1:
|
||
|
|
raise AppError(422, "VALIDATION_ERROR", "shortlist_size must be > 0")
|
||
|
|
project.shortlist_size = size
|
||
|
|
if "status" in payload:
|
||
|
|
status = str(payload["status"])
|
||
|
|
if status not in {item.value for item in ProjectStatus}:
|
||
|
|
raise AppError(422, "VALIDATION_ERROR", "invalid project status")
|
||
|
|
project.status = status
|
||
|
|
if status == ProjectStatus.ARCHIVED.value:
|
||
|
|
project.archived_at = now_utc()
|
||
|
|
project.updated_at = now_utc()
|
||
|
|
self.session.flush()
|
||
|
|
return self.get(project.id)
|
||
|
|
|
||
|
|
def get(self, project_id: UUID) -> dict[str, Any]:
|
||
|
|
project = self._require_project(project_id)
|
||
|
|
criteria = self.latest_criteria(project_id)
|
||
|
|
return self._out(project, criteria)
|
||
|
|
|
||
|
|
def get_criteria(self, project_id: UUID) -> dict[str, Any]:
|
||
|
|
project = self._require_project(project_id)
|
||
|
|
criteria = self.latest_criteria(project_id)
|
||
|
|
items = [] if criteria is None else self.criteria_items(project.id, criteria.version)
|
||
|
|
return jsonable(
|
||
|
|
{
|
||
|
|
"project_id": project.id,
|
||
|
|
"criteria_version": None if criteria is None else criteria.version,
|
||
|
|
"items": items,
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
def list(self, limit: int, offset: int) -> tuple[list[dict[str, Any]], int]:
|
||
|
|
total = self.session.scalar(select(func.count()).select_from(BuyingProjectRecord)) or 0
|
||
|
|
rows = self.session.scalars(
|
||
|
|
select(BuyingProjectRecord).order_by(BuyingProjectRecord.created_at).limit(limit).offset(offset)
|
||
|
|
).all()
|
||
|
|
return [self._summary(row) for row in rows], total
|
||
|
|
|
||
|
|
def latest_criteria(self, project_id: UUID) -> BuyingProjectCriteriaRecord | None:
|
||
|
|
return self.session.scalars(
|
||
|
|
select(BuyingProjectCriteriaRecord)
|
||
|
|
.where(BuyingProjectCriteriaRecord.project_id == project_id)
|
||
|
|
.order_by(BuyingProjectCriteriaRecord.version.desc())
|
||
|
|
).first()
|
||
|
|
|
||
|
|
def criteria_version(self, project_id: UUID, version: int) -> BuyingProjectCriteriaRecord | None:
|
||
|
|
return self.session.get(BuyingProjectCriteriaRecord, (project_id, version))
|
||
|
|
|
||
|
|
def criteria_items(self, project_id: UUID, version: int) -> list[dict[str, Any]]:
|
||
|
|
rows = self.session.scalars(
|
||
|
|
select(ProjectCriterionRecord)
|
||
|
|
.where(ProjectCriterionRecord.project_id == project_id, ProjectCriterionRecord.version == version)
|
||
|
|
.order_by(ProjectCriterionRecord.item_id)
|
||
|
|
).all()
|
||
|
|
return [criterion_payload(row) for row in rows]
|
||
|
|
|
||
|
|
def available_criteria(self, project_id: UUID) -> dict[str, Any]:
|
||
|
|
project = self._require_project(project_id)
|
||
|
|
sources = self.categories.sources_for_category(project.product_category_id)
|
||
|
|
source = sources[0]["source"] if sources else "geizhals"
|
||
|
|
return self.categories.available_criteria(project.product_category_id, source)
|
||
|
|
|
||
|
|
def _store_criteria(
|
||
|
|
self,
|
||
|
|
project_id: UUID,
|
||
|
|
version: int,
|
||
|
|
parsed: dict[str, Any],
|
||
|
|
category: str | None,
|
||
|
|
created_at,
|
||
|
|
) -> None:
|
||
|
|
self.session.add(
|
||
|
|
BuyingProjectCriteriaRecord(
|
||
|
|
project_id=project_id,
|
||
|
|
version=version,
|
||
|
|
definition_hash=parsed["definition_hash"],
|
||
|
|
created_at=created_at,
|
||
|
|
)
|
||
|
|
)
|
||
|
|
self.session.flush()
|
||
|
|
for item in parsed["items"]:
|
||
|
|
definition = self._resolve_definition(item, category)
|
||
|
|
self.session.add(
|
||
|
|
ProjectCriterionRecord(
|
||
|
|
id=uuid4(),
|
||
|
|
project_id=project_id,
|
||
|
|
version=version,
|
||
|
|
item_id=item["id"],
|
||
|
|
criterion_definition_id=definition.id,
|
||
|
|
criterion_key=definition.key,
|
||
|
|
operator=item["operator"],
|
||
|
|
value=item["value"],
|
||
|
|
enabled_for_discovery=item["enabled_for_discovery"],
|
||
|
|
created_at=created_at,
|
||
|
|
)
|
||
|
|
)
|
||
|
|
self.session.flush()
|
||
|
|
|
||
|
|
def _resolve_definition(self, item: dict[str, Any], category: str | None):
|
||
|
|
if item.get("criterion_definition_id") is not None:
|
||
|
|
definition = get_definition(self.session, item["criterion_definition_id"])
|
||
|
|
if definition is None:
|
||
|
|
raise AppError(422, "UNKNOWN_CRITERION", "criterion definition not found")
|
||
|
|
return definition
|
||
|
|
existing = get_definition_by_key(self.session, str(item["criterion"]))
|
||
|
|
if existing is not None:
|
||
|
|
return existing
|
||
|
|
if not item.get("value_type"):
|
||
|
|
raise AppError(
|
||
|
|
422,
|
||
|
|
"UNKNOWN_CRITERION",
|
||
|
|
"criterion definition missing; provide value_type or create it first",
|
||
|
|
{"criterion": item["criterion"]},
|
||
|
|
)
|
||
|
|
return upsert_definition(
|
||
|
|
self.session,
|
||
|
|
{
|
||
|
|
"key": item["criterion"],
|
||
|
|
"origin": "INTERNAL",
|
||
|
|
"product_category_id": item.get("category") or category,
|
||
|
|
"label": item["criterion"],
|
||
|
|
"value_type": item["value_type"],
|
||
|
|
"unit": item.get("unit"),
|
||
|
|
"semantic": item.get("semantic"),
|
||
|
|
"category": item.get("category") or category,
|
||
|
|
"scoring_attribute_key": item["criterion"],
|
||
|
|
},
|
||
|
|
)
|
||
|
|
|
||
|
|
def _require_project(self, project_id: UUID) -> BuyingProjectRecord:
|
||
|
|
project = self.session.get(BuyingProjectRecord, project_id)
|
||
|
|
if project is None:
|
||
|
|
raise AppError(404, "NOT_FOUND", "buying project not found")
|
||
|
|
return project
|
||
|
|
|
||
|
|
def _out(self, project: BuyingProjectRecord, criteria: BuyingProjectCriteriaRecord | None) -> dict[str, Any]:
|
||
|
|
items = [] if criteria is None else self.criteria_items(project.id, criteria.version)
|
||
|
|
return jsonable(
|
||
|
|
{
|
||
|
|
"id": project.id,
|
||
|
|
"name": project.name,
|
||
|
|
"status": project.status,
|
||
|
|
"workspace_id": project.workspace_id,
|
||
|
|
"product_category_id": project.product_category_id,
|
||
|
|
"market": project.market,
|
||
|
|
"currency": project.currency,
|
||
|
|
"target_budget": project.target_budget,
|
||
|
|
"template_id": project.template_id,
|
||
|
|
"template_version": project.template_version,
|
||
|
|
"profile_id": project.template_id,
|
||
|
|
"profile_version": project.template_version,
|
||
|
|
"shortlist_size": project.shortlist_size,
|
||
|
|
"created_at": project.created_at,
|
||
|
|
"updated_at": project.updated_at,
|
||
|
|
"archived_at": project.archived_at,
|
||
|
|
"criteria_version": project.criteria_version if criteria is None else criteria.version,
|
||
|
|
"criteria": None if criteria is None else {"items": items},
|
||
|
|
"criteria_hash": None if criteria is None else criteria.definition_hash,
|
||
|
|
"ranking_status": RankingStatus.NOT_CONFIGURED.value
|
||
|
|
if project.template_id is None
|
||
|
|
else RankingStatus.CONFIGURED.value,
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
def _summary(self, project: BuyingProjectRecord) -> dict[str, Any]:
|
||
|
|
category = self.session.get(ProductCategoryRecord, project.product_category_id)
|
||
|
|
searches = list(
|
||
|
|
self.session.scalars(
|
||
|
|
select(DiscoverySearchRecord).where(DiscoverySearchRecord.buying_project_id == project.id)
|
||
|
|
)
|
||
|
|
)
|
||
|
|
active = next((row for row in searches if row.status not in INACTIVE_SEARCH), searches[0] if searches else None)
|
||
|
|
last_run = None
|
||
|
|
if active is not None:
|
||
|
|
last_run = self.session.scalars(
|
||
|
|
select(SearchRunRecord)
|
||
|
|
.where(SearchRunRecord.discovery_search_id == active.id)
|
||
|
|
.order_by(SearchRunRecord.started_at.desc())
|
||
|
|
).first()
|
||
|
|
candidates = list(
|
||
|
|
self.session.scalars(
|
||
|
|
select(ProjectCandidateRecord).where(ProjectCandidateRecord.buying_project_id == project.id)
|
||
|
|
)
|
||
|
|
)
|
||
|
|
counts = {"NEW": 0, "CHANGED": 0, "MISSING": 0}
|
||
|
|
if last_run is not None:
|
||
|
|
for item in self.session.scalars(
|
||
|
|
select(SearchRunItemRecord).where(SearchRunItemRecord.search_run_id == last_run.id)
|
||
|
|
):
|
||
|
|
if item.diff_status in counts:
|
||
|
|
counts[item.diff_status] += 1
|
||
|
|
return jsonable(
|
||
|
|
{
|
||
|
|
"id": project.id,
|
||
|
|
"name": project.name,
|
||
|
|
"product_category": None
|
||
|
|
if category is None
|
||
|
|
else {"id": category.id, "name": category.name, "status": category.status},
|
||
|
|
"status": project.status,
|
||
|
|
"last_successful_run_at": None if active is None else active.last_successful_run_at,
|
||
|
|
"next_run_at": None if active is None else active.next_run_at,
|
||
|
|
"discovery_status": None if active is None else active.status,
|
||
|
|
"candidate_count": len(candidates),
|
||
|
|
"shortlist_count": sum(1 for row in candidates if row.status == CandidateStatus.SHORTLISTED.value),
|
||
|
|
"new_count": last_run.new_count if last_run is not None else counts[DiffStatus.NEW.value],
|
||
|
|
"changed_count": last_run.changed_count if last_run is not None else counts[DiffStatus.CHANGED.value],
|
||
|
|
"missing_count": last_run.missing_count if last_run is not None else counts[DiffStatus.MISSING.value],
|
||
|
|
"current_search_result_count": None if last_run is None else last_run.result_count,
|
||
|
|
"ranking_status": RankingStatus.NOT_CONFIGURED.value
|
||
|
|
if project.template_id is None
|
||
|
|
else RankingStatus.CONFIGURED.value,
|
||
|
|
}
|
||
|
|
)
|