from __future__ import annotations from app.core.knowledge.contracts import ( KnowledgeSnapshot, PointChange, SnapshotDiff, ) def diff_snapshots(old: KnowledgeSnapshot, new: KnowledgeSnapshot) -> SnapshotDiff: if (old.source_type, old.source_uid) != (new.source_type, new.source_uid): raise ValueError("knowledge snapshots must describe the same source object") old_points = {point.point_key: point for point in old.points} new_points = {point.point_key: point for point in new.points} groups: dict[str, list[PointChange]] = { "added": [], "modified": [], "deleted": [], "unchanged": [], } for point_key in sorted(set(old_points) | set(new_points)): old_point = old_points.get(point_key) new_point = new_points.get(point_key) if old_point is None: kind = "added" elif new_point is None: kind = "deleted" elif ( old_point.content_hash, old_point.metadata_hash, old_point.permission_hash, ) == ( new_point.content_hash, new_point.metadata_hash, new_point.permission_hash, ): kind = "unchanged" else: kind = "modified" groups[kind].append( PointChange( point_key=point_key, change_kind=kind, old=old_point, new=new_point, permission_changed=bool( old_point and new_point and old_point.permission_hash != new_point.permission_hash ), ) ) return SnapshotDiff( added=tuple(groups["added"]), modified=tuple(groups["modified"]), deleted=tuple(groups["deleted"]), unchanged=tuple(groups["unchanged"]), )