Spaces:
Sleeping
Sleeping
File size: 1,538 Bytes
2e818da | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 | """Typed proposals emitted by Commit before either memory backend is mutated."""
from __future__ import annotations
import hashlib
import re
from datetime import datetime, timezone
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field
class MemoryCandidate(BaseModel):
model_config = ConfigDict(strict=True, extra="forbid")
candidate_id: str
destination: Literal["project", "student"]
project_id: str
kind: str
statement: str
attribution: Literal[
"explicit_student",
"idea_observer_interaction",
"idea_observer_profile_proposal",
]
confidence: float = Field(ge=0.0, le=1.0)
evidence_ids: list[str] = Field(default_factory=list)
interaction_ids: list[str] = Field(default_factory=list)
observed_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
@property
def recurrence_key(self) -> str:
normalized = re.sub(r"\s+", " ", self.statement.casefold()).strip()
return hashlib.sha256(f"{self.kind}\0{normalized}".encode("utf-8")).hexdigest()
def make_candidate_id(
destination: str,
project_id: str,
kind: str,
statement: str,
) -> str:
normalized = re.sub(r"\s+", " ", statement.casefold()).strip()
digest = hashlib.sha256(
f"{destination}\0{project_id}\0{kind}\0{normalized}".encode("utf-8")
).hexdigest()
return f"mc_{digest}"
def interaction_ref(value: str) -> str:
return "ji_" + hashlib.sha256(value.strip().encode("utf-8")).hexdigest()[:24]
|