|
|
@@ -0,0 +1,737 @@
|
|
|
+"""Governed, explainable, and reversible cross-source device resolution."""
|
|
|
+
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import unicodedata
|
|
|
+from collections.abc import Callable
|
|
|
+from dataclasses import dataclass, replace
|
|
|
+from datetime import datetime
|
|
|
+from itertools import combinations
|
|
|
+from typing import Any
|
|
|
+
|
|
|
+from app.core.common.identifiers import new_governance_uid
|
|
|
+from app.core.common.timezone_utils import now_china
|
|
|
+from app.core.data_research.device_assets import ASSET_TYPES, DeviceAssetDetail
|
|
|
+from app.core.data_research.errors import (
|
|
|
+ DeviceEntityConflict,
|
|
|
+ DeviceEntityForbidden,
|
|
|
+ DeviceEntityInvalid,
|
|
|
+ DeviceEntityNotFound,
|
|
|
+)
|
|
|
+
|
|
|
+MAX_MATCH_ASSETS = 500
|
|
|
+MAX_NEW_CANDIDATES = 2_000
|
|
|
+MAX_PAGE_SIZE = 100
|
|
|
+AUTO_MERGE_THRESHOLD = 0.98
|
|
|
+SIGNAL_WEIGHTS = (
|
|
|
+ ("name", 0.50),
|
|
|
+ ("location", 0.15),
|
|
|
+ ("organization", 0.10),
|
|
|
+ ("responsible_person", 0.05),
|
|
|
+ ("model", 0.15),
|
|
|
+ ("source_code", 0.05),
|
|
|
+)
|
|
|
+SECRET_KEYS = frozenset(
|
|
|
+ {
|
|
|
+ "password",
|
|
|
+ "passwd",
|
|
|
+ "credential",
|
|
|
+ "credentials",
|
|
|
+ "token",
|
|
|
+ "api_key",
|
|
|
+ "authorization",
|
|
|
+ "connection_string",
|
|
|
+ "connection_url",
|
|
|
+ "phone",
|
|
|
+ "mobile",
|
|
|
+ "email",
|
|
|
+ "contact_phone",
|
|
|
+ "contact_email",
|
|
|
+ "raw_measurements",
|
|
|
+ "measurements",
|
|
|
+ }
|
|
|
+)
|
|
|
+
|
|
|
+
|
|
|
+@dataclass(frozen=True)
|
|
|
+class DevicePairScore:
|
|
|
+ confidence: float
|
|
|
+ explanation: tuple[dict[str, Any], ...]
|
|
|
+
|
|
|
+
|
|
|
+@dataclass(frozen=True)
|
|
|
+class DeviceEntityCandidateRecord:
|
|
|
+ uid: str
|
|
|
+ left_asset_uid: str
|
|
|
+ right_asset_uid: str
|
|
|
+ canonical_asset_uid: str | None
|
|
|
+ status: str
|
|
|
+ suggestion_source: str
|
|
|
+ confidence: float
|
|
|
+ explanation: tuple[dict[str, Any], ...]
|
|
|
+ evidence_uids: tuple[str, ...]
|
|
|
+ model_provider: str | None
|
|
|
+ model_name: str | None
|
|
|
+ current_version: int
|
|
|
+ created_by: str | None
|
|
|
+ reviewed_by: str | None
|
|
|
+ created_at: datetime | None = None
|
|
|
+ updated_at: datetime | None = None
|
|
|
+
|
|
|
+
|
|
|
+@dataclass(frozen=True)
|
|
|
+class DeviceEntityReviewRecord:
|
|
|
+ uid: str
|
|
|
+ candidate_uid: str
|
|
|
+ version: int
|
|
|
+ decision: str
|
|
|
+ reason: str
|
|
|
+ actor_uid: str
|
|
|
+ created_at: datetime | None = None
|
|
|
+
|
|
|
+
|
|
|
+@dataclass(frozen=True)
|
|
|
+class DeviceEntityMergeRecord:
|
|
|
+ uid: str
|
|
|
+ candidate_uid: str
|
|
|
+ canonical_asset_uid: str
|
|
|
+ member_asset_uid: str
|
|
|
+ review_uid: str
|
|
|
+ snapshot: dict[str, Any]
|
|
|
+ actor_uid: str
|
|
|
+ created_at: datetime | None = None
|
|
|
+
|
|
|
+
|
|
|
+@dataclass(frozen=True)
|
|
|
+class DeviceEntityRollbackRecord:
|
|
|
+ uid: str
|
|
|
+ merge_uid: str
|
|
|
+ candidate_uid: str
|
|
|
+ reason: str
|
|
|
+ snapshot: dict[str, Any]
|
|
|
+ actor_uid: str
|
|
|
+ created_at: datetime | None = None
|
|
|
+
|
|
|
+
|
|
|
+@dataclass(frozen=True)
|
|
|
+class DeviceEntityGenerationResult:
|
|
|
+ records: tuple[DeviceEntityCandidateRecord, ...]
|
|
|
+ created_count: int
|
|
|
+ existing_count: int
|
|
|
+ evaluated_pair_count: int
|
|
|
+ auto_merged_count: int
|
|
|
+
|
|
|
+
|
|
|
+def _normalized(value: Any) -> str:
|
|
|
+ text = unicodedata.normalize("NFKC", str(value or "")).casefold()
|
|
|
+ return "".join(character for character in text if character.isalnum())
|
|
|
+
|
|
|
+
|
|
|
+def _mapping_value(detail: DeviceAssetDetail, name: str) -> str:
|
|
|
+ if name == "source_code":
|
|
|
+ values = sorted(
|
|
|
+ {
|
|
|
+ _normalized(mapping.source_code)
|
|
|
+ for mapping in detail.mappings
|
|
|
+ if _normalized(mapping.source_code)
|
|
|
+ }
|
|
|
+ )
|
|
|
+ return "|".join(values)
|
|
|
+ return ""
|
|
|
+
|
|
|
+
|
|
|
+def _signal_value(detail: DeviceAssetDetail, name: str) -> str:
|
|
|
+ if name == "model":
|
|
|
+ return _normalized((detail.asset.attributes or {}).get("model"))
|
|
|
+ if name == "source_code":
|
|
|
+ return _mapping_value(detail, name)
|
|
|
+ return _normalized(getattr(detail.asset, name, None))
|
|
|
+
|
|
|
+
|
|
|
+def score_device_pair(
|
|
|
+ left: DeviceAssetDetail,
|
|
|
+ right: DeviceAssetDetail,
|
|
|
+) -> DevicePairScore:
|
|
|
+ explanation = []
|
|
|
+ confidence = 0.0
|
|
|
+ for signal, weight in SIGNAL_WEIGHTS:
|
|
|
+ left_value = _signal_value(left, signal)
|
|
|
+ right_value = _signal_value(right, signal)
|
|
|
+ matched = bool(left_value and right_value and left_value == right_value)
|
|
|
+ if matched:
|
|
|
+ confidence += weight
|
|
|
+ explanation.append(
|
|
|
+ {
|
|
|
+ "signal": signal,
|
|
|
+ "matched": matched,
|
|
|
+ "weight": weight,
|
|
|
+ "left": left_value or None,
|
|
|
+ "right": right_value or None,
|
|
|
+ }
|
|
|
+ )
|
|
|
+ return DevicePairScore(
|
|
|
+ confidence=round(confidence, 6),
|
|
|
+ explanation=tuple(explanation),
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def _required_text(
|
|
|
+ payload: dict[str, Any],
|
|
|
+ name: str,
|
|
|
+ *,
|
|
|
+ maximum: int = 1_000,
|
|
|
+) -> str:
|
|
|
+ value = str(payload.get(name) or "").strip()
|
|
|
+ if not value:
|
|
|
+ raise DeviceEntityInvalid(f"{name} is required")
|
|
|
+ if len(value) > maximum:
|
|
|
+ raise DeviceEntityInvalid(f"{name} exceeds {maximum} characters")
|
|
|
+ return value
|
|
|
+
|
|
|
+
|
|
|
+def _bounded_float(payload, name, *, default=None):
|
|
|
+ raw = payload.get(name, default)
|
|
|
+ try:
|
|
|
+ value = float(raw)
|
|
|
+ except (TypeError, ValueError) as exc:
|
|
|
+ raise DeviceEntityInvalid(f"{name} must be a number") from exc
|
|
|
+ if value < 0 or value > 1:
|
|
|
+ raise DeviceEntityInvalid(f"{name} must be between 0 and 1")
|
|
|
+ return value
|
|
|
+
|
|
|
+
|
|
|
+def _positive_int(payload, name, *, default, maximum):
|
|
|
+ raw = payload.get(name, default)
|
|
|
+ try:
|
|
|
+ value = int(raw)
|
|
|
+ except (TypeError, ValueError) as exc:
|
|
|
+ raise DeviceEntityInvalid(f"{name} must be an integer") from exc
|
|
|
+ if value < 1 or value > maximum:
|
|
|
+ raise DeviceEntityInvalid(
|
|
|
+ f"{name} must be between 1 and {maximum}"
|
|
|
+ )
|
|
|
+ return value
|
|
|
+
|
|
|
+
|
|
|
+def _reject_sensitive(value: Any) -> None:
|
|
|
+ if isinstance(value, dict):
|
|
|
+ for key, item in value.items():
|
|
|
+ if str(key).strip().lower() in SECRET_KEYS:
|
|
|
+ raise DeviceEntityInvalid(
|
|
|
+ "sensitive fields are not allowed in entity resolution"
|
|
|
+ )
|
|
|
+ _reject_sensitive(item)
|
|
|
+ elif isinstance(value, (list, tuple)):
|
|
|
+ for item in value:
|
|
|
+ _reject_sensitive(item)
|
|
|
+
|
|
|
+
|
|
|
+def _evidence_uids(payload, *, required=True) -> tuple[str, ...]:
|
|
|
+ values = payload.get("evidence_uids")
|
|
|
+ if not isinstance(values, list):
|
|
|
+ raise DeviceEntityInvalid("evidence_uids must be an array")
|
|
|
+ records = tuple(
|
|
|
+ dict.fromkeys(
|
|
|
+ str(value).strip()
|
|
|
+ for value in values
|
|
|
+ if str(value).strip()
|
|
|
+ )
|
|
|
+ )
|
|
|
+ if required and not records:
|
|
|
+ raise DeviceEntityInvalid("evidence_uids are required")
|
|
|
+ if len(records) > 100 or any(len(item) > 300 for item in records):
|
|
|
+ raise DeviceEntityInvalid("evidence_uids exceed the governed boundary")
|
|
|
+ return records
|
|
|
+
|
|
|
+
|
|
|
+def _pair(left_uid: str, right_uid: str) -> tuple[str, str]:
|
|
|
+ left_uid = str(left_uid).strip()
|
|
|
+ right_uid = str(right_uid).strip()
|
|
|
+ if not left_uid or not right_uid or left_uid == right_uid:
|
|
|
+ raise DeviceEntityInvalid("candidate requires two distinct assets")
|
|
|
+ return tuple(sorted((left_uid, right_uid)))
|
|
|
+
|
|
|
+
|
|
|
+def _source_uids(detail: DeviceAssetDetail) -> set[str]:
|
|
|
+ return {str(mapping.source_uid) for mapping in detail.mappings}
|
|
|
+
|
|
|
+
|
|
|
+def _asset_snapshot(detail: DeviceAssetDetail) -> dict[str, Any]:
|
|
|
+ return {
|
|
|
+ "uid": detail.asset.uid,
|
|
|
+ "asset_type": detail.asset.asset_type,
|
|
|
+ "name": detail.asset.name,
|
|
|
+ "location": detail.asset.location,
|
|
|
+ "organization": detail.asset.organization,
|
|
|
+ "responsible_person": detail.asset.responsible_person,
|
|
|
+ "model": (detail.asset.attributes or {}).get("model"),
|
|
|
+ "source_mappings": [
|
|
|
+ {
|
|
|
+ "uid": mapping.uid,
|
|
|
+ "source_uid": mapping.source_uid,
|
|
|
+ "source_entity": mapping.source_entity,
|
|
|
+ "source_code": mapping.source_code,
|
|
|
+ }
|
|
|
+ for mapping in detail.mappings
|
|
|
+ ],
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+class DeviceEntityResolutionService:
|
|
|
+ def __init__(
|
|
|
+ self,
|
|
|
+ repository,
|
|
|
+ *,
|
|
|
+ review_authorizer: Callable[[str], Any] = lambda _actor_uid: None,
|
|
|
+ uid_factory: Callable[[], str] = new_governance_uid,
|
|
|
+ now_factory: Callable[[], datetime] = now_china,
|
|
|
+ auto_merge_enabled: bool = False,
|
|
|
+ commit: Callable[[], Any] = lambda: None,
|
|
|
+ rollback: Callable[[], Any] = lambda: None,
|
|
|
+ ):
|
|
|
+ self.repository = repository
|
|
|
+ self.review_authorizer = review_authorizer
|
|
|
+ self.uid_factory = uid_factory
|
|
|
+ self.now_factory = now_factory
|
|
|
+ self.auto_merge_enabled = bool(auto_merge_enabled)
|
|
|
+ self.commit = commit
|
|
|
+ self.rollback_transaction = rollback
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _validate_pair(left, right):
|
|
|
+ if left is None or right is None:
|
|
|
+ raise DeviceEntityNotFound("candidate asset was not found")
|
|
|
+ if left.asset.status != "active" or right.asset.status != "active":
|
|
|
+ raise DeviceEntityInvalid("candidate assets must be active")
|
|
|
+ if left.asset.asset_type != right.asset.asset_type:
|
|
|
+ raise DeviceEntityInvalid("candidate assets must have the same type")
|
|
|
+ if _source_uids(left) & _source_uids(right):
|
|
|
+ raise DeviceEntityInvalid(
|
|
|
+ "candidate assets must come from different source systems"
|
|
|
+ )
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _rule_evidence(left, right):
|
|
|
+ return tuple(
|
|
|
+ dict.fromkeys(
|
|
|
+ (
|
|
|
+ left.asset.uid,
|
|
|
+ *(mapping.uid for mapping in left.mappings),
|
|
|
+ right.asset.uid,
|
|
|
+ *(mapping.uid for mapping in right.mappings),
|
|
|
+ )
|
|
|
+ )
|
|
|
+ )
|
|
|
+
|
|
|
+ def generate(self, payload, actor_uid) -> DeviceEntityGenerationResult:
|
|
|
+ if not isinstance(payload, dict):
|
|
|
+ raise DeviceEntityInvalid("payload must be an object")
|
|
|
+ asset_type = str(payload.get("asset_type") or "device").strip()
|
|
|
+ if asset_type not in ASSET_TYPES:
|
|
|
+ raise DeviceEntityInvalid("asset_type is not supported")
|
|
|
+ threshold = _bounded_float(payload, "threshold", default=0.7)
|
|
|
+ limit = _positive_int(
|
|
|
+ payload,
|
|
|
+ "limit",
|
|
|
+ default=MAX_MATCH_ASSETS,
|
|
|
+ maximum=MAX_MATCH_ASSETS,
|
|
|
+ )
|
|
|
+ details = self.repository.list_matchable_assets(
|
|
|
+ asset_type,
|
|
|
+ limit=limit,
|
|
|
+ )
|
|
|
+ records = []
|
|
|
+ created_count = 0
|
|
|
+ existing_count = 0
|
|
|
+ evaluated = 0
|
|
|
+ auto_merged_count = 0
|
|
|
+ try:
|
|
|
+ for first, second in combinations(details, 2):
|
|
|
+ if len(records) >= MAX_NEW_CANDIDATES:
|
|
|
+ break
|
|
|
+ left_uid, right_uid = _pair(
|
|
|
+ first.asset.uid,
|
|
|
+ second.asset.uid,
|
|
|
+ )
|
|
|
+ left = first if first.asset.uid == left_uid else second
|
|
|
+ right = second if second.asset.uid == right_uid else first
|
|
|
+ try:
|
|
|
+ self._validate_pair(left, right)
|
|
|
+ except DeviceEntityInvalid:
|
|
|
+ continue
|
|
|
+ evaluated += 1
|
|
|
+ score = score_device_pair(left, right)
|
|
|
+ if score.confidence < threshold:
|
|
|
+ continue
|
|
|
+ existing = self.repository.find_open_pair(
|
|
|
+ left_uid,
|
|
|
+ right_uid,
|
|
|
+ )
|
|
|
+ if existing is not None:
|
|
|
+ records.append(existing)
|
|
|
+ existing_count += 1
|
|
|
+ continue
|
|
|
+ now = self.now_factory()
|
|
|
+ candidate = DeviceEntityCandidateRecord(
|
|
|
+ uid=self.uid_factory(),
|
|
|
+ left_asset_uid=left_uid,
|
|
|
+ right_asset_uid=right_uid,
|
|
|
+ canonical_asset_uid=None,
|
|
|
+ status="pending",
|
|
|
+ suggestion_source="rule",
|
|
|
+ confidence=score.confidence,
|
|
|
+ explanation=score.explanation,
|
|
|
+ evidence_uids=self._rule_evidence(left, right),
|
|
|
+ model_provider=None,
|
|
|
+ model_name=None,
|
|
|
+ current_version=1,
|
|
|
+ created_by=actor_uid,
|
|
|
+ reviewed_by=None,
|
|
|
+ created_at=now,
|
|
|
+ updated_at=now,
|
|
|
+ )
|
|
|
+ candidate = self.repository.create_candidate(candidate)
|
|
|
+ created_count += 1
|
|
|
+ if (
|
|
|
+ self.auto_merge_enabled
|
|
|
+ and candidate.confidence >= AUTO_MERGE_THRESHOLD
|
|
|
+ ):
|
|
|
+ self.review_authorizer(actor_uid)
|
|
|
+ candidate, _review, _merge = self._decide(
|
|
|
+ candidate,
|
|
|
+ decision="auto_approve",
|
|
|
+ canonical_asset_uid=left_uid,
|
|
|
+ reason="strict deterministic auto-merge threshold met",
|
|
|
+ actor_uid=actor_uid,
|
|
|
+ )
|
|
|
+ auto_merged_count += 1
|
|
|
+ records.append(candidate)
|
|
|
+ self.commit()
|
|
|
+ except Exception:
|
|
|
+ self.rollback_transaction()
|
|
|
+ raise
|
|
|
+ return DeviceEntityGenerationResult(
|
|
|
+ records=tuple(records),
|
|
|
+ created_count=created_count,
|
|
|
+ existing_count=existing_count,
|
|
|
+ evaluated_pair_count=evaluated,
|
|
|
+ auto_merged_count=auto_merged_count,
|
|
|
+ )
|
|
|
+
|
|
|
+ def submit_ai_candidate(self, payload, actor_uid):
|
|
|
+ if not isinstance(payload, dict):
|
|
|
+ raise DeviceEntityInvalid("payload must be an object")
|
|
|
+ _reject_sensitive(payload)
|
|
|
+ left_uid, right_uid = _pair(
|
|
|
+ payload.get("left_asset_uid"),
|
|
|
+ payload.get("right_asset_uid"),
|
|
|
+ )
|
|
|
+ left = self.repository.get_asset_detail(left_uid)
|
|
|
+ right = self.repository.get_asset_detail(right_uid)
|
|
|
+ self._validate_pair(left, right)
|
|
|
+ if self.repository.active_merge_for_member(left_uid) or (
|
|
|
+ self.repository.active_merge_for_member(right_uid)
|
|
|
+ ):
|
|
|
+ raise DeviceEntityConflict(
|
|
|
+ "candidate asset already belongs to an active merge"
|
|
|
+ )
|
|
|
+ if self.repository.find_open_pair(left_uid, right_uid):
|
|
|
+ raise DeviceEntityConflict("candidate pair already has an active merge")
|
|
|
+ confidence = _bounded_float(payload, "confidence")
|
|
|
+ provider = _required_text(payload, "model_provider", maximum=200)
|
|
|
+ model_name = _required_text(payload, "model_name", maximum=200)
|
|
|
+ explanation_text = _required_text(
|
|
|
+ payload,
|
|
|
+ "explanation",
|
|
|
+ maximum=2_000,
|
|
|
+ )
|
|
|
+ evidence = _evidence_uids(payload)
|
|
|
+ now = self.now_factory()
|
|
|
+ candidate = DeviceEntityCandidateRecord(
|
|
|
+ uid=self.uid_factory(),
|
|
|
+ left_asset_uid=left_uid,
|
|
|
+ right_asset_uid=right_uid,
|
|
|
+ canonical_asset_uid=None,
|
|
|
+ status="pending",
|
|
|
+ suggestion_source="ai",
|
|
|
+ confidence=confidence,
|
|
|
+ explanation=(
|
|
|
+ {
|
|
|
+ "signal": "ai_explanation",
|
|
|
+ "matched": True,
|
|
|
+ "weight": confidence,
|
|
|
+ "left": explanation_text,
|
|
|
+ "right": None,
|
|
|
+ },
|
|
|
+ ),
|
|
|
+ evidence_uids=evidence,
|
|
|
+ model_provider=provider,
|
|
|
+ model_name=model_name,
|
|
|
+ current_version=1,
|
|
|
+ created_by=actor_uid,
|
|
|
+ reviewed_by=None,
|
|
|
+ created_at=now,
|
|
|
+ updated_at=now,
|
|
|
+ )
|
|
|
+ try:
|
|
|
+ candidate = self.repository.create_candidate(candidate)
|
|
|
+ self.commit()
|
|
|
+ return candidate
|
|
|
+ except Exception:
|
|
|
+ self.rollback_transaction()
|
|
|
+ raise
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _version(value):
|
|
|
+ try:
|
|
|
+ version = int(value)
|
|
|
+ except (TypeError, ValueError) as exc:
|
|
|
+ raise DeviceEntityInvalid("expected_version must be an integer") from exc
|
|
|
+ if version < 1:
|
|
|
+ raise DeviceEntityInvalid("expected_version must be positive")
|
|
|
+ return version
|
|
|
+
|
|
|
+ def _candidate(self, uid, *, for_update=False):
|
|
|
+ candidate = self.repository.get_candidate(
|
|
|
+ str(uid),
|
|
|
+ for_update=for_update,
|
|
|
+ )
|
|
|
+ if candidate is None:
|
|
|
+ raise DeviceEntityNotFound(f"candidate {uid} was not found")
|
|
|
+ return candidate
|
|
|
+
|
|
|
+ def _decide(
|
|
|
+ self,
|
|
|
+ candidate,
|
|
|
+ *,
|
|
|
+ decision,
|
|
|
+ canonical_asset_uid,
|
|
|
+ reason,
|
|
|
+ actor_uid,
|
|
|
+ ):
|
|
|
+ if decision not in {"approve", "reject", "auto_approve"}:
|
|
|
+ raise DeviceEntityInvalid("decision is not supported")
|
|
|
+ now = self.now_factory()
|
|
|
+ review = DeviceEntityReviewRecord(
|
|
|
+ uid=self.uid_factory(),
|
|
|
+ candidate_uid=candidate.uid,
|
|
|
+ version=candidate.current_version,
|
|
|
+ decision=decision,
|
|
|
+ reason=reason,
|
|
|
+ actor_uid=actor_uid,
|
|
|
+ created_at=now,
|
|
|
+ )
|
|
|
+ review = self.repository.append_review(review)
|
|
|
+ if decision == "reject":
|
|
|
+ candidate = replace(
|
|
|
+ candidate,
|
|
|
+ status="rejected",
|
|
|
+ current_version=candidate.current_version + 1,
|
|
|
+ reviewed_by=actor_uid,
|
|
|
+ updated_at=now,
|
|
|
+ )
|
|
|
+ return self.repository.update_candidate(candidate), review, None
|
|
|
+
|
|
|
+ pair = {candidate.left_asset_uid, candidate.right_asset_uid}
|
|
|
+ if canonical_asset_uid not in pair:
|
|
|
+ raise DeviceEntityInvalid(
|
|
|
+ "canonical_asset_uid must belong to the candidate pair"
|
|
|
+ )
|
|
|
+ member_uid = next(uid for uid in pair if uid != canonical_asset_uid)
|
|
|
+ if self.repository.active_merge_for_member(canonical_asset_uid):
|
|
|
+ raise DeviceEntityConflict(
|
|
|
+ "canonical asset already belongs to an active merge"
|
|
|
+ )
|
|
|
+ if self.repository.active_merge_for_member(member_uid):
|
|
|
+ raise DeviceEntityConflict(
|
|
|
+ "member asset already belongs to an active merge"
|
|
|
+ )
|
|
|
+ canonical = self.repository.get_asset_detail(canonical_asset_uid)
|
|
|
+ member = self.repository.get_asset_detail(member_uid)
|
|
|
+ self._validate_pair(canonical, member)
|
|
|
+ merge = DeviceEntityMergeRecord(
|
|
|
+ uid=self.uid_factory(),
|
|
|
+ candidate_uid=candidate.uid,
|
|
|
+ canonical_asset_uid=canonical_asset_uid,
|
|
|
+ member_asset_uid=member_uid,
|
|
|
+ review_uid=review.uid,
|
|
|
+ snapshot={
|
|
|
+ "candidate": {
|
|
|
+ "uid": candidate.uid,
|
|
|
+ "confidence": candidate.confidence,
|
|
|
+ "suggestion_source": candidate.suggestion_source,
|
|
|
+ "evidence_uids": list(candidate.evidence_uids),
|
|
|
+ },
|
|
|
+ "canonical": _asset_snapshot(canonical),
|
|
|
+ "member": _asset_snapshot(member),
|
|
|
+ },
|
|
|
+ actor_uid=actor_uid,
|
|
|
+ created_at=now,
|
|
|
+ )
|
|
|
+ merge = self.repository.create_merge(merge)
|
|
|
+ candidate = replace(
|
|
|
+ candidate,
|
|
|
+ canonical_asset_uid=canonical_asset_uid,
|
|
|
+ status="merged",
|
|
|
+ current_version=candidate.current_version + 1,
|
|
|
+ reviewed_by=actor_uid,
|
|
|
+ updated_at=now,
|
|
|
+ )
|
|
|
+ return self.repository.update_candidate(candidate), review, merge
|
|
|
+
|
|
|
+ def review(self, candidate_uid, payload, actor_uid):
|
|
|
+ if not isinstance(payload, dict):
|
|
|
+ raise DeviceEntityInvalid("payload must be an object")
|
|
|
+ decision = str(payload.get("decision") or "").strip()
|
|
|
+ reason = _required_text(payload, "reason", maximum=1_000)
|
|
|
+ expected_version = self._version(payload.get("expected_version"))
|
|
|
+ self.review_authorizer(actor_uid)
|
|
|
+ try:
|
|
|
+ candidate = self._candidate(candidate_uid, for_update=True)
|
|
|
+ if candidate.status != "pending":
|
|
|
+ raise DeviceEntityConflict(
|
|
|
+ "candidate is not pending review"
|
|
|
+ )
|
|
|
+ if candidate.current_version != expected_version:
|
|
|
+ raise DeviceEntityConflict("candidate version conflict")
|
|
|
+ result = self._decide(
|
|
|
+ candidate,
|
|
|
+ decision=decision,
|
|
|
+ canonical_asset_uid=str(
|
|
|
+ payload.get("canonical_asset_uid") or ""
|
|
|
+ ).strip(),
|
|
|
+ reason=reason,
|
|
|
+ actor_uid=actor_uid,
|
|
|
+ )
|
|
|
+ self.commit()
|
|
|
+ return result
|
|
|
+ except Exception:
|
|
|
+ self.rollback_transaction()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def rollback(self, merge_uid, payload, actor_uid):
|
|
|
+ if not isinstance(payload, dict):
|
|
|
+ raise DeviceEntityInvalid("payload must be an object")
|
|
|
+ expected_version = self._version(payload.get("expected_version"))
|
|
|
+ reason = _required_text(payload, "reason", maximum=1_000)
|
|
|
+ self.review_authorizer(actor_uid)
|
|
|
+ try:
|
|
|
+ merge = self.repository.get_merge(
|
|
|
+ str(merge_uid),
|
|
|
+ for_update=True,
|
|
|
+ )
|
|
|
+ if merge is None:
|
|
|
+ raise DeviceEntityNotFound(f"merge {merge_uid} was not found")
|
|
|
+ if self.repository.list_rollbacks(merge.uid):
|
|
|
+ raise DeviceEntityConflict("merge was already rolled back")
|
|
|
+ candidate = self._candidate(
|
|
|
+ merge.candidate_uid,
|
|
|
+ for_update=True,
|
|
|
+ )
|
|
|
+ if candidate.status != "merged":
|
|
|
+ raise DeviceEntityConflict("candidate is not actively merged")
|
|
|
+ if candidate.current_version != expected_version:
|
|
|
+ raise DeviceEntityConflict("candidate version conflict")
|
|
|
+ now = self.now_factory()
|
|
|
+ rollback = DeviceEntityRollbackRecord(
|
|
|
+ uid=self.uid_factory(),
|
|
|
+ merge_uid=merge.uid,
|
|
|
+ candidate_uid=candidate.uid,
|
|
|
+ reason=reason,
|
|
|
+ snapshot={
|
|
|
+ "merge": {
|
|
|
+ "uid": merge.uid,
|
|
|
+ "canonical_asset_uid": merge.canonical_asset_uid,
|
|
|
+ "member_asset_uid": merge.member_asset_uid,
|
|
|
+ "review_uid": merge.review_uid,
|
|
|
+ "created_at": (
|
|
|
+ merge.created_at.isoformat()
|
|
|
+ if merge.created_at
|
|
|
+ else None
|
|
|
+ ),
|
|
|
+ },
|
|
|
+ "candidate_version_before": candidate.current_version,
|
|
|
+ },
|
|
|
+ actor_uid=actor_uid,
|
|
|
+ created_at=now,
|
|
|
+ )
|
|
|
+ rollback = self.repository.create_rollback(rollback)
|
|
|
+ candidate = replace(
|
|
|
+ candidate,
|
|
|
+ status="rolled_back",
|
|
|
+ current_version=candidate.current_version + 1,
|
|
|
+ reviewed_by=actor_uid,
|
|
|
+ updated_at=now,
|
|
|
+ )
|
|
|
+ candidate = self.repository.update_candidate(candidate)
|
|
|
+ self.commit()
|
|
|
+ return candidate, rollback
|
|
|
+ except Exception:
|
|
|
+ self.rollback_transaction()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def search(self, filters, *, page=1, page_size=20):
|
|
|
+ if not isinstance(filters, dict):
|
|
|
+ raise DeviceEntityInvalid("filters must be an object")
|
|
|
+ normalized = {}
|
|
|
+ status = str(filters.get("status") or "").strip()
|
|
|
+ if status:
|
|
|
+ if status not in {"pending", "merged", "rejected", "rolled_back"}:
|
|
|
+ raise DeviceEntityInvalid("status is not supported")
|
|
|
+ normalized["status"] = status
|
|
|
+ source = str(filters.get("suggestion_source") or "").strip()
|
|
|
+ if source:
|
|
|
+ if source not in {"rule", "ai", "manual"}:
|
|
|
+ raise DeviceEntityInvalid("suggestion_source is not supported")
|
|
|
+ normalized["suggestion_source"] = source
|
|
|
+ page = _positive_int(
|
|
|
+ {"page": page},
|
|
|
+ "page",
|
|
|
+ default=1,
|
|
|
+ maximum=1_000_000,
|
|
|
+ )
|
|
|
+ page_size = _positive_int(
|
|
|
+ {"page_size": page_size},
|
|
|
+ "page_size",
|
|
|
+ default=20,
|
|
|
+ maximum=MAX_PAGE_SIZE,
|
|
|
+ )
|
|
|
+ return self.repository.search_candidates(
|
|
|
+ normalized,
|
|
|
+ page=page,
|
|
|
+ page_size=page_size,
|
|
|
+ )
|
|
|
+
|
|
|
+ def get(self, candidate_uid):
|
|
|
+ return self._candidate(candidate_uid)
|
|
|
+
|
|
|
+ def reviews(self, candidate_uid):
|
|
|
+ self._candidate(candidate_uid)
|
|
|
+ return self.repository.list_reviews(str(candidate_uid))
|
|
|
+
|
|
|
+ def merges(self, candidate_uid):
|
|
|
+ self._candidate(candidate_uid)
|
|
|
+ return self.repository.list_merges(str(candidate_uid))
|
|
|
+
|
|
|
+ def rollbacks(self, merge_uid):
|
|
|
+ merge = self.repository.get_merge(str(merge_uid))
|
|
|
+ if merge is None:
|
|
|
+ raise DeviceEntityNotFound(f"merge {merge_uid} was not found")
|
|
|
+ return self.repository.list_rollbacks(str(merge_uid))
|
|
|
+
|
|
|
+
|
|
|
+__all__ = [
|
|
|
+ "AUTO_MERGE_THRESHOLD",
|
|
|
+ "DeviceEntityCandidateRecord",
|
|
|
+ "DeviceEntityConflict",
|
|
|
+ "DeviceEntityForbidden",
|
|
|
+ "DeviceEntityGenerationResult",
|
|
|
+ "DeviceEntityInvalid",
|
|
|
+ "DeviceEntityMergeRecord",
|
|
|
+ "DeviceEntityNotFound",
|
|
|
+ "DeviceEntityResolutionService",
|
|
|
+ "DeviceEntityReviewRecord",
|
|
|
+ "DeviceEntityRollbackRecord",
|
|
|
+ "DevicePairScore",
|
|
|
+ "score_device_pair",
|
|
|
+]
|