123 lines
3.9 KiB
Python
123 lines
3.9 KiB
Python
"""Helpers for preserving Alexander request/response artifacts in MemPalace.
|
|
|
|
This module provides a small, typed bridge between raw conversation turns and
|
|
MemPalace drawers stored in the shared `sovereign` room. The goal is not to
|
|
solve all future speaker-tagging needs at once; it gives the Nexus one
|
|
canonical artifact shape that other miners and bridges can reuse.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
from typing import Iterable
|
|
|
|
_ALEXANDER_ALIASES = {
|
|
"alexander",
|
|
"alexander whitestone",
|
|
"rockachopa",
|
|
"triptimmy",
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ConversationArtifact:
|
|
requester: str
|
|
responder: str
|
|
request_text: str
|
|
response_text: str
|
|
room: str = "sovereign"
|
|
timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"))
|
|
metadata: dict = field(default_factory=dict)
|
|
|
|
@property
|
|
def text(self) -> str:
|
|
return (
|
|
f"# Conversation Artifact\n\n"
|
|
f"## Alexander Request\n{self.request_text.strip()}\n\n"
|
|
f"## Wizard Response\n{self.response_text.strip()}\n"
|
|
)
|
|
|
|
|
|
def normalize_speaker(name: str | None) -> str:
|
|
cleaned = " ".join((name or "").strip().lower().split())
|
|
if cleaned in _ALEXANDER_ALIASES:
|
|
return "alexander"
|
|
return cleaned.replace(" ", "_") or "unknown"
|
|
|
|
|
|
def build_request_response_artifact(
|
|
*,
|
|
requester: str,
|
|
responder: str,
|
|
request_text: str,
|
|
response_text: str,
|
|
source: str = "",
|
|
timestamp: str | None = None,
|
|
request_timestamp: str | None = None,
|
|
response_timestamp: str | None = None,
|
|
) -> ConversationArtifact:
|
|
requester_slug = normalize_speaker(requester)
|
|
responder_slug = normalize_speaker(responder)
|
|
ts = timestamp or datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
metadata = {
|
|
"artifact_type": "alexander_request_response",
|
|
"requester": requester_slug,
|
|
"responder": responder_slug,
|
|
"speaker_tags": [f"speaker:{requester_slug}", f"speaker:{responder_slug}"],
|
|
"source": source,
|
|
"timestamp": ts,
|
|
}
|
|
if request_timestamp:
|
|
metadata["request_timestamp"] = request_timestamp
|
|
if response_timestamp:
|
|
metadata["response_timestamp"] = response_timestamp
|
|
return ConversationArtifact(
|
|
requester=requester_slug,
|
|
responder=responder_slug,
|
|
request_text=request_text,
|
|
response_text=response_text,
|
|
timestamp=ts,
|
|
metadata=metadata,
|
|
)
|
|
|
|
|
|
def extract_alexander_request_pairs(
|
|
turns: Iterable[dict],
|
|
*,
|
|
responder: str,
|
|
source: str = "",
|
|
) -> list[ConversationArtifact]:
|
|
responder_slug = normalize_speaker(responder)
|
|
pending_request: dict | None = None
|
|
artifacts: list[ConversationArtifact] = []
|
|
|
|
for turn in turns:
|
|
speaker = normalize_speaker(
|
|
turn.get("speaker") or turn.get("username") or turn.get("author") or turn.get("name")
|
|
)
|
|
text = (turn.get("text") or turn.get("content") or "").strip()
|
|
if not text:
|
|
continue
|
|
|
|
if speaker == "alexander":
|
|
pending_request = turn
|
|
continue
|
|
|
|
if speaker == responder_slug and pending_request is not None:
|
|
artifacts.append(
|
|
build_request_response_artifact(
|
|
requester="alexander",
|
|
responder=responder_slug,
|
|
request_text=(pending_request.get("text") or pending_request.get("content") or "").strip(),
|
|
response_text=text,
|
|
source=source,
|
|
request_timestamp=pending_request.get("timestamp"),
|
|
response_timestamp=turn.get("timestamp"),
|
|
timestamp=turn.get("timestamp") or pending_request.get("timestamp"),
|
|
)
|
|
)
|
|
pending_request = None
|
|
|
|
return artifacts
|