Changeset 02d288d in flexoentity for flexoentity/flexo_entity.py

Timestamp:

10/23/25 13:27:08 (3 months ago)

Author:

Enrico Schwass <ennoausberlin@…>

Branches:

master

Children:

4ceca57

Parents:

6a7dec1

Message:

improve hash generation and collision handler - move signature from FlexOID to FlexoEntity

File:

• flexoentity/flexo_entity.py (modified) (8 diffs)

flexoentity/flexo_entity.py

@@ -89 +89 @@
 
 
-@dataclass
+@dataclass(kw_only=True)
 class FlexoEntity(ABC):
     domain: str
     etype: EntityType 
     state: EntityState
-
+    flexo_id: Optional[FlexOID] = field(default=None)
+    signature: str = field(default_factory=str)
+    origin: Optional[str] = field(default=None)
+    
     OID_PATTERN = re.compile(
         r"^(?P<domain>[A-Z0-9]+)-(?P<etype>[A-Z]+)"
@@ -101 +104 @@
 
     def __str__(self) -> str:
-        return f"{self.domain}-{self.etype}{self.date}-{self.unique_hash}@{self.version:03d}{self.state}"
+        return f"{self.domain_code()}-{self.etype}{self.date}-{self.unique_hash}@{self.version:03d}{self.state}"
 
     @classmethod
@@ -121 +124 @@
         raise NotImplementedError("Subclasses must define text_seed property")
 
+    def canonical_seed(self) -> str:
+        return canonical_seed(self.text_seed)
+
     @classmethod
     @abstractmethod
@@ -127 +133 @@
         raise NotImplementedError("Subclasses must implement default()")
 
+    def domain_code(self) -> str:
+        """Return canonical domain code for serialization and ID generation."""
+        return self.domain.domain if hasattr(self.domain, "domain") else self.domain
+
     def __post_init__(self):
-        """Generate ID and content fingerprint."""
-        self.flexo_id = FlexOID.generate(self.domain,
+        """
+        Generate ID and content fingerprint.
+        
+        All entities must carry a `.domain` attribute exposing a domain code string.
+        This may be a `Domain` instance or a temporary wrapper used by the `Domain`
+        class itself to avoid circular initialization.
+        """
+       
+        self.flexo_id = FlexOID.generate(self.domain_code(),
                                          self.etype.short(),
                                          self.state.short(),
                                          self.text_seed,
                                          1)
+        seed = canonical_seed(self.text_seed)
+        self.signature = hashlib.blake2s(seed.encode("utf-8"), digest_size=8).hexdigest().upper()
 
     def __str__(self):
         return (
             f"{self.etype.name}({self.flexo_id}, {self.state.name}, "
-            f"sig={self.flexo_id.signature}..., v{self.version})"
-        )
-
+            f"sig={self.signature}..., v{self.version})"
+        )
     def to_dict(self):
         return {
-            "domain": self.domain,
+            "domain": self.domain_code(),
             "etype": self.etype.name,
             "state": self.state.name,
             "flexo_id": str(self.flexo_id),
+            "signature": self.signature,
+            "origin": self.origin,
         }
-
+    
     @classmethod
     def from_dict(cls, data):
+        from flexoentity.domain import Domain  # avoid circular import
+        domain_obj = Domain(
+            domain=data["domain"],
+            etype=EntityType.DOMAIN,
+            state=EntityState.DRAFT,  # default when reconstructing context
+        )
         obj = cls(
-            data["domain"],
-            EntityType[data["etype"]],
-            data["text_seed"],
-            EntityState[data["state"]],
-        )
-        obj.flexo_id = FlexOID(data["flexo_id"], data.get("signature", ""))
+            domain=domain_obj,
+            etype=EntityType[data["etype"]],
+            state=EntityState[data["state"]],
+        )
+        obj.flexo_id = FlexOID.from_string(data["flexo_id"])
+        obj.signature = data.get("signature", "")
+        obj.origin = data.get("origin")
         return obj
-
+      
     def to_json(self, *, indent: int | None = None) -> str:
         """Serialize entity (and its FlexOID) into JSON."""
@@ -184 +211 @@
             EntityState.PUBLISHED,
         )
-    
-    # ───────────────────────────────────────────────────────────────
+    def _update_signature(self) -> str:
+        """Always recompute the entity's content signature."""
+        seed = self.canonical_seed()
+        return hashlib.blake2s(seed.encode("utf-8"), digest_size=8).hexdigest().upper()
+
     def _update_fingerprint(self) -> bool:
-        """Recalculate fingerprint and return True if content changed."""
-        # extract version from current flexo_id
-        new_oid = FlexOID.generate(self.domain, self.etype.short(), self.state.short(), self.text_seed, self.flexo_id.version)
-        if new_oid.signature != self.flexo_id.signature:
-            self.flexo_id = new_oid
+        """Update FlexOID if the content signature changed."""
+        new_sig = self._update_signature()
+        if new_sig != self.signature:
+            self.signature = new_sig
+            self.flexo_id = FlexOID.generate(self.domain_code(),
+                                             self.etype.short(),
+                                             self.state.short(),
+                                             self.text_seed,
+                                             self.flexo_id.version)
             return True
         return False
-
+   
     # ───────────────────────────────────────────────────────────────
     def _transition(self, target_state: EntityState):
@@ -234 +268 @@
         if self.state == EntityState.DRAFT:
             new_version = self.flexo_id.version + 1
-            new_fid = FlexOID.generate(
-                self.domain,
+            new_fid = FlexOID.generate(self.domain_code(),
                 self.etype.short(),
                 EntityState.APPROVED.short(),
@@ -262 +295 @@
         """Start new lineage when obsolete."""
         self.flexo_id = FlexOID.clone_new_base(
-            self.domain,
+            self.domain_code(),
             self.etype.short(),
             self.state.short(),
@@ -276 +309 @@
     @staticmethod
     def verify_integrity(entity) -> bool:
-        # --- inhaltlicher (kryptographischer) Check ---
-        # Hash ohne State, Signatur mit State
-        hash_seed = canonical_seed(f"{entity.domain}:{entity.etype.short()}:{entity.text_seed}")
-        sig_seed  = f"{hash_seed}:{entity.state.short()}"
-
-        expected_sig = hashlib.blake2s(sig_seed.encode("utf-8"), digest_size=8).hexdigest().upper()
-        return expected_sig == entity.flexo_id.signature
+        """Verify that an entity’s content signature matches its actual content."""
+        expected_sig = hashlib.blake2s(
+            canonical_seed(entity.text_seed).encode("utf-8"), digest_size=8
+        ).hexdigest().upper()
+        return expected_sig == entity.signature
 
     def allowed_transitions(self) -> list[str]: