"""Provenance tracking for translation audit trails. Captures comprehensive input information for every translation, enabling full reproducibility and audit capability. """ from __future__ import annotations import hashlib from dataclasses import dataclass, field from datetime import datetime from typing import Any def compute_hash(text: str | None) -> str: """Compute SHA256 hash of text. Args: text: Text to hash, or None. Returns: Hex-encoded SHA256 hash, or empty string if text is None. """ if not text: return "" return hashlib.sha256(text.encode("utf-8")).hexdigest() @dataclass class TranslationProvenance: """Complete provenance record for a single translation. Tracks all inputs that contributed to a translation, enabling: - Full audit trails - Reproducibility verification - Change detection (via content hashes) - Debugging and quality investigation Attributes: source_texts: Source texts used (keyed by type: greek, hebrew, kjv, etc.). source_hashes: SHA256 hashes of source texts for change detection. preceding_verses: List of verse refs used as preceding context. following_verses: List of verse refs used as following context. pericope_id: ID of the pericope this verse belongs to, if any. style_config_hash: Hash of the style configuration used. glossary_hash: Hash of glossary terms used. prompt_version: Version identifier for the prompt template. model: Model used for translation. system_prompt: Complete system prompt sent to Claude. user_prompt: Complete user prompt sent to Claude. tm_matches_used: Translation memory matches used. attempt_number: Which attempt this was (1 = first try). started_at: When translation started. completed_at: When translation completed. evaluation_score: Quality score from evaluation (if evaluated). evaluation_flags: Quality flags from evaluation. chapter_summary: Chapter summary used as context. book_metadata: Book metadata used as context. """ # Source inputs with hashes for change detection source_texts: dict[str, str] = field(default_factory=dict) source_hashes: dict[str, str] = field(default_factory=dict) # Context inputs preceding_verses: list[str] = field(default_factory=list) following_verses: list[str] = field(default_factory=list) pericope_id: str | None = None # Configuration (frozen at translation time) style_config_hash: str = "" glossary_hash: str = "" prompt_version: str = "v1.0" model: str = "" # Full prompts (complete audit trail) system_prompt: str = "" user_prompt: str = "" # Translation memory inputs tm_matches_used: list[dict[str, Any]] = field(default_factory=list) # Execution details attempt_number: int = 1 started_at: datetime | None = None completed_at: datetime | None = None # Quality evaluation_score: float | None = None evaluation_flags: list[str] = field(default_factory=list) # Additional context chapter_summary: str | None = None book_metadata: dict[str, Any] = field(default_factory=dict) def compute_source_hashes(self) -> None: """Compute hashes for all source texts.""" self.source_hashes = { key: compute_hash(text) for key, text in self.source_texts.items() } def to_dict(self) -> dict[str, Any]: """Convert provenance to a dictionary for JSON serialization. Returns: Dictionary representation of provenance. """ return { "source_texts": self.source_texts, "source_hashes": self.source_hashes, "preceding_verses": self.preceding_verses, "following_verses": self.following_verses, "pericope_id": self.pericope_id, "style_config_hash": self.style_config_hash, "glossary_hash": self.glossary_hash, "prompt_version": self.prompt_version, "model": self.model, "system_prompt": self.system_prompt, "user_prompt": self.user_prompt, "tm_matches_used": self.tm_matches_used, "attempt_number": self.attempt_number, "started_at": self.started_at.isoformat() if self.started_at else None, "completed_at": self.completed_at.isoformat() if self.completed_at else None, "evaluation_score": self.evaluation_score, "evaluation_flags": self.evaluation_flags, "chapter_summary": self.chapter_summary, "book_metadata": self.book_metadata, } @classmethod def from_dict(cls, data: dict[str, Any]) -> TranslationProvenance: """Create provenance from a dictionary. Args: data: Dictionary representation. Returns: TranslationProvenance instance. """ return cls( source_texts=data.get("source_texts", {}), source_hashes=data.get("source_hashes", {}), preceding_verses=data.get("preceding_verses", []), following_verses=data.get("following_verses", []), pericope_id=data.get("pericope_id"), style_config_hash=data.get("style_config_hash", ""), glossary_hash=data.get("glossary_hash", ""), prompt_version=data.get("prompt_version", "v1.0"), model=data.get("model", ""), system_prompt=data.get("system_prompt", ""), user_prompt=data.get("user_prompt", ""), tm_matches_used=data.get("tm_matches_used", []), attempt_number=data.get("attempt_number", 1), started_at=( datetime.fromisoformat(data["started_at"]) if data.get("started_at") else None ), completed_at=( datetime.fromisoformat(data["completed_at"]) if data.get("completed_at") else None ), evaluation_score=data.get("evaluation_score"), evaluation_flags=data.get("evaluation_flags", []), chapter_summary=data.get("chapter_summary"), book_metadata=data.get("book_metadata", {}), ) def has_source_change(self, other: TranslationProvenance) -> bool: """Check if source texts have changed from another provenance. Args: other: Another provenance to compare against. Returns: True if any source hashes differ. """ for key, hash_value in self.source_hashes.items(): if other.source_hashes.get(key) != hash_value: return True return False def get_inputs_summary(self) -> dict[str, Any]: """Get a summary of inputs for logging/display. Returns: Dictionary with input summary. """ return { "sources": list(self.source_texts.keys()), "preceding_context": len(self.preceding_verses), "following_context": len(self.following_verses), "pericope": self.pericope_id is not None, "tm_matches": len(self.tm_matches_used), "model": self.model, "attempt": self.attempt_number, } def compute_glossary_hash(terms: list[Any]) -> str: """Compute hash of glossary terms used. Args: terms: List of glossary terms (any type with string representation). Returns: SHA256 hash of sorted term representations. """ if not terms: return "" # Sort by string representation for consistency term_strs = sorted(str(t) for t in terms) combined = "|".join(term_strs) return compute_hash(combined) def compute_style_hash(style_config: dict[str, Any]) -> str: """Compute hash of style configuration. Args: style_config: Style configuration dictionary. Returns: SHA256 hash of JSON-serialized config. """ if not style_config: return "" import json # Sort keys for consistency json_str = json.dumps(style_config, sort_keys=True) return compute_hash(json_str)